feat: add device AI role switching

talkingq-url/banban/routers/__init__.py

@@ -7,6 +7,7 @@ from banban.routers.device_location import router as device_location_router
 from banban.routers.devices import router as devices_router
 from banban.routers.im import router as im_router
 from banban.routers.parents import router as parents_router
+from banban.routers.roles import router as roles_router
 from banban.routers.wechat_auth import router as wechat_auth_router
 from banban.routers.mqtt_router import router as mqtt_router
 
@@ -22,7 +23,8 @@ banban_router.include_router(device_im_router, tags=["banban-device-im"])
 banban_router.include_router(device_location_router, tags=["banban-device-location"])
 banban_router.include_router(im_router, tags=["banban-im"])
 banban_router.include_router(parents_router, tags=["banban-parents"])
+banban_router.include_router(roles_router, tags=["banban-roles"])
 banban_router.include_router(mqtt_router, tags=["banban-mqtt"])
 
 
-__all__ = ["banban_router"]
+__all__ = ["banban_router"]

talkingq-url/banban/routers/roles.py

@@ -0,0 +1,134 @@
+from collections.abc import Mapping
+from typing import List
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+
+from banban.security import get_current_user_id
+from banban.service.device import device_service
+from handlers.prompt_sound_handler import send_welcome_sound
+from services.conversation_history import DeviceConversationHistory, conversation_history_manager
+from services.device_config import DeviceConfig, device_config_manager
+from services.role_manager import role_manager
+
+
+router = APIRouter(prefix="/roles", tags=["banban-roles"])
+
+
+class RoleSummaryResponse(BaseModel):
+    role_key: str
+    name: str
+    description: str | None = None
+    default_language: str | None = None
+    languages: List[str] = []
+
+
+class DeviceRoleResponse(BaseModel):
+    device_id: str
+    role_key: str
+    name: str
+    description: str | None = None
+    preferred_language: str | None = None
+    languages: List[str] = []
+
+
+class DeviceRoleUpdateRequest(BaseModel):
+    role_key: str
+    language: str | None = None
+    play_welcome: bool = False
+
+
+def _role_languages(role: Mapping) -> list[str]:
+    multilingual = role.get("multilingual")
+    if isinstance(multilingual, dict):
+        return list(multilingual.keys())
+    default_language = role.get("default_language")
+    return [str(default_language)] if default_language else []
+
+
+def _role_to_summary(role: Mapping) -> RoleSummaryResponse:
+    return RoleSummaryResponse(
+        role_key=str(role.get("role_key") or ""),
+        name=str(role.get("name") or role.get("role_key") or ""),
+        description=role.get("description"),
+        default_language=role.get("default_language"),
+        languages=_role_languages(role),
+    )
+
+
+@router.get("", response_model=list[RoleSummaryResponse])
+async def list_roles(
+    current_user_id: int = Depends(get_current_user_id),
+) -> list[RoleSummaryResponse]:
+    del current_user_id
+    roles = await role_manager.get_all_roles()
+    return [
+        _role_to_summary(role)
+        for role in sorted(roles.values(), key=lambda item: str(item.get("role_key") or ""))
+    ]
+
+
+@router.get("/devices/{device_id}", response_model=DeviceRoleResponse)
+async def get_device_role(
+    device_id: str,
+    current_user_id: int = Depends(get_current_user_id),
+) -> DeviceRoleResponse:
+    await device_service.ensure_device_access(device_id=device_id, user_id=current_user_id)
+    config = await device_config_manager.get_config(device_id, force_refresh=True)
+    role = await role_manager.get_role(config.selected_role_key)
+    if not role:
+        raise HTTPException(status_code=404, detail="role not found")
+
+    return DeviceRoleResponse(
+        device_id=device_id,
+        role_key=config.selected_role_key,
+        name=str(role.get("name") or config.selected_role_key),
+        description=role.get("description"),
+        preferred_language=config.preferred_language,
+        languages=_role_languages(role),
+    )
+
+
+@router.put("/devices/{device_id}", response_model=DeviceRoleResponse)
+async def update_device_role(
+    device_id: str,
+    payload: DeviceRoleUpdateRequest,
+    current_user_id: int = Depends(get_current_user_id),
+) -> DeviceRoleResponse:
+    await device_service.ensure_device_access(device_id=device_id, user_id=current_user_id)
+
+    role_key = payload.role_key.strip()
+    if not role_key:
+        raise HTTPException(status_code=422, detail="role_key is required")
+
+    role = await role_manager.get_role(role_key)
+    if not role:
+        raise HTTPException(status_code=404, detail="role not found")
+
+    current_config = await device_config_manager.get_config(device_id, force_refresh=True)
+    language = payload.language or current_config.preferred_language or role.get("default_language")
+    available_languages = _role_languages(role)
+    if language and available_languages and language not in available_languages:
+        raise HTTPException(status_code=422, detail="language is not supported by role")
+
+    await device_config_manager.set_config(
+        device_id,
+        DeviceConfig(selected_role_key=role_key, preferred_language=language),
+    )
+    await conversation_history_manager.set_history(
+        device_id,
+        DeviceConversationHistory(),
+        role_key,
+    )
+
+    if payload.play_welcome:
+        await send_welcome_sound(device_id, role_key, language)
+
+    return DeviceRoleResponse(
+        device_id=device_id,
+        role_key=role_key,
+        name=str(role.get("name") or role_key),
+        description=role.get("description"),
+        preferred_language=language,
+        languages=available_languages,
+    )

talkingq-url/banban_main.py

@@ -12,6 +12,7 @@ from banban.routers.device_im import router as device_im_router
 from banban.routers.devices import router as devices_router
 from banban.routers.im import router as im_router
 from banban.routers.parents import router as parents_router
+from banban.routers.roles import router as roles_router
 from banban.routers.wechat_auth import router as wechat_auth_router
 from database.connection import get_db_manager
 from initialization import init_directories
@@ -26,6 +27,7 @@ banban_router.include_router(devices_router, tags=["banban-devices"])
 banban_router.include_router(device_im_router, tags=["banban-device-im"])
 banban_router.include_router(im_router, tags=["banban-im"])
 banban_router.include_router(parents_router, tags=["banban-parents"])
+banban_router.include_router(roles_router, tags=["banban-roles"])
 
 
 @asynccontextmanager