hunjing/backend/app/domains/xingyi/api.py

"""形仪 API：真形锚点 / 形仪候选定相 / 声音档案与克隆授权 / ASR 转写。"""

from datetime import datetime

from fastapi import APIRouter, Depends, File, UploadFile, status
from pydantic import BaseModel, Field

from app.api.v1.deps import get_current_user
from app.domains.xingyi import service
from app.schemas.auth import CurrentUser

form_router = APIRouter(prefix="/characters/{character_id}", tags=["xingyi"])


class AnchorUpsertRequest(BaseModel):
    features: dict = Field(default_factory=dict, description="稳定辨识特征")
    age_presentation: str = "adult"
    permissions: dict = Field(default_factory=dict)
    mutable_scope: dict = Field(default_factory=dict)


class FormCreateRequest(BaseModel):
    kind: str = "costume"
    source: str = "upload"
    description: str = Field(min_length=1, max_length=4000)
    media_ref: str | None = None
    live2d: dict | None = None


class VoiceCreateRequest(BaseModel):
    source: str
    name: str = Field(min_length=1, max_length=120)
    params: dict = Field(default_factory=dict)
    grant_purpose: str | None = None
    grant_expires_at: datetime | None = None
    grant_distribution: str | None = None


# ── 真形锚点 ──


@form_router.get("/form-anchor")
async def get_anchor(
    character_id: str, user: CurrentUser = Depends(get_current_user)
) -> dict:
    return await service.get_anchor(character_id, user.owner_scope)


@form_router.put("/form-anchor")
async def upsert_anchor(
    character_id: str, payload: AnchorUpsertRequest, user: CurrentUser = Depends(get_current_user)
) -> dict:
    return await service.upsert_anchor(
        character_id, user.owner_scope,
        features=payload.features, age_presentation=payload.age_presentation,
        permissions=payload.permissions, mutable_scope=payload.mutable_scope,
    )


# ── 形仪候选与定相 ──


@form_router.get("/forms", status_code=status.HTTP_200_OK)
async def list_forms_api(
    character_id: str, user: CurrentUser = Depends(get_current_user), status_filter: str | None = None
) -> list[dict]:
    return await service.list_forms(character_id, user.owner_scope, status=status_filter)


@form_router.post("/forms", status_code=status.HTTP_201_CREATED)
async def create_form_api(
    character_id: str, payload: FormCreateRequest, user: CurrentUser = Depends(get_current_user)
) -> dict:
    return await service.create_form(
        character_id, user.owner_scope,
        kind=payload.kind, source=payload.source, description=payload.description,
        media_ref=payload.media_ref, live2d=payload.live2d,
    )


@form_router.get("/forms/primary")
async def current_primary_api(
    character_id: str, user: CurrentUser = Depends(get_current_user)
) -> dict | None:
    return await service.current_primary(character_id, user.owner_scope)


@form_router.post("/forms/{form_id}/adopt")
async def adopt_form_api(
    character_id: str, form_id: str, user: CurrentUser = Depends(get_current_user)
) -> dict:
    return await service.adopt_form(form_id, user.owner_scope)


@form_router.post("/forms/{form_id}/discard")
async def discard_form_api(
    character_id: str, form_id: str, user: CurrentUser = Depends(get_current_user)
) -> dict:
    return await service.discard_form(form_id, user.owner_scope)


# ── 声音档案 ──


@form_router.get("/voices")
async def list_voices_api(
    character_id: str, user: CurrentUser = Depends(get_current_user)
) -> list[dict]:
    return await service.list_voices(character_id, user.owner_scope)


@form_router.post("/voices", status_code=status.HTTP_201_CREATED)
async def create_voice_api(
    character_id: str, payload: VoiceCreateRequest, user: CurrentUser = Depends(get_current_user)
) -> dict:
    return await service.create_voice(
        character_id, user.owner_scope,
        source=payload.source, name=payload.name, params=payload.params,
        grant_purpose=payload.grant_purpose, grant_expires_at=payload.grant_expires_at,
        grant_distribution=payload.grant_distribution,
    )


@form_router.post("/voices/{voice_id}/revoke-grant")
async def revoke_voice_grant_api(
    character_id: str, voice_id: str, user: CurrentUser = Depends(get_current_user)
) -> dict:
    return await service.revoke_voice_grant(voice_id, user.owner_scope)


@form_router.post("/voices/{voice_id}/assert-usable")
async def assert_voice_usable_api(
    character_id: str, voice_id: str, user: CurrentUser = Depends(get_current_user)
) -> dict:
    """合成前置校验（授权克隆：未撤回且未过期才可用）。"""
    return await service.assert_voice_usable(voice_id, user.owner_scope)


# ── ASR 转写（元炁 /v1/audio/transcriptions 透传） ──


@form_router.post("/voice/transcriptions")
async def transcribe_audio_api(
    character_id: str,
    user: CurrentUser = Depends(get_current_user),
    audio: UploadFile = File(...),
) -> dict:
    """语音转文本（元炁 SenseVoice）。转写文本由前端作为镜树问节内容提交。

    多模态关闭（YUANQI 未启用）时本端点 503——文字相伴链路不受影响（回归由
    测试钉死：关闭全部多模态后镜树/缘契照常）。
    """
    from app.domains.xingyi.asr import transcribe

    await service._owned(character_id, user.owner_scope)
    content = await audio.read()
    return await transcribe(content, audio.filename or "audio.webm")