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

"""审批严格关联轮次和指定处理人，管理员也不能代签或审核自己的申请。"""
from sqlalchemy import select, update
from app.domain.models import ApprovalChain, ApprovalRecord, ApprovalRound, AuditRun, Reimbursement, User, Role
from app.application.services import alert, get, log, serialize
from app.infrastructure.database import transaction
from app.shared.common import BusinessError, ROLES, transition, utcnow


async def validate_chain(session, body):
    """链配置在保存阶段就验证人员和角色，减少运行后才发现无人可审的挂单。"""
    from app.shared.common import cents
    minimum = cents(body.amount_min) if body.amount_min is not None else None
    maximum = cents(body.amount_max) if body.amount_max is not None else None
    if minimum is not None and maximum is not None and minimum > maximum:
        raise BusinessError("金额下限不能大于上限")
    seen, roles = set(), []
    for node in body.nodes:
        user = await get(session, User, node.approver_id)
        role = await get(session, Role, user.role_id)
        if not user.active or role.code != node.role or node.approver_id in seen:
            raise BusinessError("审批人必须有效、角色一致且不能重复")
        seen.add(node.approver_id)
        roles.append(role.code)
    if body.route in {"to_finance", "to_manual"} and "auditor" not in roles:
        raise BusinessError("财务和人工复核链必须包含审核员")
    if body.route == "to_supervisor" and not any(ROLES[role] >= ROLES["supervisor"] for role in roles):
        raise BusinessError("主管链必须包含有审批权限的人")
    if body.department_id:
        from app.domain.models import Department
        department = await get(session, Department, body.department_id)
        if not department.active:
            raise BusinessError("不能为停用部门配置新链")
    return minimum, maximum


async def start_approval(audit_id):
    """审核事件可能重复消费，唯一 audit_id 和状态护栏保证只启动一个轮次。"""
    async with transaction() as session:
        audit = await get(session, AuditRun, audit_id)
        reimbursement = await get(session, Reimbursement, audit.reimbursement_id)
        if await session.scalar(select(ApprovalRound).where(ApprovalRound.audit_id == audit_id)):
            return
        if reimbursement.revision != audit.revision or reimbursement.status != "pending_audit" or audit.status != "completed":
            return
        if audit.decision == "auto_approve":
            transition(reimbursement, "in_approval")
            round_item = ApprovalRound(reimbursement_id=reimbursement.id, revision=audit.revision, audit_id=audit.id,
                                       chain_snapshot={"system": True, "nodes": []}, status="approved")
            session.add(round_item)
            await session.flush()
            session.add(ApprovalRecord(round_id=round_item.id, node_index=0, approver_id=None, result="approve",
                                       opinion="低风险且全部必需检查通过，系统自动批准", decided_at=utcnow()))
            transition(reimbursement, "approved")
        else:
            candidates = (await session.scalars(select(ApprovalChain).where(ApprovalChain.active.is_(True), ApprovalChain.route == audit.decision))).all()
            candidates = [chain for chain in candidates if (chain.type is None or chain.type == reimbursement.type)
                and (chain.department_id is None or chain.department_id == reimbursement.department_id)
                and (chain.amount_min_cents is None or chain.amount_min_cents <= reimbursement.total_cents)
                and (chain.amount_max_cents is None or chain.amount_max_cents >= reimbursement.total_cents)]
            candidates.sort(key=lambda chain: (sum(x is not None for x in [chain.type, chain.department_id, chain.amount_min_cents, chain.amount_max_cents]), chain.priority, chain.version, chain.id), reverse=True)
            valid = []
            for chain in candidates:
                users = [await session.get(User, node["approver_id"]) for node in chain.nodes_json]
                roles = [await session.get(Role, user.role_id) if user else None for user in users]
                if all(user and user.active and user.id != reimbursement.user_id and role and role.code == node["role"]
                       for user, role, node in zip(users, roles, chain.nodes_json)):
                    valid.append(chain)
            if not valid:
                await alert(session, f"chain:{audit_id}", "chain_missing", "没有匹配的有效且非自审审批链，请经理配置审批链", "error", audit_id, reimbursement.id)
                # 先提交告警，再由外层异常触发 outbox 退避重试。
                missing = True
            else:
                missing = False
                chain = valid[0]
                transition(reimbursement, "in_approval")
                round_item = ApprovalRound(reimbursement_id=reimbursement.id, revision=audit.revision, audit_id=audit.id,
                                           chain_snapshot=serialize(chain))
                session.add(round_item)
                await session.flush()
                session.add(ApprovalRecord(round_id=round_item.id, node_index=0, approver_id=chain.nodes_json[0]["approver_id"]))
        if audit.decision == "auto_approve" or not missing:
            log(session, None, "approval.started", "reimbursement", reimbursement.id, after={"status": reimbursement.status}, trace=f"audit-{audit_id}")
    if audit.decision != "auto_approve" and missing:
        raise BusinessError("没有有效审批链", 409, "CHAIN_MISSING")


async def decide_approval(session, approval_id, actor, body):
    """条件更新防重签，先验证轮次和身份，签署结果与下一节点同事务提交。"""
    record = await get(session, ApprovalRecord, approval_id)
    round_item = await get(session, ApprovalRound, record.round_id)
    reimbursement = await get(session, Reimbursement, round_item.reimbursement_id)
    if record.approver_id != actor["id"] or reimbursement.user_id == actor["id"]:
        raise BusinessError("只能由指定的非提交人审批", 403, "APPROVER_REQUIRED")
    if record.result is not None or round_item.status != "active" or reimbursement.revision != round_item.revision or reimbursement.status != "in_approval":
        raise BusinessError("审批已处理或不是当前轮次", 409, "APPROVAL_CONFLICT")
    # 用户角色变化立即生效，冻结人员安排不意味着允许已降级身份继续签署。
    expected_role = round_item.chain_snapshot["nodes"][record.node_index]["role"]
    if actor["role"] != expected_role:
        raise BusinessError("审批人当前角色与冻结链不一致，需恢复身份后处理", 409, "CHAIN_INVALID")
    changed = await session.execute(update(ApprovalRecord).where(ApprovalRecord.id == record.id, ApprovalRecord.result.is_(None))
        .values(result=body.action, opinion=body.opinion, decided_at=utcnow()))
    if changed.rowcount != 1:
        raise BusinessError("审批已由其他请求处理", 409, "APPROVAL_CONFLICT")
    if body.action == "reject":
        round_item.status = "rejected"
        transition(reimbursement, "rejected")
    elif body.action == "return":
        round_item.status = "returned"
        transition(reimbursement, "draft")
        reimbursement.revision += 1
    else:
        nodes = round_item.chain_snapshot["nodes"]
        next_index = record.node_index + 1
        if next_index >= len(nodes):
            round_item.status = "approved"
            transition(reimbursement, "approved")
        else:
            user = await get(session, User, nodes[next_index]["approver_id"])
            role = await get(session, Role, user.role_id)
            if not user.active or user.id == reimbursement.user_id or role.code != nodes[next_index]["role"]:
                raise BusinessError("下一审批人已停用或属于提交人，不能推进", 409, "CHAIN_INVALID")
            session.add(ApprovalRecord(round_id=round_item.id, node_index=next_index, approver_id=user.id))
    log(session, actor, "approval." + body.action, "reimbursement", reimbursement.id,
        after={"approval_id": approval_id, "opinion": body.opinion, "status": reimbursement.status}, trace=f"audit-{round_item.audit_id}")
    await session.flush()
    return {"approval": serialize(record), "round": serialize(round_item), "reimbursement": serialize(reimbursement)}