"""角色域路由（WP-02）。""" from typing import Annotated, Any from fastapi import APIRouter, Depends, HTTPException, Request, status from pydantic import BaseModel, ConfigDict, Field from sqlalchemy import select from app.auth.dependencies import get_current_user, require_state_change from app.characters import service from app.conversations.routes import _domain_of # 复用域解析 from app.db.session import DbSession from app.models import Character, CharacterVersion, User router = APIRouter(prefix="/v1/characters", tags=["characters"]) class DriveIn(BaseModel): model_config = ConfigDict(extra="forbid") name: str description: str intensity: int = Field(default=1, ge=1, le=5) class DrivesUpdateIn(BaseModel): model_config = ConfigDict(extra="forbid") drives: list[DriveIn] = Field(max_length=16) def _character_out(c: Character) -> dict[str, Any]: return { "id": c.id, "name": c.name, "status": c.status, "origin": c.origin, "based_on_real_person": c.based_on_real_person, "identity_baseline": c.identity_baseline, "drives": c.drives, "current_version_id": c.current_version_id, "created_at": c.created_at.isoformat(), } def _version_out(v: CharacterVersion) -> dict[str, Any]: return { "id": v.id, "character_id": v.character_id, "kind": v.kind, "status": v.status, "payload": v.payload, "created_at": v.created_at.isoformat(), } @router.get("") async def list_characters( db: DbSession, user: Annotated[User, Depends(get_current_user)], limit: int = 50, cursor: str | None = None, ) -> dict[str, Any]: if not 1 <= limit <= 200: raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "limit 取值范围 1-200") domain = await _domain_of(db, user) stmt = select(Character).where(Character.personal_domain_id == domain.id) stmt = stmt.order_by(Character.created_at.asc(), Character.id.asc()).limit(limit + 1) rows = list((await db.scalars(stmt)).all()) has_more = len(rows) > limit return { "data": [_character_out(c) for c in rows[:limit]], "next_cursor": rows[limit].id if has_more else None, "has_more": has_more, } @router.post("", status_code=201) async def create_character( db: DbSession, user: Annotated[User, Depends(require_state_change)], request: Request, body: dict[str, Any], ) -> dict[str, Any]: domain = await _domain_of(db, user) character = await service.create_character( db, domain.id, body, getattr(request.state, "trace_id", None) ) return _character_out(character) @router.get("/{character_id}") async def get_character( character_id: str, db: DbSession, user: Annotated[User, Depends(get_current_user)], ) -> dict[str, Any]: domain = await _domain_of(db, user) character = await service.require_character(db, domain.id, character_id) return _character_out(character) @router.post("/{character_id}/activations", status_code=201) async def activate( character_id: str, db: DbSession, user: Annotated[User, Depends(require_state_change)], request: Request, ) -> dict[str, Any]: domain = await _domain_of(db, user) character = await service.activate_character( db, domain.id, character_id, getattr(request.state, "trace_id", None) ) return _character_out(character) @router.post("/{character_id}/first-statement", status_code=201) async def submit_first_statement( character_id: str, db: DbSession, user: Annotated[User, Depends(require_state_change)], request: Request, body: dict[str, str], ) -> dict[str, Any]: domain = await _domain_of(db, user) version = await service.submit_first_statement( db, domain.id, character_id, body.get("text", ""), getattr(request.state, "trace_id", None), ) return _version_out(version) @router.post("/{character_id}/first-statement/confirmations", status_code=201) async def confirm_first_statement( character_id: str, db: DbSession, user: Annotated[User, Depends(require_state_change)], request: Request, body: dict[str, str], ) -> dict[str, Any]: domain = await _domain_of(db, user) version = await service.confirm_first_statement( db, domain.id, character_id, body.get("version_id", ""), getattr(request.state, "trace_id", None), ) return _version_out(version) LIFECYCLE_ACTIONS = { "rests": "rest", "archives": "archive", "revivals": "revive", "death-requests": "request_death", "death-cancellations": "cancel_death", "death-confirmations": "confirm_death", } @router.post("/{character_id}/reforgings", status_code=201) async def reforge( character_id: str, db: DbSession, user: Annotated[User, Depends(require_state_change)], request: Request, body: dict[str, Any], ) -> dict[str, Any]: domain = await _domain_of(db, user) version = await service.reforge_character( db, domain.id, character_id, body, getattr(request.state, "trace_id", None) ) return _version_out(version) @router.patch("/{character_id}/drives") async def update_drives( character_id: str, db: DbSession, user: Annotated[User, Depends(require_state_change)], request: Request, body: DrivesUpdateIn, ) -> dict[str, Any]: domain = await _domain_of(db, user) await service.update_drives( db, domain.id, character_id, [d.model_dump() for d in body.drives], getattr(request.state, "trace_id", None), ) character = await service.require_character(db, domain.id, character_id) return _character_out(character) @router.post("/{character_id}/persona-candidates", status_code=201) async def submit_persona_candidate( character_id: str, db: DbSession, user: Annotated[User, Depends(require_state_change)], request: Request, body: dict[str, Any], ) -> dict[str, Any]: domain = await _domain_of(db, user) version = await service.submit_persona_candidate( db, domain.id, character_id, body, getattr(request.state, "trace_id", None) ) return _version_out(version) @router.post("/{character_id}/versions/{version_id}/decisions", status_code=201) async def review_version( character_id: str, version_id: str, db: DbSession, user: Annotated[User, Depends(require_state_change)], request: Request, body: dict[str, bool], ) -> dict[str, Any]: domain = await _domain_of(db, user) version = await service.review_version( db, domain.id, character_id, version_id, bool(body.get("approve")), getattr(request.state, "trace_id", None), ) return _version_out(version) @router.post("/{character_id}/{action}", status_code=200) async def lifecycle( character_id: str, action: str, db: DbSession, user: Annotated[User, Depends(require_state_change)], request: Request, ) -> dict[str, Any]: if action not in LIFECYCLE_ACTIONS: raise HTTPException(status.HTTP_404_NOT_FOUND, "未知生命周期动作") domain = await _domain_of(db, user) character = await service.lifecycle_action( db, domain.id, character_id, LIFECYCLE_ACTIONS[action], getattr(request.state, "trace_id", None), ) return _character_out(character)