harness_zbt/backend/app/main.py

"""FastAPI 应用入口。

启动命令 ``uv run uvicorn app.main:app --reload`` 中：
- ``app.main`` 表示本文件（app/main.py）；
- 最后的 ``app`` 表示文件末尾创建的 FastAPI 实例。
"""

from collections.abc import Awaitable, Callable
from datetime import UTC, datetime

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from starlette.responses import Response

from app.api.v1.admin_auth import create_admin_router
from app.api.v1.admin_business import create_admin_business_router
from app.api.v1.agent import create_agent_router
from app.api.v1.enrollment import create_enrollment_router
from app.api.v1.h5_auth import create_h5_router
from app.api.v1.h5_products import create_h5_product_router
from app.api.v1.system import router as system_router
from app.core.config import Settings, get_settings
from app.core.errors import AppError
from app.core.identifiers import new_request_id
from app.core.passwords import PasswordService
from app.core.security import JwtService
from app.domains.agent.repository import AgentRepository
from app.domains.agent.runtime import (
    AgentRuntime,
    UnavailableAgentRuntime,
)
from app.domains.agent.service import AgentThreadService
from app.domains.agent.tools import build_agent_tool_registry
from app.domains.attribution.repository import (
    AttributionRepository,
    InMemoryAttributionRepository,
)
from app.domains.attribution.service import AttributionService
from app.domains.catalog.repository import CatalogRepository
from app.domains.catalog.service import ProductCatalogService
from app.domains.enrollment.repository import EnrollmentRepository
from app.domains.enrollment.service import EnrollmentService
from app.domains.identity.repository import (
    IdentityRepository,
)
from app.domains.identity.service import AdminAuthService, H5AuthService
from app.harness.kernel import AgentKernel
from app.harness.model_gateway import DeepSeekModelGateway
from app.infrastructure.mysql.agent_repositories import SqlAlchemyAgentRepository
from app.infrastructure.mysql.attribution_repositories import (
    SqlAlchemyAttributionRepository,
)
from app.infrastructure.mysql.enrollment_repositories import (
    SqlAlchemyEnrollmentRepository,
)
from app.infrastructure.mysql.repositories import (
    SqlAlchemyCatalogRepository,
    SqlAlchemyIdentityRepository,
)
from app.infrastructure.mysql.sessions import (
    DatabaseName,
    create_mysql_engine,
    create_session_factory,
)
from app.infrastructure.redis.agent_state import (
    AgentStateStore,
    InMemoryAgentStateStore,
    RedisAgentStateStore,
)


def create_app(
    *,
    settings: Settings | None = None,
    identity_repository: IdentityRepository | None = None,
    catalog_repository: CatalogRepository | None = None,
    agent_repository: AgentRepository | None = None,
    agent_runtime: AgentRuntime | None = None,
    enrollment_repository: EnrollmentRepository | None = None,
    attribution_repository: AttributionRepository | None = None,
    agent_state_store: AgentStateStore | None = None,
    clock: Callable[[], datetime] | None = None,
) -> FastAPI:
    """创建并组装一个 FastAPI 应用。

    参数后的 ``| None`` 表示该参数可以不传。正常启动时使用真实配置和
    MySQL 仓储；测试时可以传入假的仓储和时钟，因此测试不必连接真实数据库。
    开头的 ``*`` 表示后面的参数必须写出参数名，可避免调用时传错位置。
    """
    app_settings = settings or get_settings()

    # 只有未注入测试仓储时才创建真实数据库连接。
    core_engine = None
    agent_engine = None
    if identity_repository is None or catalog_repository is None or enrollment_repository is None:
        core_engine = create_mysql_engine(app_settings, DatabaseName.CORE)
        session_factory = create_session_factory(core_engine)

    # ``a or b``：优先使用调用方传入的对象，否则创建生产环境的默认对象。
    repository = identity_repository or SqlAlchemyIdentityRepository(session_factory)
    product_repository = catalog_repository or SqlAlchemyCatalogRepository(session_factory)
    if agent_repository is None:
        agent_engine = create_mysql_engine(app_settings, DatabaseName.AGENT)
        agent_session_factory = create_session_factory(agent_engine)
        runtime_repository: AgentRepository = SqlAlchemyAgentRepository(agent_session_factory)
    else:
        runtime_repository = agent_repository

    # 用可替换的 clock 获取时间，可以让与时间有关的测试保持稳定。
    current_time = clock or (lambda: datetime.now(UTC))

    # Service 负责业务规则，Repository 负责数据读写，Router 负责接收 HTTP 请求。
    jwt_service = JwtService(app_settings)
    h5_auth_service = H5AuthService(app_settings, repository, jwt_service)
    catalog_service = ProductCatalogService(product_repository, current_time)
    attribution_repository_adapter = attribution_repository or (
        InMemoryAttributionRepository()
        if app_settings.app_env == "test"
        else SqlAlchemyAttributionRepository(session_factory)
    )
    attribution_service = AttributionService(
        attribution_repository_adapter,
        current_time,
    )
    transactional_repository = enrollment_repository or SqlAlchemyEnrollmentRepository(
        session_factory
    )
    enrollment_service = EnrollmentService(
        transactional_repository,
        catalog_service,
        current_time,
        app_settings.mock_payment_callback_secret,
        attribution_service,
    )
    admin_auth_service = AdminAuthService(
        app_settings,
        repository,
        jwt_service,
        PasswordService(),
    )
    configured_agent_runtime = agent_runtime
    if configured_agent_runtime is None and app_settings.deepseek_api_key:
        configured_agent_runtime = AgentKernel(
            settings=app_settings,
            model_gateway=DeepSeekModelGateway(app_settings),
            tools=build_agent_tool_registry(
                catalog_service,
                enrollment_service,
                attribution_service,
            ),
        )
    runtime_state_store = agent_state_store or (
        InMemoryAgentStateStore(app_settings.agent_rate_limit)
        if app_settings.app_env == "test"
        else RedisAgentStateStore(app_settings)
    )
    thread_service = AgentThreadService(
        runtime_repository,
        current_time,
        configured_agent_runtime or UnavailableAgentRuntime(),
        runtime_state_store,
    )
    application = FastAPI(
        title="智保通第一阶段API",
        version="0.1.0",
    )
    application.state.identity_repository = repository
    application.state.catalog_repository = product_repository
    application.state.core_engine = core_engine
    application.state.agent_engine = agent_engine
    application.state.agent_state_store = runtime_state_store
    application.state.attribution_repository = attribution_repository_adapter
    application.router.add_event_handler("shutdown", runtime_state_store.close)

    # CORS 决定哪些前端地址可以在浏览器中调用本后端。
    application.add_middleware(
        CORSMiddleware,
        allow_origins=app_settings.app_cors_origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    @application.middleware("http")
    async def request_context(
        request: Request,
        call_next: Callable[[Request], Awaitable[Response]],
    ) -> Response:
        # 每个请求都有 request_id，方便把前端报错与后端日志对应起来。
        request.state.request_id = request.headers.get("X-Request-ID") or new_request_id()
        response = await call_next(request)
        response.headers["X-Request-ID"] = request.state.request_id
        return response

    @application.exception_handler(AppError)
    async def app_error_handler(request: Request, error: AppError) -> JSONResponse:
        """把业务层抛出的 AppError 统一转换成 JSON 错误响应。"""
        return JSONResponse(
            status_code=error.status_code,
            content={
                "success": False,
                "error": {
                    "code": error.code,
                    "message": error.message,
                    "details": error.details or {},
                    "retryable": error.retryable,
                },
                "meta": {
                    "request_id": getattr(request.state, "request_id", new_request_id()),
                },
            },
        )

    @application.exception_handler(RequestValidationError)
    async def validation_error_handler(
        request: Request,
        error: RequestValidationError,
    ) -> JSONResponse:
        """把框架参数校验错误转换为稳定包络，不回传原始敏感输入。"""
        fields = [
            {
                "field": ".".join(str(part) for part in item["loc"]),
                "message": item["msg"],
            }
            for item in error.errors()
        ]
        return JSONResponse(
            status_code=400,
            content={
                "success": False,
                "error": {
                    "code": "VALIDATION_ERROR",
                    "message": "请求参数无效",
                    "details": {"fields": fields},
                    "retryable": False,
                },
                "meta": {
                    "request_id": getattr(request.state, "request_id", new_request_id()),
                },
            },
        )

    # 注册路由；统一前缀使所有第一版接口都从 /api/v1 开始。
    application.include_router(system_router, prefix="/api/v1")
    application.include_router(
        create_h5_router(
            h5_auth_service,
            app_settings,
            attribution_service,
        ),
        prefix="/api/v1",
    )
    application.include_router(
        create_h5_product_router(catalog_service, h5_auth_service),
        prefix="/api/v1",
    )
    application.include_router(
        create_admin_router(admin_auth_service, app_settings),
        prefix="/api/v1",
    )
    application.include_router(
        create_agent_router(
            thread_service,
            h5_auth_service,
            admin_auth_service,
        ),
        prefix="/api/v1",
    )
    application.include_router(
        create_enrollment_router(
            enrollment_service,
            h5_auth_service,
            app_settings,
        ),
        prefix="/api/v1",
    )
    application.include_router(
        create_admin_business_router(
            enrollment_service,
            admin_auth_service,
            catalog_service,
            attribution_service,
        ),
        prefix="/api/v1",
    )
    return application


# Uvicorn 导入 ``app.main:app`` 时，会执行这里并得到可运行的 ASGI 应用。
app = create_app()