hunjing/backend/app/db/init_data.py

"""默认数据初始化。"""

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import settings
from app.core.ids import new_permission_id, new_role_id
from app.core.logging import log_with_context
from app.db.governance_seed import ensure_governance_seed
from app.models.rbac import Permission, Role, RolePermission, UserRole
from app.models.user import User
from app.services.auth import create_user
from app.services.characters import (
    ensure_all_system_templates,
    ensure_default_character_for_owner,
)

# ── 默认权限清单 ──────────────────────────────────────────
# resource:action 格式。前端 PermissionGuard 使用同一套字符串。
DEFAULT_PERMISSIONS: list[dict[str, str]] = [
    # 用户管理
    {"resource": "users", "action": "read", "description": "查看用户列表"},
    {"resource": "users", "action": "create", "description": "创建用户"},
    {"resource": "users", "action": "update", "description": "更新用户信息"},
    {"resource": "users", "action": "delete", "description": "删除用户"},
    # 工具管理
    {"resource": "tools", "action": "read", "description": "查看工具列表"},
    {"resource": "tools", "action": "admin", "description": "管理工具权限策略"},
    # 人物模板管理
    {"resource": "character_templates", "action": "read", "description": "查看系统人物模板"},
    {"resource": "character_templates", "action": "create", "description": "创建系统人物模板"},
    {"resource": "character_templates", "action": "update", "description": "更新系统人物模板"},
    {"resource": "character_templates", "action": "delete", "description": "删除系统人物模板"},
    # 安全审计
    {"resource": "audit", "action": "read", "description": "查看审计日志"},
    # 司命台治理
    {"resource": "goals", "action": "read", "description": "查看目标"},
    {"resource": "goals", "action": "update", "description": "修改目标"},
    {"resource": "perception", "action": "read", "description": "查看感知通道"},
    {"resource": "perception", "action": "update", "description": "修改感知授权"},
    {"resource": "world", "action": "read", "description": "查看世界实体"},
    {"resource": "world", "action": "update", "description": "修改世界实体"},
    {"resource": "plans", "action": "read", "description": "查看计划"},
    {"resource": "plans", "action": "update", "description": "修改计划"},
    {"resource": "traces", "action": "read", "description": "查看链路追踪"},
    {"resource": "tenants", "action": "read", "description": "查看多租户概览"},
    {"resource": "agents", "action": "read", "description": "查看天工阁"},
    {"resource": "agents", "action": "update", "description": "管理天工阁角色"},
    # 系统管理
    {"resource": "system", "action": "admin", "description": "系统配置管理"},
]


async def seed_default_data(session: AsyncSession) -> None:
    # 1) 确保系统模板
    templates = await ensure_all_system_templates(session)
    log_with_context("已确保系统模板", count=len(templates))

    # 2) 创建默认管理员用户（仅首次启动）
    result = await session.execute(select(User).limit(1))
    admin_user = result.scalar_one_or_none()
    if admin_user is None:
        admin_username = settings.default_admin_username or "admin"
        admin_email = settings.default_admin_email or "admin@hunjing.local"
        admin_password = settings.default_admin_password or "admin123"
        admin_user = await create_user(
            session,
            username=admin_username,
            email=admin_email,
            password=admin_password,
            is_admin=True,
        )
        await session.flush()
        log_with_context(
            "已创建默认管理员用户",
            username=admin_username,
            user_id=admin_user.id,
        )

    # 3) RBAC 种子数据
    await _seed_rbac(session, admin_user)

    # 4) 使用 admin_user 的 owner_scope 创建默认人物
    owner_scope = admin_user.id
    character = await ensure_default_character_for_owner(session, owner_scope)
    log_with_context(
        "已确保默认用户人物",
        character_id=character.id,
        source_template_id=character.source_template_id,
        owner_scope=owner_scope,
    )

    # 4b) 兼容测试环境：确保 default_owner_scope 也有激活人物
    #       repo/service 测试通过 get_active_character_model(session) 查找人物，
    #       其默认 owner_scope 来自 settings.default_owner_scope，与 admin_user.id 不一致。
    if settings.default_owner_scope and settings.default_owner_scope != owner_scope:
        await ensure_default_character_for_owner(session, settings.default_owner_scope)
    await session.commit()

    await ensure_governance_seed(session)
    log_with_context("已确保司命台治理种子")


async def _seed_rbac(session: AsyncSession, admin_user: User) -> None:
    """创建默认角色、权限，并为 admin 用户分配管理员角色。"""

    # 检查是否已初始化
    existing = await session.execute(select(Role).limit(1))
    if existing.scalar_one_or_none() is not None:
        return

    # 创建所有权限
    perm_records: dict[str, Permission] = {}
    for p in DEFAULT_PERMISSIONS:
        perm = Permission(
            id=new_permission_id(),
            resource=p["resource"],
            action=p["action"],
            description=p["description"],
        )
        session.add(perm)
        perm_records[perm.code] = perm

    # 创建角色
    admin_role = Role(
        id=new_role_id(),
        name="admin",
        description="管理员 — 拥有所有权限",
        is_system=True,
    )
    user_role = Role(
        id=new_role_id(),
        name="user",
        description="普通用户 — 基础操作权限",
        is_system=True,
    )
    session.add(admin_role)
    session.add(user_role)

    # admin 角色：所有权限
    for perm in perm_records.values():
        session.add(RolePermission(role_id=admin_role.id, permission_id=perm.id))

    # user 角色：仅基础读取权限
    user_perms = [p for p in DEFAULT_PERMISSIONS if p["action"] in ("read",)]
    for p in user_perms:
        perm = perm_records.get(f"{p['resource']}:{p['action']}")
        if perm:
            session.add(RolePermission(role_id=user_role.id, permission_id=perm.id))

    # 分配 admin 角色给当前管理员用户
    session.add(UserRole(user_id=admin_user.id, role_id=admin_role.id))

    await session.flush()
    log_with_context(
        "RBAC 种子数据已初始化",
        roles=2,
        permissions=len(perm_records),
        admin_user=admin_user.username,
    )