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

"""报销与审核接口：提交快返回，上传逐文件提交后推送真实 SSE 进度。"""
import asyncio
import hashlib
import json
import time
from datetime import datetime
from zoneinfo import ZoneInfo
from fastapi import APIRouter, Depends, File, Form, Header, Query, Request, UploadFile
from fastapi.responses import FileResponse, StreamingResponse
from sqlalchemy import select, func
from sqlalchemy.dialects.sqlite import insert
from app.api.dependencies import current_user, read_db, write_db
from app.api.management import ok
from app.domain.models import AgentExecution, ApprovalRecord, ApprovalRound, AuditRun, HumanFeedback, Invoice, OutboxEvent, Reimbursement, RuleEvaluation, SequenceCounter
from app.application.services import audit_snapshot, get, guard, log, normalized_fields, readable, require, revision_guard, serialize
from app.infrastructure.database import Session, transaction
from app.infrastructure.files import file_path, save_file, validate_file
from app.shared.common import BusinessError, cents, paginate, transition, utcnow
from app.shared.config import settings
from app.shared import schemas as S

router = APIRouter()


async def detail(session, row):
    """详情合并真实关联记录，前端刷新即可恢复上传、审核和审批进度。"""
    invoices = (await session.scalars(select(Invoice).where(Invoice.reimbursement_id == row.id, Invoice.active.is_(True)).order_by(Invoice.id))).all()
    audits = (await session.scalars(select(AuditRun).where(AuditRun.reimbursement_id == row.id).order_by(AuditRun.id.desc()))).all()
    rounds = (await session.scalars(select(ApprovalRound).where(ApprovalRound.reimbursement_id == row.id).order_by(ApprovalRound.id.desc()))).all()
    records = (await session.scalars(select(ApprovalRecord).where(ApprovalRecord.round_id.in_([item.id for item in rounds])).order_by(ApprovalRecord.id))).all() if rounds else []
    return {**serialize(row), "invoices": [serialize(item) for item in invoices], "audits": [serialize(item) for item in audits],
            "approval_rounds": [serialize(item) for item in rounds], "approvals": [serialize(item) for item in records]}


@router.get("/reimbursements", tags=["报销"])
async def reimbursements(request: Request, page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100),
    status: str | None = None, type: int | None = None, actor=Depends(current_user), session=Depends(read_db)):
    """功能：按授权范围列单。参数：分页/状态/类型。返回：报销分页。异常：401未登录，422非法分页。"""
    rows = (await session.scalars(select(Reimbursement).order_by(Reimbursement.id.desc()))).all()
    rows = [row for row in rows if readable(actor, row) and (status is None or row.status == status) and (type is None or row.type == type)]
    return ok(request, paginate([serialize(item) for item in rows], page, page_size))


@router.post("/reimbursements", status_code=201, tags=["报销"])
async def create_reimbursement(body: S.ReimbursementCreate, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：创建本人草稿。参数：标题/类型/含税额。返回：草稿。异常：422金额，409序号耗尽。"""
    from app.domain.models import Department
    department = await get(session, Department, actor["department_id"])
    if not department.active:
        raise BusinessError("所属部门已停用，不能发起新申请", 409)
    day = datetime.now(ZoneInfo(settings.business_timezone)).strftime("%Y%m%d")
    await session.execute(insert(SequenceCounter).values(date_key=day, value=1).on_conflict_do_update(index_elements=["date_key"], set_={"value": SequenceCounter.value + 1}))
    counter = await session.get(SequenceCounter, day)
    if counter.value > 999999:
        raise BusinessError("当日单号序列已耗尽", 409)
    row = Reimbursement(number=f"BX{day}{counter.value:06d}", user_id=actor["id"], department_id=actor["department_id"],
        type=body.type, title=body.title, description=body.description, total_cents=cents(body.total_amount))
    session.add(row)
    await session.flush()
    log(session, actor, "reimbursement.created", "reimbursement", row.id, after=serialize(row))
    return ok(request, serialize(row))


@router.get("/reimbursements/{identifier}", tags=["报销"])
async def reimbursement_detail(identifier: int, request: Request, actor=Depends(current_user), session=Depends(read_db)):
    """功能：查看单据全历史。参数：ID。返回：发票/审核/审批。异常：403无权，404不存在。"""
    row = await get(session, Reimbursement, identifier)
    guard(actor, row)
    return ok(request, await detail(session, row))


@router.patch("/reimbursements/{identifier}", tags=["报销"])
async def patch_reimbursement(identifier: int, body: S.ReimbursementPatch, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：修改本人可编辑单。参数：revision及业务字段。返回：新版本。异常：403所有权，409状态/版本，422金额。"""
    row = await get(session, Reimbursement, identifier)
    guard(actor, row, True)
    revision_guard(row, body.revision)
    before = serialize(row)
    for key in ("title", "description", "type"):
        if getattr(body, key) is not None:
            setattr(row, key, getattr(body, key))
    if body.total_amount is not None:
        row.total_cents = cents(body.total_amount)
    row.revision += 1
    log(session, actor, "reimbursement.updated", "reimbursement", row.id, before, serialize(row))
    return ok(request, serialize(row))


def sse(event, data, identifier=None):
    """统一 SSE 帧格式，中文 JSON 不影响以双换行识别事件边界。"""
    return (f"id: {identifier}\n" if identifier is not None else "") + f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"


@router.post("/reimbursements/{identifier}/invoices", tags=["发票"])
async def upload_invoices(identifier: int, request: Request, revision: int = Form(), metadata: str = Form(default="[]"),
    files: list[UploadFile] = File(), actor=Depends(current_user), session=Depends(read_db)):
    """功能：逐文件上传并流式留痕。参数：multipart files/revision/metadata数组。返回：SSE真实进度。异常：前置403/409/422；流内error含部分成功ID。"""
    row = await get(session, Reimbursement, identifier)
    guard(actor, row, True)
    revision_guard(row, revision)
    if not 1 <= len(files) <= settings.max_batch_files:
        raise BusinessError("批量文件数超过限制")
    try:
        raw_metadata = json.loads(metadata)
        if not isinstance(raw_metadata, list) or len(raw_metadata) not in {0, len(files)}:
            raise ValueError()
        normalized = [normalized_fields(raw_metadata[index] if raw_metadata else {}) for index in range(len(files))]
    except Exception:
        raise BusinessError("metadata 必须是与文件一一对应的合法发票字段数组")

    async def stream():
        current_revision = revision
        uploaded = []
        try:
            yield sse("started", {"total": len(files), "revision": current_revision})
            for index, file in enumerate(files):
                if await request.is_disconnected():
                    break
                content = await file.read(settings.max_upload_mb * 1024 * 1024 + 1)
                mime, extension = await asyncio.to_thread(validate_file, content)
                key = await asyncio.to_thread(save_file, content, extension)
                try:
                    async with transaction() as writer:
                        reimbursement = await get(writer, Reimbursement, identifier)
                        guard(actor, reimbursement, True)
                        revision_guard(reimbursement, current_revision)
                        fields = normalized[index]
                        item = Invoice(reimbursement_id=identifier, filename=(file.filename or "invoice")[:250], file_key=key,
                            file_hash=hashlib.sha256(content).hexdigest(), content_type=mime, declared_fields=fields,
                            net_cents=cents(fields["amount"]) if fields.get("amount") is not None else None,
                            tax_cents=cents(fields["tax_amount"]) if fields.get("tax_amount") is not None else None)
                        writer.add(item)
                        reimbursement.revision += 1
                        next_revision = reimbursement.revision
                        await writer.flush()
                        log(writer, actor, "invoice.uploaded", "reimbursement", identifier, after={"invoice_id": item.id, "hash": item.file_hash})
                except Exception:
                    # 尚未关联数据库的孤立文件可以安全清理；已提交历史文件绝不随意删除。
                    file_path(key).unlink(missing_ok=True)
                    raise
                # 只有事务真正提交后才能向客户端报告新版本和成功 ID。
                current_revision = next_revision
                uploaded.append(item.id)
                yield sse("progress", {"current": index + 1, "total": len(files), "filename": item.filename, "invoice_id": item.id, "revision": current_revision})
            yield sse("completed", {"reimbursement_id": identifier, "uploaded_ids": uploaded, "revision": current_revision})
        except Exception as exc:
            yield sse("error", {"code": exc.code if isinstance(exc, BusinessError) else "UPLOAD_FAILED",
                "message": str(exc) if isinstance(exc, BusinessError) else "文件处理失败", "uploaded_ids": uploaded, "revision": current_revision})
        finally:
            for file in files:
                await file.close()
    return StreamingResponse(stream(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})


@router.patch("/invoices/{identifier}", tags=["发票"])
async def patch_invoice(identifier: int, body: S.InvoicePatch, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：用户补录未知字段。参数：ID/单据revision/fields。返回：发票与新revision。异常：403非本人，409状态，422字段。"""
    invoice = await get(session, Invoice, identifier)
    row = await get(session, Reimbursement, invoice.reimbursement_id)
    guard(actor, row, True)
    revision_guard(row, body.revision)
    if not invoice.active:
        raise BusinessError("发票已移除", 409)
    fields = normalized_fields(body.fields)
    before = serialize(invoice)
    invoice.declared_fields = fields
    invoice.net_cents = cents(fields["amount"]) if fields.get("amount") is not None else None
    invoice.tax_cents = cents(fields["tax_amount"]) if fields.get("tax_amount") is not None else None
    row.revision += 1
    log(session, actor, "invoice.updated", "reimbursement", row.id, before, serialize(invoice))
    return ok(request, {"invoice": serialize(invoice), "revision": row.revision})


@router.delete("/invoices/{identifier}", tags=["发票"])
async def delete_invoice(identifier: int, request: Request, revision: int = Query(ge=1), actor=Depends(current_user), session=Depends(write_db)):
    """功能：从可编辑单移除发票并保留历史证据。参数：ID/revision。返回：新版本。异常：403/404/409。"""
    invoice = await get(session, Invoice, identifier)
    row = await get(session, Reimbursement, invoice.reimbursement_id)
    guard(actor, row, True)
    revision_guard(row, revision)
    invoice.active = False
    row.revision += 1
    log(session, actor, "invoice.removed", "reimbursement", row.id, after={"invoice_id": identifier})
    return ok(request, {"removed": True, "revision": row.revision})


@router.get("/invoices/{identifier}/file", tags=["发票"])
async def download_invoice(identifier: int, actor=Depends(current_user), session=Depends(read_db)):
    """功能：按所有权读取原始文件。参数：ID。返回：文件流。异常：403权限，404文件不存在；不接受磁盘路径。"""
    invoice = await get(session, Invoice, identifier)
    guard(actor, await get(session, Reimbursement, invoice.reimbursement_id))
    return FileResponse(file_path(invoice.file_key), media_type=invoice.content_type, filename=invoice.filename)


async def create_audit(session, row, actor, key, reason="提交审核"):
    """快照、审核占位和事件与单据同事务提交，后台只能读取完整已提交输入。"""
    active = await session.scalar(select(AuditRun).where(AuditRun.reimbursement_id == row.id, AuditRun.status.in_(["queued", "running"])))
    if active:
        raise BusinessError("已有活动审核", 409, "AUDIT_ACTIVE")
    row.revision += 1
    if row.status != "pending_audit":
        transition(row, "pending_audit")
    row.submitted_at = utcnow()
    snapshot = await audit_snapshot(session, row)
    if not snapshot["invoice_files"]:
        raise BusinessError("请至少上传一张发票")
    number = (await session.scalar(select(func.count()).select_from(AuditRun).where(AuditRun.reimbursement_id == row.id))) + 1
    audit = AuditRun(reimbursement_id=row.id, revision=row.revision, run_no=number, snapshot_json=snapshot)
    session.add(audit)
    await session.flush()
    session.add(OutboxEvent(event_type="audit", business_key=key, payload_json={"audit_id": audit.id}))
    log(session, actor, "reimbursement.submitted", "reimbursement", row.id, after={"audit_id": audit.id, "revision": row.revision, "reason": reason}, trace=f"audit-{audit.id}")
    return {"reimbursement": serialize(row), "audit_id": audit.id}


@router.post("/reimbursements/{identifier}/submit", status_code=202, tags=["报销"])
async def submit(identifier: int, body: S.Revision, request: Request,
    idempotency_key: str = Header(alias="Idempotency-Key", min_length=8, max_length=80), actor=Depends(current_user), session=Depends(write_db)):
    """功能：异步提交。参数：revision/Idempotency-Key。返回：202审核ID。异常：403所有权，409版本/活动任务，422无发票。"""
    row = await get(session, Reimbursement, identifier)
    if row.user_id != actor["id"]:
        raise BusinessError("只能提交本人单据", 403)
    key = f"submit:{actor['id']}:{identifier}:{idempotency_key}"
    previous = await session.scalar(select(OutboxEvent).where(OutboxEvent.business_key == key))
    if previous:
        return ok(request, {"reimbursement": serialize(row), "audit_id": previous.payload_json["audit_id"]}, "幂等返回已受理任务")
    guard(actor, row, True)
    revision_guard(row, body.revision)
    return ok(request, await create_audit(session, row, actor, key))


@router.post("/reimbursements/{identifier}/withdraw", tags=["报销"])
async def withdraw(identifier: int, body: S.Revision, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：撤回待审单使旧任务失效。参数：ID/revision。返回：终态。异常：403非本人，409非待审核/过期版本。"""
    row = await get(session, Reimbursement, identifier)
    if row.user_id != actor["id"]:
        raise BusinessError("只能撤回本人单据", 403)
    revision_guard(row, body.revision)
    transition(row, "withdrawn")
    row.revision += 1
    runs = (await session.scalars(select(AuditRun).where(AuditRun.reimbursement_id == row.id, AuditRun.status.in_(["queued", "running"])))).all()
    for run in runs:
        run.status, run.finished_at = "cancelled", utcnow()
    log(session, actor, "reimbursement.withdrawn", "reimbursement", row.id, after=serialize(row))
    return ok(request, serialize(row))


@router.get("/audit", tags=["审核"])
async def audits(request: Request, page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100),
    status: str | None = None, risk_level: int | None = None, decision: str | None = None, reimbursement_id: int | None = None,
    actor=Depends(current_user), session=Depends(read_db)):
    """功能：列出有权查看的历史审核。参数：状态/风险/决策/单据/分页。返回：审核分页。异常：401身份。"""
    rows = (await session.scalars(select(AuditRun).order_by(AuditRun.id.desc()))).all()
    values = []
    for row in rows:
        reimbursement = await get(session, Reimbursement, row.reimbursement_id)
        if readable(actor, reimbursement) and (status is None or row.status == status) and (risk_level is None or row.risk_level == risk_level) and (decision is None or row.decision == decision) and (reimbursement_id is None or row.reimbursement_id == reimbursement_id):
            data = serialize(row)
            data.pop("snapshot", None)
            data.pop("result", None)
            values.append({**data, "reimbursement_no": reimbursement.number, "title": reimbursement.title})
    return ok(request, paginate(values, page, page_size))


async def audit_detail(session, audit):
    nodes = (await session.scalars(select(AgentExecution).where(AgentExecution.audit_id == audit.id).order_by(AgentExecution.id))).all()
    evaluations = (await session.scalars(select(RuleEvaluation).where(RuleEvaluation.audit_id == audit.id).order_by(RuleEvaluation.id))).all()
    feedback = (await session.scalars(select(HumanFeedback).where(HumanFeedback.audit_id == audit.id))).all()
    data = serialize(audit)
    # 文件键和向量是内部细节，历史证据返回文本与版本即可。
    snapshot = {**data["snapshot"]}
    snapshot["invoice_files"] = [{key: value for key, value in item.items() if key != "file_key"} for item in snapshot.get("invoice_files", [])]
    snapshot["policies"] = [{key: value for key, value in item.items() if key != "vector"} for item in snapshot.get("policies", [])]
    return {**data, "snapshot": snapshot, "nodes": [serialize(item) for item in nodes], "rules": [serialize(item) for item in evaluations], "feedback": [serialize(item) for item in feedback]}


@router.get("/audit/{identifier}", tags=["审核"])
async def get_audit(identifier: int, request: Request, actor=Depends(current_user), session=Depends(read_db)):
    """功能：完整证据与版本详情。参数：审核ID。返回：快照/节点/规则/反馈。异常：403无权，404不存在。"""
    audit = await get(session, AuditRun, identifier)
    guard(actor, await get(session, Reimbursement, audit.reimbursement_id))
    return ok(request, await audit_detail(session, audit))


@router.get("/audit/{identifier}/events", tags=["审核"])
async def audit_events(identifier: int, request: Request, last_event_id: str | None = Header(default=None, alias="Last-Event-ID"),
    actor=Depends(current_user), session=Depends(read_db)):
    """功能：实时审核进度。参数：ID/Last-Event-ID。返回：SSE节点/完成/失败/心跳。异常：前置403/404；断线用详情恢复。"""
    audit = await get(session, AuditRun, identifier)
    guard(actor, await get(session, Reimbursement, audit.reimbursement_id))
    try:
        last = int(last_event_id or 0)
    except ValueError:
        raise BusinessError("Last-Event-ID 必须为整数")

    async def stream():
        cursor = last
        yield sse("started", {"audit_id": identifier})
        while not await request.is_disconnected():
            async with Session() as reader:
                current = await get(reader, AuditRun, identifier)
                nodes = (await reader.scalars(select(AgentExecution).where(AgentExecution.audit_id == identifier, AgentExecution.id > cursor).order_by(AgentExecution.id))).all()
                for node in nodes:
                    cursor = node.id
                    yield sse("node", serialize(node), cursor)
                if current.status in {"completed", "failed", "cancelled"}:
                    yield sse("completed" if current.status == "completed" else "failed", {"status": current.status, "decision": current.decision, "error_code": current.error_code})
                    break
            yield ": heartbeat\n\n"
            await asyncio.sleep(1)
    return StreamingResponse(stream(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})


@router.post("/audit/{identifier}/rerun", status_code=202, tags=["审核"])
async def rerun(identifier: int, body: S.Rerun, request: Request,
    idempotency_key: str = Header(alias="Idempotency-Key", min_length=8, max_length=80), actor=Depends(current_user), session=Depends(write_db)):
    """功能：授权重跑保留旧结果。参数：审核ID/理由/幂等键。返回：新审核ID。异常：403权限，409终态/在途审批。"""
    require(actor, "auditor")
    audit = await get(session, AuditRun, identifier)
    row = await get(session, Reimbursement, audit.reimbursement_id)
    key = f"rerun:{actor['id']}:{identifier}:{idempotency_key}"
    previous = await session.scalar(select(OutboxEvent).where(OutboxEvent.business_key == key))
    if previous:
        return ok(request, {"audit_id": previous.payload_json["audit_id"], "reimbursement": serialize(row)})
    pending = await session.scalar(select(ApprovalRound).where(ApprovalRound.reimbursement_id == row.id, ApprovalRound.status == "active"))
    if pending or row.status not in {"pending_audit", "rejected"}:
        raise BusinessError("在途审批、草稿或终态不能重跑", 409, "RERUN_FORBIDDEN")
    # 旧审核完成后的派单事件失效，避免重跑新版本后又拉起旧轮次。
    return ok(request, await create_audit(session, row, actor, key, body.reason))


@router.post("/audit/{identifier}/feedback", tags=["反馈"])
async def feedback(identifier: int, body: S.Feedback, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：人工审批后独立反馈。参数：已处理审批ID/纠正/理由。返回：反馈及迭代建议。异常：403非签署人，409轮次不匹配。"""
    started = time.perf_counter()
    from app.domain.models import AgentConfig
    from app.application.seed import latest_configs
    agent = (await latest_configs(session, AgentConfig, "node_name"))["feedback"]
    if not agent.enabled:
        raise BusinessError("反馈节点已停用", 409)
    audit = await get(session, AuditRun, identifier)
    record = await get(session, ApprovalRecord, body.approval_record_id)
    round_item = await get(session, ApprovalRound, record.round_id)
    if record.approver_id != actor["id"] or record.result is None:
        raise BusinessError("只有已完成该审批的人可以反馈", 403)
    if round_item.audit_id != audit.id:
        raise BusinessError("审批与审核轮次不匹配", 409)
    item = HumanFeedback(audit_id=audit.id, approval_record_id=record.id, actor_id=actor["id"], corrected_fields=body.corrected_fields,
        reason=body.reason, model_update_needed=bool(body.corrected_fields))
    session.add(item)
    await session.flush()
    previous = (await session.scalars(select(AgentExecution).where(AgentExecution.audit_id == audit.id, AgentExecution.node_name == "feedback"))).all()
    session.add(AgentExecution(audit_id=audit.id, node_name="feedback", attempt=len(previous) + 1, status="success", duration_ms=int((time.perf_counter() - started) * 1000),
        input_json={"feedback_id": item.id, "agent_version": agent.version}, output_json={"model_update_needed": item.model_update_needed, "training_executed": False}))
    log(session, actor, "audit.feedback", "audit", audit.id, after=serialize(item), trace=f"audit-{audit.id}")
    return ok(request, serialize(item))