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

"""审批、知识和运营 API：静态路径先注册，避免 /export 等被动态 ID 吞掉。"""
import csv
from datetime import datetime, timezone
from decimal import Decimal
from io import BytesIO, StringIO
from pathlib import Path
from xml.sax.saxutils import escape
from fastapi import APIRouter, Depends, File, Query, Request, UploadFile
from fastapi.responses import Response
from sqlalchemy import select
from app.api.dependencies import current_user, read_db, write_db
from app.api.management import ok
from app.domain.models import Alert, ApprovalChain, ApprovalRecord, ApprovalRound, AuditRun, KnowledgeDocument, OperationLog, OutboxEvent, Reimbursement
from app.application.approvals import decide_approval, validate_chain
from app.application.services import get, guard, log, policy_snapshot, readable, replace_chunks, require, search_policies, serialize
from app.infrastructure.ai import AIUnavailable
from app.shared.common import BusinessError, cents, money, paginate, utcnow
from app.shared.config import settings
from app.shared import schemas as S

router = APIRouter()


@router.get("/approvals/chains", tags=["审批链"])
async def chains(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(ApprovalChain).order_by(ApprovalChain.id.desc()))).all()
    return ok(request, paginate([serialize(item) for item in rows], page, page_size))


@router.post("/approvals/chains", status_code=201, tags=["审批链"])
async def create_chain(body: S.ChainBody, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：创建匹配条件与顺序链。参数：route/范围/nodes。返回：链配置。异常：403权限，422非法人员或金额范围。"""
    require(actor, "manager")
    minimum, maximum = await validate_chain(session, body)
    row = ApprovalChain(name=body.name, route=body.route, type=body.type, department_id=body.department_id,
        amount_min_cents=minimum, amount_max_cents=maximum, priority=body.priority, nodes_json=[node.model_dump() for node in body.nodes], active=body.active)
    session.add(row)
    await session.flush()
    log(session, actor, "chain.created", "chain", row.id, after=serialize(row))
    # 新增链也可能修复“无人接单”，恢复已经耗尽重试的派单事件。
    events = (await session.scalars(select(OutboxEvent).where(OutboxEvent.event_type == "approval", OutboxEvent.status == "failed"))).all()
    for event in events:
        event.status, event.attempts, event.next_attempt_at = "pending", 0, utcnow()
    return ok(request, serialize(row))


@router.patch("/approvals/chains/{identifier}", tags=["审批链"])
async def patch_chain(identifier: int, body: S.ChainBody, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：更新下一轮链配置。参数：ID/expected_version/完整链。返回：新版本。异常：403、404、409版本、422非法链。"""
    require(actor, "manager")
    row = await get(session, ApprovalChain, identifier)
    if body.expected_version != row.version:
        raise BusinessError("审批链版本已更新", 409, "VERSION_CONFLICT")
    minimum, maximum = await validate_chain(session, body)
    before = serialize(row)
    row.name, row.route, row.type, row.department_id = body.name, body.route, body.type, body.department_id
    row.amount_min_cents, row.amount_max_cents, row.priority = minimum, maximum, body.priority
    row.nodes_json, row.active, row.version = [node.model_dump() for node in body.nodes], body.active, row.version + 1
    log(session, actor, "chain.updated", "chain", row.id, before, serialize(row))
    # 修复配置后允许失败派单事件再次处理，恢复挂单不必篡改单据状态。
    events = (await session.scalars(select(OutboxEvent).where(OutboxEvent.event_type == "approval", OutboxEvent.status == "failed"))).all()
    for event in events:
        event.status, event.attempts, event.next_attempt_at = "pending", 0, utcnow()
    return ok(request, serialize(row))


@router.get("/approvals", tags=["审批"])
async def approvals(request: Request, page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100),
    pending: bool | None = None, reimbursement_id: int | None = None, actor=Depends(current_user), session=Depends(read_db)):
    """功能：查看本人待办与授权历史。参数：pending/单据/分页。返回：记录与关联信息。异常：401身份。"""
    rows = (await session.scalars(select(ApprovalRecord).order_by(ApprovalRecord.id.desc()))).all()
    values = []
    for record in rows:
        round_item = await get(session, ApprovalRound, record.round_id)
        reimbursement = await get(session, Reimbursement, round_item.reimbursement_id)
        is_pending = record.result is None and round_item.status == "active" and reimbursement.revision == round_item.revision
        can_see = record.approver_id == actor["id"] or pending is not True and readable(actor, reimbursement)
        if can_see and (pending is None or is_pending == pending) and (reimbursement_id is None or reimbursement.id == reimbursement_id):
            values.append({**serialize(record), "audit_id": round_item.audit_id, "reimbursement_id": reimbursement.id,
                "number": reimbursement.number, "title": reimbursement.title, "total_amount": money(reimbursement.total_cents),
                "round_status": round_item.status, "can_decide": is_pending and record.approver_id == actor["id"] and reimbursement.user_id != actor["id"]})
    return ok(request, paginate(values, page, page_size))


@router.post("/approvals/{identifier}/decide", tags=["审批"])
async def approval_decision(identifier: int, body: S.Decide, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：处理指定审批并原子推进下一棒。参数：ID/action/opinion。返回：记录/轮次/单据。异常：403自审/非指定人，409重签。"""
    require(actor, "supervisor")
    return ok(request, await decide_approval(session, identifier, actor, body))


@router.post("/knowledge/documents/import", status_code=201, tags=["知识库"])
async def import_document(request: Request, file: UploadFile = File(), actor=Depends(current_user), session=Depends(write_db)):
    """功能：导入真实文本政策。参数：UTF-8 txt/md文件。返回：政策和索引状态。异常：403权限，422文件类型/大小/编码。"""
    require(actor, "auditor")
    if Path(file.filename or "").suffix.lower() not in {".txt", ".md"}:
        raise BusinessError("政策导入只接受 UTF-8 txt/md")
    try:
        raw = await file.read(600001)
        content = raw.decode("utf-8-sig")
        body = S.KnowledgeBody(title=Path(file.filename).stem[:200], content=content)
        if len(raw) > 600000:
            raise ValueError()
    except Exception:
        raise BusinessError("文件必须是有效 UTF-8，且正文最多200000字符")
    finally:
        await file.close()
    row = KnowledgeDocument(title=body.title, content=body.content, category=body.category)
    session.add(row)
    await session.flush()
    await replace_chunks(session, row)
    log(session, actor, "knowledge.imported", "knowledge", row.id, after={"title": row.title, "version": row.version})
    return ok(request, serialize(row))


@router.get("/knowledge/documents", tags=["知识库"])
async def documents(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)):
    """功能：阅读政策与索引状态。参数：分页。返回：含正文政策列表。异常：401未登录。"""
    rows = (await session.scalars(select(KnowledgeDocument).order_by(KnowledgeDocument.id.desc()))).all()
    return ok(request, paginate([serialize(item) for item in rows], page, page_size))


@router.post("/knowledge/documents", status_code=201, tags=["知识库"])
async def create_document(body: S.KnowledgeBody, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：保存用户制度原文并分块。参数：标题/分类/正文。返回：未索引文档。异常：403权限，422正文不合法。"""
    require(actor, "auditor")
    row = KnowledgeDocument(title=body.title, content=body.content, category=body.category, active=body.active)
    session.add(row)
    await session.flush()
    await replace_chunks(session, row)
    log(session, actor, "knowledge.created", "knowledge", row.id, after={"title": row.title, "version": row.version})
    return ok(request, serialize(row))


@router.patch("/knowledge/documents/{identifier}", tags=["知识库"])
async def patch_document(identifier: int, body: S.KnowledgeBody, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：修订正文使旧向量失效。参数：ID/预期版本/正文。返回：新版本。异常：403权限，404不存在，409并发。"""
    require(actor, "auditor")
    row = await get(session, KnowledgeDocument, identifier)
    if body.expected_version != row.version:
        raise BusinessError("政策版本已更新", 409, "VERSION_CONFLICT")
    before = {"title": row.title, "version": row.version, "active": row.active}
    row.title, row.content, row.category, row.active = body.title, body.content, body.category, body.active
    row.version += 1
    await replace_chunks(session, row)
    log(session, actor, "knowledge.updated", "knowledge", row.id, before, {"title": row.title, "version": row.version, "active": row.active})
    return ok(request, serialize(row))


@router.post("/knowledge/documents/{identifier}/reindex", status_code=202, tags=["知识库"])
async def reindex_document(identifier: int, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：排队真实向量化。参数：文档ID。返回：索引任务ID。异常：403权限，404不存在，409已有任务，503无模型。"""
    require(actor, "auditor")
    if not settings.embedding_base_url and not settings.llm_base_url:
        raise BusinessError("未配置真实 Embedding 服务，不能生成向量", 503, "EMBEDDING_DISABLED")
    row = await get(session, KnowledgeDocument, identifier)
    key = f"index:{identifier}:{row.version}:{settings.embedding_model}"
    event = await session.scalar(select(OutboxEvent).where(OutboxEvent.business_key == key))
    if event and event.status in {"pending", "processing"}:
        return ok(request, {"task_id": event.id, "status": row.index_status})
    if event:
        event.status, event.attempts, event.next_attempt_at = "pending", 0, utcnow()
    else:
        event = OutboxEvent(event_type="index", business_key=key, payload_json={"document_id": identifier, "version": row.version})
        session.add(event)
        await session.flush()
    row.index_status = "queued"
    log(session, actor, "knowledge.reindex", "knowledge", identifier, after={"task_id": event.id})
    return ok(request, {"task_id": event.id, "status": row.index_status})


@router.post("/knowledge/search", tags=["知识库"])
async def search(body: S.SearchBody, request: Request, actor=Depends(current_user), session=Depends(read_db)):
    """功能：显式关键词/语义试验。参数：query/mode/top_k/category。返回：模式及原文出处。异常：503真实向量不可用。"""
    try:
        values = await search_policies(body.query, await policy_snapshot(session), body.mode, body.top_k, body.category)
    except AIUnavailable as exc:
        raise BusinessError(str(exc), 503, "SEMANTIC_UNAVAILABLE")
    return ok(request, {"mode": body.mode, "items": values})


def time_bounds(since, until):
    """过滤边界统一转换 UTC，拒绝不合法或逆序区间。"""
    try:
        def parse(value):
            if not value:
                return None
            item = datetime.fromisoformat(value.replace("Z", "+00:00"))
            return item.astimezone(timezone.utc).replace(tzinfo=None) if item.tzinfo else item
        start, end = parse(since), parse(until)
        if start and end and start > end:
            raise ValueError()
        return start, end
    except ValueError:
        raise BusinessError("时间范围必须为有效且有序的 ISO 日期/时间")


async def filtered_logs(session, actor, action, trace, actor_id, resource, since, until):
    require(actor, "auditor")
    start, end = time_bounds(since, until)
    query = select(OperationLog).order_by(OperationLog.id.desc())
    if action:
        query = query.where(OperationLog.action.contains(action))
    if trace:
        query = query.where(OperationLog.trace_id == trace)
    if actor_id is not None:
        query = query.where(OperationLog.actor_id == actor_id)
    if resource:
        query = query.where(OperationLog.resource_type == resource)
    if start:
        query = query.where(OperationLog.created_at >= start)
    if end:
        query = query.where(OperationLog.created_at <= end)
    return (await session.scalars(query)).all()


@router.get("/logs/export", tags=["日志"])
async def logs_export(action: str | None = None, trace: str | None = None, actor_id: int | None = None, resource: str | None = None,
    since: str | None = None, until: str | None = None, actor=Depends(current_user), session=Depends(read_db)):
    """功能：导出授权日志。参数：与列表相同筛选。返回：UTF-8 BOM CSV。异常：403权限，422时间格式。"""
    rows = await filtered_logs(session, actor, action, trace, actor_id, resource, since, until)
    stream = StringIO()
    writer = csv.writer(stream)
    writer.writerow(["记录ID", "时间UTC", "操作人ID", "动作", "资源", "资源ID", "追踪ID"])
    for row in rows:
        writer.writerow([row.id, str(row.created_at), row.actor_id, row.action, row.resource_type, row.resource_id, row.trace_id])
    return Response(("\ufeff" + stream.getvalue()).encode("utf-8"), media_type="text/csv", headers={"Content-Disposition": 'attachment; filename="audit-logs.csv"'})


@router.get("/logs", tags=["日志"])
async def logs(request: Request, page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100),
    action: str | None = None, trace: str | None = None, actor_id: int | None = None, resource: str | None = None,
    since: str | None = None, until: str | None = None, actor=Depends(current_user), session=Depends(read_db)):
    """功能：按人/动作/资源/追踪查证。参数：筛选和分页。返回：只读日志。异常：403权限，422时间格式。"""
    rows = await filtered_logs(session, actor, action, trace, actor_id, resource, since, until)
    return ok(request, paginate([serialize(item) for item in rows], page, page_size))


@router.get("/alerts", tags=["告警"])
async def alerts(request: Request, page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100),
    status: str | None = None, severity: str | None = None, type: str | None = None, actor=Depends(current_user), session=Depends(read_db)):
    """功能：查询授权告警。参数：状态/严重度/类型/分页。返回：持久告警。异常：403权限不足。"""
    require(actor, "supervisor")
    rows = (await session.scalars(select(Alert).order_by(Alert.id.desc()))).all()
    values = []
    for row in rows:
        if row.reimbursement_id:
            reimbursement = await get(session, Reimbursement, row.reimbursement_id)
            if not readable(actor, reimbursement):
                continue
        elif actor["role"] not in {"auditor", "manager", "admin"}:
            continue
        if (status is None or row.status == status) and (severity is None or row.severity == severity) and (type is None or row.type == type):
            values.append(serialize(row))
    return ok(request, paginate(values, page, page_size))


async def manage_alert(session, identifier, actor, opinion, target):
    require(actor, "supervisor")
    row = await get(session, Alert, identifier)
    if row.reimbursement_id:
        guard(actor, await get(session, Reimbursement, row.reimbursement_id))
    else:
        require(actor, "auditor")
    expected = "open" if target == "acknowledged" else "acknowledged"
    if row.status != expected:
        raise BusinessError("告警必须按未处理→已确认→已解决推进", 409, "ALERT_STATE")
    before = serialize(row)
    row.status, row.opinion = target, opinion
    if target == "acknowledged":
        row.ack_by, row.acknowledged_at = actor["id"], utcnow()
    else:
        row.resolved_by, row.resolved_at = actor["id"], utcnow()
    log(session, actor, "alert." + target, "alert", row.id, before, serialize(row))
    return serialize(row)


@router.post("/alerts/{identifier}/acknowledge", tags=["告警"])
async def acknowledge(identifier: int, body: S.Opinion, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：确认异常负责人。参数：ID/意见。返回：已确认告警。异常：403范围，404不存在，409已处理。"""
    return ok(request, await manage_alert(session, identifier, actor, body.opinion, "acknowledged"))


@router.post("/alerts/{identifier}/resolve", tags=["告警"])
async def resolve(identifier: int, body: S.Opinion, request: Request, actor=Depends(current_user), session=Depends(write_db)):
    """功能：记录告警处理完成。参数：ID/意见。返回：已解决告警，原风险不变。异常：403范围，409未确认。"""
    return ok(request, await manage_alert(session, identifier, actor, body.opinion, "resolved"))


async def statistic_rows(session, actor, since, until, department_id):
    start, end = time_bounds(since, until)
    rows = (await session.scalars(select(Reimbursement).order_by(Reimbursement.id))).all()
    return [row for row in rows if readable(actor, row) and (department_id is None or row.department_id == department_id)
        and (start is None or row.created_at >= start) and (end is None or row.created_at <= end)]


@router.get("/stats/overview", tags=["统计"])
async def overview(request: Request, since: str | None = None, until: str | None = None, department_id: int | None = None,
    actor=Depends(current_user), session=Depends(read_db)):
    """功能：实时授权统计。参数：时间/部门。返回：数量、金额、状态、风险、部门。异常：401身份，422时间。"""
    rows = await statistic_rows(session, actor, since, until, department_id)
    ids = [row.id for row in rows]
    runs = (await session.scalars(select(AuditRun).where(AuditRun.reimbursement_id.in_(ids)).order_by(AuditRun.id))).all() if ids else []
    latest = {}
    for run in runs:
        latest[run.reimbursement_id] = run
    statuses, risks, departments = {}, {}, {}
    for row in rows:
        statuses[row.status] = statuses.get(row.status, 0) + 1
        group = departments.setdefault(row.department_id, {"department_id": row.department_id, "count": 0, "total_cents": 0})
        group["count"] += 1
        group["total_cents"] += row.total_cents
    for run in latest.values():
        risks[str(run.risk_level)] = risks.get(str(run.risk_level), 0) + 1
    return ok(request, {"total_count": len(rows), "total_amount": money(sum(row.total_cents for row in rows)),
        "approved_count": statuses.get("approved", 0), "pending_count": statuses.get("pending_audit", 0) + statuses.get("in_approval", 0),
        "statuses": statuses, "risks": risks, "departments": [{"department_id": item["department_id"], "count": item["count"], "total_amount": money(item["total_cents"])} for item in departments.values()]})


@router.get("/stats/export", tags=["统计"])
async def stats_export(format: str = Query(default="xlsx", pattern=r"^(xlsx|pdf)$"), since: str | None = None,
    until: str | None = None, department_id: int | None = None, actor=Depends(current_user), session=Depends(read_db)):
    """功能：导出授权财务明细。参数：format/时间/部门。返回：Excel或中文PDF。异常：403权限，422格式，503字体问题。"""
    require(actor, "manager")
    rows = await statistic_rows(session, actor, since, until, department_id)
    data = [[row.number, row.title, row.status, money(row.total_cents)] for row in rows]
    output = BytesIO()
    if format == "xlsx":
        from openpyxl import Workbook
        from openpyxl.styles import Font, PatternFill
        book = Workbook()
        sheet = book.active
        sheet.title = "报销明细"
        sheet.append(["报销单号", "报销说明", "状态", "含税金额（元）"])
        for line in data:
            sheet.append(["'" + value if isinstance(value, str) and value.startswith(("=", "+", "-", "@")) else value for value in line])
        for cell in sheet[1]:
            cell.font = Font(color="FFFFFF", bold=True)
            cell.fill = PatternFill("solid", fgColor="183C47")
        for column, width in zip("ABCD", [26, 45, 22, 22]):
            sheet.column_dimensions[column].width = width
        sheet.freeze_panes = "A2"
        sheet.auto_filter.ref = sheet.dimensions
        book.save(output)
        mime = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
    else:
        from reportlab.pdfbase import pdfmetrics
        from reportlab.pdfbase.cidfonts import UnicodeCIDFont
        from reportlab.pdfbase.ttfonts import TTFont
        from reportlab.lib import colors
        from reportlab.lib.styles import ParagraphStyle
        from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle
        if settings.pdf_font_path:
            pdfmetrics.registerFont(TTFont("ChineseReport", settings.pdf_font_path))
            font = "ChineseReport"
        else:
            pdfmetrics.registerFont(UnicodeCIDFont("STSong-Light"))
            font = "STSong-Light"
        style = ParagraphStyle("Chinese", fontName=font, fontSize=9, leading=13)
        doc = SimpleDocTemplate(output, title="财务审计报销明细", leftMargin=28, rightMargin=28)
        content = [Paragraph("财务审计报销明细", ParagraphStyle("Title", fontName=font, fontSize=18, leading=24)),
                   Spacer(1, 12), Paragraph(f"共 {len(rows)} 笔；含税合计 {money(sum(row.total_cents for row in rows))} 元", style), Spacer(1, 12)]
        lines = [[Paragraph(escape(str(value)), style) for value in line] for line in [["报销单号", "说明", "状态", "金额（元）"]] + data]
        table = Table(lines, colWidths=[132, 175, 105, 105], repeatRows=1)
        table.setStyle(TableStyle([("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#e4eee9")),
                                   ("GRID", (0, 0), (-1, -1), 0.4, colors.HexColor("#c5d2d0")), ("VALIGN", (0, 0), (-1, -1), "TOP")]))
        content.append(table)
        doc.build(content)
        mime = "application/pdf"
    return Response(output.getvalue(), media_type=mime, headers={"Content-Disposition": f'attachment; filename="audit-report.{format}"'})