shanxia2.0/backend/app/main.py

"""FastAPI 应用入口（骨架工程）"""
import asyncio
from contextlib import asynccontextmanager
import logging
from datetime import datetime

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from sqlalchemy import text
from sqlalchemy.exc import SQLAlchemyError

from app.core.config import settings
from app.core.database import SessionLocal, engine
from app.core.api_request_log_middleware import ApiRequestLogMiddleware
from app.core.api_crypto_middleware import ApiCryptoMiddleware
from app.models import models  # noqa: F401
import app.models.bx_models  # noqa: F401
import app.models.db_models  # noqa: F401
import app.models.pl_models  # noqa: F401
import app.models.duty_models  # noqa: F401
from app.api import (
    public_router,
    menu_router,
    admin_router,
    workflow_router,
    leave_router,
    crm_router,
    monitor_router,
    bx_router,
    db_router,
    mini_auth_router,
)
from app.api.pl import pl_router
from app.api.pl.public_transport import router as public_pl_transport_router
from app.api.pl.public_transport_site import router as public_pl_transport_site_router
from app.api.pl.public_site import router as public_pl_site_router
from app.api.pl.public_io_order import (
    inbound_public_router as public_pl_inbound_router,
    outbound_public_router as public_pl_outbound_router,
)
from app.api.db.public_inventory import router as public_db_inventory_router
from app.api.db.public_vehicles import router as public_db_vehicles_router
from app.api.duty import router as duty_router
from app.api.public_duty import router as public_duty_router


@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理"""
    from app.core import login_crypto

    login_crypto.ensure_initialized()

    from app.tasks import init_scheduled_tasks

    try:
        init_scheduled_tasks(SessionLocal)
    except Exception as e:
        logging.warning("定时任务调度器启动失败: %s", e)

    keepalive_task = asyncio.create_task(_db_pool_keepalive_loop())

    yield

    keepalive_task.cancel()
    try:
        await keepalive_task
    except asyncio.CancelledError:
        pass

    from app.tasks import shutdown_scheduler

    try:
        shutdown_scheduler()
    except Exception:
        pass


async def _db_pool_keepalive_loop() -> None:
    """定期 ping MySQL，避免远程库/NAT 回收空闲 TCP 后首请求长时间卡住。"""
    interval = max(60, int(settings.DB_KEEPALIVE_INTERVAL_SEC))

    def _ping() -> None:
        db = SessionLocal()
        try:
            db.execute(text("SELECT 1"))
        except Exception:
            try:
                engine.dispose()
            except Exception:
                pass
        finally:
            db.close()

    while True:
        await asyncio.sleep(interval)
        try:
            await asyncio.to_thread(_ping)
        except Exception as exc:
            logging.debug("db pool keepalive failed: %s", exc)


app = FastAPI(
    title=settings.APP_NAME,
    description="山峡化工生产一体化门户管理端后端",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
    lifespan=lifespan,
)

_cors_kw: dict = {
    "allow_origins": settings.BACKEND_CORS_ORIGINS,
    "allow_credentials": True,
    "allow_methods": ["*"],
    "allow_headers": ["*"],
}
if settings.DEBUG:
    _cors_kw["allow_origin_regex"] = r"^https?://(localhost|127\.0\.0\.1)(:\d+)?$"
app.add_middleware(CORSMiddleware, **_cors_kw)
app.add_middleware(ApiRequestLogMiddleware)
app.add_middleware(ApiCryptoMiddleware)

logger = logging.getLogger("shanxia2.0")


def _error_json(status_code: int, code: int, message: str) -> JSONResponse:
    return JSONResponse(
        status_code=status_code,
        content={
            "code": code,
            "message": message,
            "data": None,
            "timestamp": int(datetime.now().timestamp() * 1000),
        },
    )


@app.exception_handler(SQLAlchemyError)
async def sqlalchemy_exception_handler(request: Request, exc: SQLAlchemyError):
    """数据库异常：返回 JSON 并保留 CORS 头，避免浏览器只报跨域错误"""
    logger.exception("Database error on %s %s", request.method, request.url.path)
    if settings.DEBUG:
        message = f"数据库不可用：{exc.orig if getattr(exc, 'orig', None) else exc}"
    else:
        message = "数据库不可用，请确认 MySQL 已启动且连接配置正确"
    return _error_json(503, 503, message)


@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception):
    """未捕获异常：统一 JSON 500，确保跨域场景下仍有 Access-Control-Allow-Origin"""
    logger.exception("Unhandled error on %s %s", request.method, request.url.path)
    if settings.DEBUG:
        message = str(exc) or "服务器内部错误"
    else:
        message = "服务器内部错误，请稍后重试"
    return _error_json(500, 500, message)


app.include_router(public_router, prefix=settings.API_V1_PREFIX)
app.include_router(mini_auth_router, prefix=settings.API_V1_PREFIX)
app.include_router(public_pl_transport_router, prefix=settings.API_V1_PREFIX)
app.include_router(public_pl_transport_site_router, prefix=settings.API_V1_PREFIX)
app.include_router(public_pl_site_router, prefix=settings.API_V1_PREFIX)
app.include_router(public_pl_inbound_router, prefix=settings.API_V1_PREFIX)
app.include_router(public_pl_outbound_router, prefix=settings.API_V1_PREFIX)
app.include_router(public_db_inventory_router, prefix=settings.API_V1_PREFIX)
app.include_router(public_db_vehicles_router, prefix=settings.API_V1_PREFIX)
app.include_router(public_duty_router, prefix=settings.API_V1_PREFIX)
app.include_router(menu_router, prefix=settings.API_V1_PREFIX)
app.include_router(admin_router, prefix=settings.API_V1_PREFIX)
app.include_router(workflow_router, prefix=settings.API_V1_PREFIX)
app.include_router(leave_router, prefix=settings.API_V1_PREFIX)
app.include_router(crm_router, prefix=settings.API_V1_PREFIX)
app.include_router(bx_router, prefix=settings.API_V1_PREFIX)
app.include_router(db_router, prefix=settings.API_V1_PREFIX)
app.include_router(pl_router, prefix=settings.API_V1_PREFIX)
app.include_router(duty_router, prefix=settings.API_V1_PREFIX)
app.include_router(monitor_router, prefix=settings.API_V1_PREFIX)


def _mask_secret(value: str) -> str:
    if not value:
        return "(empty)"
    if len(value) <= 6:
        return "*" * len(value)
    return f"{value[:3]}***{value[-3:]}"


_api_crypto_boot_msg = (
    "API_CRYPTO_BOOT "
    f"enabled={settings.API_ENCRYPT_ENABLED} "
    f"fail_open={settings.API_ENCRYPT_FAIL_OPEN} "
    f"fixed_secret={_mask_secret(settings.API_ENCRYPT_FIXED_SECRET)} "
    f"ts_skew_sec={settings.API_ENCRYPT_TS_SKEW_SEC} "
    f"nonce_ttl_sec={settings.API_ENCRYPT_NONCE_TTL_SEC} "
    f"exclude_count={len([x for x in (settings.API_ENCRYPT_EXCLUDE_PATHS or '').split(',') if x.strip()])}"
)
print(_api_crypto_boot_msg)
logger.info(_api_crypto_boot_msg)


@app.get("/")
def root():
    return {"name": settings.APP_NAME, "version": "1.0.0", "docs": "/docs"}


@app.get("/health")
def health_check():
    return {"status": "healthy"}