code/ai_finance_agent/backend/app/application/seed.py

"""幂等种子数据只补缺失项，重启不清空业务或恢复管理员已经修改的配置。"""
from sqlalchemy import select
from app.domain.models import AgentConfig, AuditRule, ApprovalChain, Department, Role, User
from app.infrastructure.security import hash_password, validate_password
from app.shared.config import settings

RULE_DEFAULTS = {
    "invoice_date_valid": ("发票日期", "error", {"max_age_days": 365}),
    "invoice_amount_valid": ("金额与合计", "error", {"large_amount": "100000.00"}),
    "invoice_header_valid": ("买卖方抬头", "error", {"company_name": ""}),
    "invoice_no_duplicate": ("同内容重复", "critical", {}),
    "tax_calculation_valid": ("税额计算", "warning", {"tolerance": "0.01"}),
    "tax_number_valid": ("税号长度", "warning", {"lengths": [15, 18, 20]}),
    "policy_compliance": ("政策合规", "error", {"required": True}),
}
AGENT_DEFAULTS = {
    "extract": ("发票提取 Agent", {"max_attempts": 3, "timeout": 20, "concurrency": 3}),
    "rule_check": ("规则校验 Agent", {}),
    "risk_detect": ("双引擎风控 Agent", {"timeout": 20, "window_days": 30, "amount_high": "50000.00", "amount_medium": "20000.00", "frequency_limit": 10, "dept_ratio_limit": 3}),
    "make_decision": ("确定性路由 Agent", {}),
    "feedback": ("人工反馈 Agent", {}),
}


async def latest_configs(session, model, key):
    """历史版本保留，仅取每个标识的最新配置作为下一轮输入。"""
    rows = (await session.scalars(select(model).order_by(model.version))).all()
    result = {}
    for row in rows:
        result[getattr(row, key)] = row
    return result


async def seed(session):
    """初始化角色和规则是必需步骤，演示账号单独受 SEED_DEMO 控制。"""
    for code, name in [("employee", "员工"), ("supervisor", "主管"), ("auditor", "审核员"), ("manager", "经理"), ("admin", "管理员")]:
        if not await session.scalar(select(Role).where(Role.code == code)):
            session.add(Role(code=code, display_name=name))
    for code, (name, severity, params) in RULE_DEFAULTS.items():
        if not await session.scalar(select(AuditRule).where(AuditRule.code == code)):
            session.add(AuditRule(code=code, name=name, severity=severity, params_json=params, scope_json={}))
    for node, (name, config) in AGENT_DEFAULTS.items():
        if not await session.scalar(select(AgentConfig).where(AgentConfig.node_name == node)):
            session.add(AgentConfig(node_name=node, name=name, config_json=config))
    await session.flush()
    if not settings.seed_demo:
        return
    validate_password(settings.demo_password)
    department = await session.scalar(select(Department).where(Department.name == "演示研发部"))
    if not department:
        department = Department(name="演示研发部")
        session.add(department)
        await session.flush()
    users = {}
    for username, name, role_code in [("alice", "演示员工", "employee"), ("bob", "演示主管", "supervisor"), ("carol", "演示财务", "auditor"), ("dave", "演示经理", "manager"), ("erin", "演示管理员", "admin")]:
        user = await session.scalar(select(User).where(User.username == username))
        if not user:
            role = await session.scalar(select(Role).where(Role.code == role_code))
            user = User(username=username, real_name=name, password_hash=hash_password(settings.demo_password),
                        department_id=department.id, role_id=role.id)
            session.add(user)
            await session.flush()
        users[username] = user
    # 默认链提供人工闭环；用户自己创建的链不会被启动脚本重置。
    for route, node_user, role in [("to_supervisor", "bob", "supervisor"), ("to_finance", "carol", "auditor"), ("to_manual", "carol", "auditor")]:
        if not await session.scalar(select(ApprovalChain).where(ApprovalChain.route == route)):
            session.add(ApprovalChain(name="演示审批链 · " + route, route=route,
                                      nodes_json=[{"approver_id": users[node_user].id, "role": role}]))