code/ai_finance_agent/backend/app/api/management.py

"""认证与配置 API：所有接口中文契约同时出现在自动生成的 OpenAPI 中。"""
from fastapi import APIRouter, Depends, Request, Query
from sqlalchemy import select
from app.api.dependencies import current_user, read_db, write_db
from app.domain.models import AgentConfig, AgentExecution, AuditRule, Department, Role, User
from app.application.seed import latest_configs
from app.application.services import get, log, require, serialize, validate_agent, validate_rule
from app.domain.rules import check_rule
from app.infrastructure.security import check_password, hash_password, issue_tokens, public_user, refresh_record, validate_password
from app.infrastructure.ai import ai, AIUnavailable
from app.shared.common import BusinessError, envelope, paginate, utcnow
from app.shared.config import settings
from app.shared import schemas as S

router = APIRouter()


def ok(request, data, message="操作成功"):
    return envelope(data, request.state.request_id, message)


@router.post("/auth/login", tags=["认证"])
async def login(body: S.Login, request: Request, session=Depends(write_db)):
    """功能：验证密码并创建会话。参数：登录名/密码。返回：双令牌与用户。异常：401凭据错误，422参数错误。"""
    user = await session.scalar(select(User).where(User.username == body.username))
    if not user or not user.active or not check_password(body.password, user.password_hash):
        raise BusinessError("用户名或密码错误", 401, "LOGIN_FAILED")
    actor = await public_user(session, user)
    log(session, actor, "auth.login", "user", user.id)
    return ok(request, await issue_tokens(session, user))


@router.post("/auth/refresh", tags=["认证"])
async def refresh(body: S.Refresh, request: Request, session=Depends(write_db)):
    """功能：刷新并撤销旧凭据。参数：refresh_token。返回：新双令牌。异常：401已撤销/过期/停用。"""
    record = await refresh_record(session, body.refresh_token)
    user = await get(session, User, record.user_id)
    if not user.active:
        raise BusinessError("账号已停用", 401, "USER_DISABLED")
    record.revoked_at = utcnow()
    return ok(request, await issue_tokens(session, user))


@router.post("/auth/logout", tags=["认证"])
async def logout(body: S.Refresh, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：撤销本人刷新会话。参数：refresh_token。返回：撤销状态。异常：401失效，403非本人。"""
    record = await refresh_record(session, body.refresh_token)
    if record.user_id != actor["id"]:
        raise BusinessError("只能退出本人会话", 403)
    record.revoked_at = utcnow()
    log(session, actor, "auth.logout", "user", actor["id"])
    return ok(request, {"revoked": True})


@router.get("/auth/me", tags=["认证"])
async def me(request: Request, actor=Depends(current_user)):
    """功能：返回当前身份。参数：Bearer令牌。返回：无敏感字段的用户。异常：401失效或未登录。"""
    return ok(request, actor)


@router.get("/users", tags=["用户"])
async def users(request: Request, page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100),
                actor=Depends(current_user), session=Depends(read_db)):
    """功能：经理查看用户用于审批配置。参数：分页。返回：无密码用户列表。异常：403权限不足。"""
    require(actor, "manager")
    rows = (await session.scalars(select(User).order_by(User.id))).all()
    return ok(request, paginate([await public_user(session, item) for item in rows], page, page_size))


@router.post("/users", status_code=201, tags=["用户"])
async def create_user(body: S.UserCreate, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：管理员创建有效用户。参数：用户/部门/角色/密码。返回：用户。异常：403权限，409重名，422弱密码。"""
    require(actor, "admin")
    validate_password(body.password)
    if await session.scalar(select(User).where(User.username == body.username)):
        raise BusinessError("登录名已存在", 409, "USERNAME_EXISTS")
    department = await get(session, Department, body.department_id)
    if not department.active:
        raise BusinessError("部门已停用")
    role = await session.scalar(select(Role).where(Role.code == body.role))
    user = User(username=body.username, real_name=body.real_name, password_hash=hash_password(body.password),
                department_id=department.id, role_id=role.id)
    session.add(user)
    await session.flush()
    data = await public_user(session, user)
    log(session, actor, "user.created", "user", user.id, after=data)
    return ok(request, data)


@router.patch("/users/{identifier}", tags=["用户"])
async def patch_user(identifier: int, body: S.UserPatch, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：修改身份/停用。参数：ID和允许字段。返回：更新用户。异常：403权限，404不存在，409停用自己。"""
    require(actor, "admin")
    user = await get(session, User, identifier)
    if user.id == actor["id"] and (body.active is False or body.role is not None and body.role != "admin"):
        raise BusinessError("不能停用或降级当前管理员，避免失去管理入口", 409)
    before = await public_user(session, user)
    if body.department_id is not None:
        department = await get(session, Department, body.department_id)
        if not department.active:
            raise BusinessError("部门已停用")
        user.department_id = body.department_id
    if body.role is not None:
        user.role_id = (await session.scalar(select(Role).where(Role.code == body.role))).id
    if body.real_name is not None:
        user.real_name = body.real_name
    if body.active is not None:
        user.active = body.active
    data = await public_user(session, user)
    log(session, actor, "user.updated", "user", identifier, before, data)
    return ok(request, data)


@router.get("/departments", tags=["部门"])
async def departments(request: Request, actor=Depends(current_user), session=Depends(read_db)):
    """功能：提供部门选项。参数：当前身份。返回：部门列表。异常：401未登录。"""
    return ok(request, [serialize(item) for item in (await session.scalars(select(Department).order_by(Department.id))).all()])


@router.post("/departments", status_code=201, tags=["部门"])
async def create_department(body: S.DepartmentBody, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：创建部门。参数：名称/启用。返回：部门。异常：403权限，409重名。"""
    require(actor, "admin")
    if await session.scalar(select(Department).where(Department.name == body.name)):
        raise BusinessError("部门名称已存在", 409)
    row = Department(name=body.name, active=body.active)
    session.add(row)
    await session.flush()
    log(session, actor, "department.created", "department", row.id, after=serialize(row))
    return ok(request, serialize(row))


@router.patch("/departments/{identifier}", tags=["部门"])
async def patch_department(identifier: int, body: S.DepartmentBody, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：更名或停用。参数：ID/名称/active。返回：部门。异常：403权限，404不存在，409重名。"""
    require(actor, "admin")
    row = await get(session, Department, identifier)
    if await session.scalar(select(Department).where(Department.name == body.name, Department.id != identifier)):
        raise BusinessError("部门名称已存在", 409)
    before = serialize(row)
    row.name, row.active = body.name, body.active
    log(session, actor, "department.updated", "department", row.id, before, serialize(row))
    return ok(request, serialize(row))


@router.get("/agents", tags=["Agent"])
async def agents(request: Request, actor=Depends(current_user), session=Depends(read_db)):
    """功能：查看节点配置和实际统计。参数：当前身份。返回：五节点列表。异常：403权限不足。"""
    require(actor, "auditor")
    rows = await latest_configs(session, AgentConfig, "node_name")
    executions = (await session.scalars(select(AgentExecution))).all()
    values = []
    for key, row in rows.items():
        items = [item for item in executions if item.node_name == key]
        values.append({**serialize(row), "executions": len(items), "degraded_count": sum(item.status == "degraded" for item in items),
                       "average_ms": round(sum(item.duration_ms for item in items) / len(items)) if items else None})
    return ok(request, values)


@router.get("/agents/{node_name}", tags=["Agent"])
async def agent_detail(node_name: str, request: Request, actor=Depends(current_user), session=Depends(read_db)):
    """功能：查看节点全部版本。参数：固定节点名。返回：当前配置/历史。异常：403权限，404未知节点。"""
    require(actor, "auditor")
    rows = (await session.scalars(select(AgentConfig).where(AgentConfig.node_name == node_name).order_by(AgentConfig.version.desc()))).all()
    if not rows:
        raise BusinessError("节点不存在", 404)
    return ok(request, {"current": serialize(rows[0]), "history": [serialize(item) for item in rows]})


@router.patch("/agents/{node_name}", tags=["Agent"])
async def patch_agent(node_name: str, body: S.ConfigPatch, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：创建不可变配置版本。参数：节点/预期版本/参数。返回：新配置。异常：403权限，409版本冲突，422非法配置。"""
    require(actor, "admin")
    old = (await latest_configs(session, AgentConfig, "node_name")).get(node_name)
    if not old:
        raise BusinessError("节点不存在", 404)
    if body.expected_version != old.version:
        raise BusinessError("配置已更新，请刷新", 409, "VERSION_CONFLICT")
    config = body.config if body.config is not None else old.config_json
    enabled = body.enabled if body.enabled is not None else old.enabled
    validate_agent(node_name, config, enabled)
    row = AgentConfig(node_name=node_name, name=old.name, version=old.version + 1, enabled=enabled, config_json=config)
    session.add(row)
    await session.flush()
    log(session, actor, "agent.updated", "agent", row.id, serialize(old), serialize(row))
    return ok(request, serialize(row))


@router.post("/agents/{node_name}/test", tags=["Agent"])
async def test_agent(node_name: str, request: Request, actor=Depends(current_user), session=Depends(read_db)):
    """功能：真实连接或确定性节点测试。参数：节点名。返回：可用/降级原因。异常：403权限，404未知节点。"""
    require(actor, "auditor")
    row = (await latest_configs(session, AgentConfig, "node_name")).get(node_name)
    if not row:
        raise BusinessError("节点不存在", 404)
    if node_name not in {"extract", "risk_detect"}:
        return ok(request, {"status": "success", "message": "本地确定性节点已加载"})
    base = (settings.vlm_base_url or settings.llm_base_url) if node_name == "extract" else settings.llm_base_url
    if not row.enabled or not base:
        return ok(request, {"status": "degraded", "message": "模型未配置或节点模型能力停用"})
    try:
        key = settings.ai_api_key
        response = await ai.http.get(base.rstrip("/") + "/models", headers={"Authorization": "Bearer " + key} if key else {}, timeout=row.config_json.get("timeout", 20))
        response.raise_for_status()
        models = [item["id"] for item in response.json().get("data", [])]
        expected = (settings.vlm_model or settings.llm_model) if node_name == "extract" else settings.llm_model
        return ok(request, {"status": "success" if expected in models else "degraded", "models": models,
                            "message": "模型列表已查询；该测试不代表实际识别效果"})
    except Exception:
        return ok(request, {"status": "degraded", "message": "真实模型服务连接失败"})


@router.get("/rules", tags=["规则"])
async def rules(request: Request, page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100),
                code: str | None = None, enabled: bool | None = None, actor=Depends(current_user), session=Depends(read_db)):
    """功能：查看生效规则。参数：分页/code/enabled。返回：最新版本列表。异常：403权限不足。"""
    require(actor, "auditor")
    rows = list((await latest_configs(session, AuditRule, "code")).values())
    rows = [item for item in rows if (code is None or item.code == code) and (enabled is None or item.enabled == enabled)]
    return ok(request, paginate([serialize(item) for item in rows], page, page_size))


@router.get("/rules/{code}", tags=["规则"])
async def rule_detail(code: str, request: Request, actor=Depends(current_user), session=Depends(read_db)):
    """功能：追踪规则历史。参数：规则code。返回：当前及所有版本。异常：403权限，404未知规则。"""
    require(actor, "auditor")
    rows = (await session.scalars(select(AuditRule).where(AuditRule.code == code).order_by(AuditRule.version.desc()))).all()
    if not rows:
        raise BusinessError("规则不存在", 404)
    return ok(request, {"current": serialize(rows[0]), "history": [serialize(item) for item in rows]})


@router.patch("/rules/{code}", tags=["规则"])
async def patch_rule(code: str, body: S.RulePatch, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：保存规则新版本。参数：预期版本/参数/范围/开关。返回：新版本。异常：403权限，409并发，422核心检查绕过。"""
    require(actor, "admin")
    old = (await latest_configs(session, AuditRule, "code")).get(code)
    if not old:
        raise BusinessError("规则不存在", 404)
    if body.expected_version != old.version:
        raise BusinessError("规则版本已更新", 409, "VERSION_CONFLICT")
    params = body.params if body.params is not None else old.params_json
    scope = body.scope if body.scope is not None else old.scope_json
    severity, enabled = body.severity or old.severity, old.enabled if body.enabled is None else body.enabled
    validate_rule(code, params, scope, severity, enabled)
    row = AuditRule(code=code, name=old.name, version=old.version + 1, enabled=enabled, severity=severity, params_json=params, scope_json=scope)
    session.add(row)
    await session.flush()
    log(session, actor, "rule.updated", "rule", row.id, serialize(old), serialize(row))
    return ok(request, serialize(row))


@router.post("/rules/{code}/test", tags=["规则"])
async def test_rule(code: str, body: S.RuleTest, request: Request, actor=Depends(current_user), session=Depends(read_db)):
    """功能：独立试算不改变正式结果。参数：invoice/context。返回：规则证据。异常：403权限，404规则，422上下文非法。"""
    require(actor, "auditor")
    row = (await latest_configs(session, AuditRule, "code")).get(code)
    if not row:
        raise BusinessError("规则不存在", 404)
    if code == "policy_compliance":
        from app.application.pipeline import policy_check
        from app.application.services import policy_snapshot
        value = await policy_check(serialize(row), body.invoice, {"reimbursement": {"type": body.context.get("type", 1)}, "context": body.context, "policies": await policy_snapshot(session)})
    else:
        try:
            value = check_rule(serialize(row), body.invoice, body.context)
        except (ValueError, TypeError, KeyError):
            raise BusinessError("试算字段或上下文不符合规则契约", 422, "RULE_TEST_INVALID")
    return ok(request, value)