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

"""应用工厂集中管理资源启停、跨域、限流和异常响应，客户端只处理一种契约。"""
import asyncio
from collections import defaultdict, deque
from contextlib import asynccontextmanager
import logging
import time
import uuid
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from sqlalchemy import text
from sqlalchemy.exc import IntegrityError
from app.api import management, business, operations
from app.application.services import request_context
from app.application.worker import worker
from app.infrastructure.ai import ai
from app.infrastructure.database import engine, initialize, Session
from app.shared.common import BusinessError, envelope, iso
from app.shared.config import settings

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")


@asynccontextmanager
async def lifespan(app):
    """消费者启动在建表与种子之后，关闭时先收尾任务再释放 HTTP/数据库连接。"""
    if ai.http.is_closed:
        import httpx
        ai.http = httpx.AsyncClient(timeout=settings.ai_timeout, follow_redirects=False)
    await initialize()
    await worker.start()
    yield
    await worker.close()
    await ai.close()
    await engine.dispose()


def create_app():
    app = FastAPI(title="AI-Agent 财务审计系统", version="1.0.0", lifespan=lifespan,
                  description="按课程设计：四节点审核、独立反馈、持久事件与人工审批。")
    app.add_middleware(CORSMiddleware, allow_origins=settings.cors_origins,
        allow_methods=["GET", "POST", "PATCH", "DELETE", "OPTIONS"], allow_headers=["Authorization", "Content-Type", "Idempotency-Key", "Last-Event-ID"],
        expose_headers=["X-Request-ID", "Content-Disposition"])
    limits = defaultdict(deque)

    @app.middleware("http")
    async def trace(request: Request, call_next):
        """请求 ID 由服务端产生，防止恶意用户污染日志；每分钟计数不会无限增长。"""
        request.state.request_id = uuid.uuid4().hex
        token = request_context.set(request.state.request_id)
        address = request.client.host if request.client else "local"
        now = time.monotonic()
        bucket = limits[address]
        while bucket and bucket[0] < now - 60:
            bucket.popleft()
        if len(limits) > 10000:
            for key in list(limits):
                if not limits[key] or limits[key][-1] < now - 60:
                    del limits[key]
        try:
            if request.method != "OPTIONS" and len(bucket) >= settings.rate_limit_per_minute:
                return JSONResponse({"error": {"code": "RATE_LIMIT", "message": "请求过于频繁，请稍后重试", "details": None}, "request_id": request.state.request_id}, status_code=429)
            if request.method != "OPTIONS":
                bucket.append(now)
            response = await call_next(request)
            response.headers["X-Request-ID"] = request.state.request_id
            response.headers["X-Content-Type-Options"] = "nosniff"
            response.headers["Referrer-Policy"] = "same-origin"
            logging.getLogger("audit.http").info("request_id=%s method=%s path=%s status=%s", request.state.request_id, request.method, request.url.path, response.status_code)
            return response
        finally:
            request_context.reset(token)

    @app.exception_handler(BusinessError)
    async def business_error(request, exc):
        """业务错误保留机器码，前端能区分版本冲突和登录失效。"""
        return JSONResponse({"error": {"code": exc.code, "message": str(exc), "details": exc.details}, "request_id": request.state.request_id}, status_code=exc.status)

    @app.exception_handler(RequestValidationError)
    async def validation_error(request, exc):
        """不返回错误请求的 input，尤其避免把密码、刷新 token 带入错误响应。"""
        errors = [{"field": ".".join(str(value) for value in item["loc"]), "message": item["msg"], "type": item["type"]} for item in exc.errors()]
        return JSONResponse({"error": {"code": "VALIDATION_ERROR", "message": "请求参数校验失败", "details": errors}, "request_id": request.state.request_id}, status_code=422)

    @app.exception_handler(IntegrityError)
    async def integrity_error(request, exc):
        """唯一/外键冲突不暴露 SQL 和数据库文件路径。"""
        return JSONResponse({"error": {"code": "DATA_CONFLICT", "message": "数据冲突或引用不存在", "details": None}, "request_id": request.state.request_id}, status_code=409)

    @app.exception_handler(Exception)
    async def internal_error(request, exc):
        logging.getLogger("audit.errors").exception("request_failed request_id=%s", request.state.request_id)
        return JSONResponse({"error": {"code": "INTERNAL_ERROR", "message": "服务处理失败，请凭请求ID联系管理员", "details": None}, "request_id": request.state.request_id}, status_code=500)

    app.include_router(management.router, prefix=settings.api_prefix)
    app.include_router(business.router, prefix=settings.api_prefix)
    app.include_router(operations.router, prefix=settings.api_prefix)

    @app.get(settings.api_prefix + "/health", tags=["健康"])
    async def health(request: Request):
        """功能：报告实际数据库和消费者状态。参数：无。返回：组件状态，无凭据。异常：503数据库不可用。"""
        try:
            async with Session() as session:
                await session.execute(text("SELECT 1"))
        except Exception:
            raise BusinessError("数据库不可用", 503, "DATABASE_UNAVAILABLE")
        return envelope({"database": "ok", "storage": "local", "worker": "running" if worker.task and not worker.task.done() else "disabled",
            "worker_event": worker.current, "last_timeout_scan": iso(worker.last_scan),
            "llm": "configured" if settings.llm_base_url else "not_configured",
            "vlm": "configured" if settings.vlm_base_url or settings.llm_base_url else "not_configured",
            "embedding": "configured" if settings.embedding_base_url else "not_configured",
            "breakers": {key.rsplit("/", 1)[-1]: value.status() for key, value in ai.breakers.items()}}, request.state.request_id)
    return app