hunjing/backend/app/services/admin_characters.py

"""司命台 · 人物模板管理聚合服务（api-spec §19 扩展）。

系统模板 CRUD，所有写操作记审计并入起居注。
路由直接调用此服务层函数（函数内部自管 ``AsyncSessionLocal``）。
"""

from app.core.config import settings
from app.core.errors import AppError
from app.core.ids import new_character_id
from app.core.time import utc_now
from app.db.session import AsyncSessionLocal
from app.models.character import Character
from app.repositories import characters as character_repo
from app.repositories.json_utils import dumps, loads_dict
from app.schemas.admin_character import (
    AdminCreateTemplateRequest,
    AdminTemplateSummaryResponse,
    AdminUpdateTemplateRequest,
)
from app.services.audit import record_audit_event


def _template_to_response(template: Character) -> AdminTemplateSummaryResponse:
    profile = loads_dict(template.profile_json)
    model_policy = loads_dict(template.model_policy_json)
    # 从 profile 中推断 is_default（无独立字段，保留为 False 未映射）
    return AdminTemplateSummaryResponse(
        key=template.template_key or "",
        name=template.name,
        display_name=template.display_name,
        language=template.language,
        privacy_level=template.privacy_level,
        is_default=False,
        profile=profile,
        model_policy=model_policy,
        created_at=template.created_at.isoformat() if template.created_at else "",
        updated_at=template.updated_at.isoformat() if template.updated_at else "",
    )


async def admin_list_templates() -> list[AdminTemplateSummaryResponse]:
    """列出所有系统模板（管理端完整视图）。"""
    async with AsyncSessionLocal() as session:
        templates = await character_repo.list_system_templates(session)
        return [_template_to_response(t) for t in templates]


async def admin_get_template(key: str) -> AdminTemplateSummaryResponse:
    """获取单个系统模板。"""
    async with AsyncSessionLocal() as session:
        template = await character_repo.get_system_template_by_key(session, key)
        if template is None:
            raise AppError("not_found", f"模板 '{key}' 不存在。")
        return _template_to_response(template)


async def admin_create_template(
    payload: AdminCreateTemplateRequest,
) -> AdminTemplateSummaryResponse:
    """创建系统模板。"""
    async with AsyncSessionLocal() as session:
        # 检查是否已存在同 key 模板
        existing = await character_repo.get_system_template_by_key(session, payload.key)
        if existing is not None:
            raise AppError("conflict", f"模板 key '{payload.key}' 已存在。")

        profile = dict(payload.profile)
        model_policy = dict(payload.model_policy)
        model_policy.setdefault("default_chat_model_hint", settings.default_model_hint)

        character = Character(
            id=new_character_id(),
            name=payload.name,
            display_name=payload.display_name,
            status="active",
            language=payload.language,
            profile_json=dumps(profile),
            model_policy_json=dumps(model_policy),
            privacy_level=payload.privacy_level,
            owner_scope="system",
            is_system_template=True,
            source_template_id=None,
            template_key=payload.key,
        )
        session.add(character)
        await session.flush()

        # birthday 兜底
        await _ensure_birthday(character)

        await record_audit_event(
            session,
            action="admin.character_template.created",
            character_id=character.id,
            actor="admin",
            decision="auto",
            input_summary=f"管理员创建系统模板 '{payload.key}'",
            resource_type="character_template",
            resource_id=character.id,
        )

        await session.commit()
        await session.refresh(character)
        return _template_to_response(character)


async def admin_update_template(
    key: str,
    payload: AdminUpdateTemplateRequest,
) -> AdminTemplateSummaryResponse:
    """更新系统模板（部分更新）。"""
    async with AsyncSessionLocal() as session:
        template = await character_repo.get_system_template_by_key(session, key)
        if template is None:
            raise AppError("not_found", f"模板 '{key}' 不存在。")

        changed = False
        update_fields = payload.model_dump(exclude_unset=True)

        if "name" in update_fields and template.name != update_fields["name"]:
            template.name = update_fields["name"]
            changed = True
        if "display_name" in update_fields and template.display_name != update_fields["display_name"]:
            template.display_name = update_fields["display_name"]
            changed = True
        if "language" in update_fields and template.language != update_fields["language"]:
            template.language = update_fields["language"]
            changed = True
        if "privacy_level" in update_fields and template.privacy_level != update_fields["privacy_level"]:
            template.privacy_level = update_fields["privacy_level"]
            changed = True

        if "profile" in update_fields:
            current_profile = loads_dict(template.profile_json)
            current_profile.update(update_fields["profile"])
            template.profile_json = dumps(current_profile)
            changed = True

        if "model_policy" in update_fields:
            current_policy = loads_dict(template.model_policy_json)
            current_policy.update(update_fields["model_policy"])
            template.model_policy_json = dumps(current_policy)
            changed = True

        if changed:
            await record_audit_event(
                session,
                action="admin.character_template.updated",
                character_id=template.id,
                actor="admin",
                decision="auto",
                input_summary=f"管理员更新系统模板 '{key}'",
                resource_type="character_template",
                resource_id=template.id,
            )
            template.updated_at = utc_now()
            await session.commit()
            await session.refresh(template)

        return _template_to_response(template)


async def admin_delete_template(key: str) -> None:
    """删除系统模板。"""
    async with AsyncSessionLocal() as session:
        template = await character_repo.get_system_template_by_key(session, key)
        if template is None:
            raise AppError("not_found", f"模板 '{key}' 不存在。")

        # 检查是否有关联的子人物
        await _check_template_in_use(session, template)

        await record_audit_event(
            session,
            action="admin.character_template.deleted",
            character_id=template.id,
            actor="admin",
            decision="auto",
            input_summary=f"管理员删除系统模板 '{key}'",
            resource_type="character_template",
            resource_id=template.id,
        )

        await session.delete(template)
        await session.commit()


async def _check_template_in_use(session, template: Character) -> None:
    """检查模板是否被用户人物引用，是则阻止删除。"""
    from sqlalchemy import func, select

    count = await session.scalar(
        select(func.count())
        .select_from(Character)
        .where(
            Character.source_template_id == template.id,
            Character.is_system_template.is_(False),
        )
    )
    if count and count > 0:
        raise AppError(
            "conflict",
            f"模板 '{template.template_key}' 已被 {count} 个人物引用，无法删除。"
            "请先确保无用户人物引用此模板后再试。",
        )


async def _ensure_birthday(character: Character) -> None:
    """为模板创建兜底 birthday（profile 中无 birthday 时用 created_at）。"""
    profile = loads_dict(character.profile_json)
    birthday = profile.get("birthday")
    if isinstance(birthday, str) and birthday.strip():
        return
    birthday_iso = character.created_at.strftime("%Y-%m-%d")
    profile["birthday"] = birthday_iso
    character.profile_json = dumps(profile)