"""Admin 后台 FastAPI app(独立进程)。 启动:uvicorn app.admin.main:admin_app --host 127.0.0.1 --port 8771 复用 App 的 DB/models/repositories/integrations;鉴权独立(admin JWT,见 app/admin/security.py)。 现有 app.main:app 不 import 本模块,两进程互不影响。 """ from __future__ import annotations import logging from collections.abc import AsyncIterator from contextlib import asynccontextmanager from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.admin.routers.ad_audit import router as ad_audit_router from app.admin.routers.ad_revenue import router as ad_revenue_router from app.admin.routers.admins import router as admins_router from app.admin.routers.audit import router as audit_router from app.admin.routers.auth import router as auth_router from app.admin.routers.config import router as config_router from app.admin.routers.dashboard import router as dashboard_router from app.admin.routers.ops_stat_config import router as ops_stat_config_router from app.admin.routers.feedback import router as feedback_router from app.admin.routers.onboarding import router as onboarding_router from app.admin.routers.ops_marquee_seed import router as ops_marquee_seed_router from app.admin.routers.price_report import router as price_report_router from app.admin.routers.users import router as users_router from app.admin.routers.wallet import router as wallet_router from app.admin.routers.withdraw import router as withdraw_router from app.core.config import settings from app.core.logging import setup_logging setup_logging(debug=settings.APP_DEBUG) logger = logging.getLogger("shagua.admin") @asynccontextmanager async def lifespan(_: FastAPI) -> AsyncIterator[None]: logger.info( "admin app started env=%s db=%s", settings.APP_ENV, settings.DATABASE_URL.split("://", 1)[0], ) yield logger.info("admin app shutting down") admin_app = FastAPI( title=f"{settings.APP_NAME} · Admin", version="0.1.0", docs_url="/admin/docs" if not settings.is_prod else None, redoc_url=None, lifespan=lifespan, ) # admin 前端独立部署。生产同域(nginx)无需 CORS;本地 next dev 跨域需放行开发源。 _dev_origins = [ "http://localhost:3001", "http://127.0.0.1:3001", "http://localhost:3000", "http://127.0.0.1:3000", ] _origins = settings.cors_origins_list or ([] if settings.is_prod else _dev_origins) if _origins: admin_app.add_middleware( CORSMiddleware, allow_origins=_origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @admin_app.get("/admin/api/health", tags=["meta"]) def health() -> dict[str, str]: return {"status": "ok", "service": "admin"} admin_app.include_router(auth_router) admin_app.include_router(dashboard_router) admin_app.include_router(ops_stat_config_router) admin_app.include_router(ops_marquee_seed_router) admin_app.include_router(users_router) admin_app.include_router(onboarding_router) admin_app.include_router(wallet_router) admin_app.include_router(withdraw_router) admin_app.include_router(price_report_router) admin_app.include_router(feedback_router) admin_app.include_router(admins_router) admin_app.include_router(audit_router) admin_app.include_router(config_router) admin_app.include_router(ad_audit_router) admin_app.include_router(ad_revenue_router)