shanxia2.0/backend/app/main.py

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

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.core.config import settings
from app.core.database import SessionLocal
from app.core.api_request_log_middleware import ApiRequestLogMiddleware
from app.core.api_crypto_middleware import ApiCryptoMiddleware
from app.models import models  # noqa: F401
from app.api import public_router, menu_router, admin_router, workflow_router, basic_info_router, collection_protocol_router, leave_router, operations_router, ethnic_router, sample_reception_router, qc_router, research_donation_router, production_router, pilot_service_router, insurance_company_router, material_router, storage_release_review_router, sample_sms_notification_router, storage_entry_sms_notification_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)

    yield

    from app.tasks import shutdown_scheduler

    try:
        shutdown_scheduler()
    except Exception:
        pass


app = FastAPI(
    title=settings.APP_NAME,
    description="zkxw 管理端骨架后端（鉴权与菜单）",
    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)

app.include_router(public_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(basic_info_router, prefix=settings.API_V1_PREFIX)
app.include_router(collection_protocol_router, prefix=settings.API_V1_PREFIX)
app.include_router(leave_router, prefix=settings.API_V1_PREFIX)
app.include_router(operations_router, prefix=settings.API_V1_PREFIX)
app.include_router(ethnic_router, prefix=settings.API_V1_PREFIX)
app.include_router(sample_reception_router, prefix=settings.API_V1_PREFIX)
app.include_router(qc_router, prefix=settings.API_V1_PREFIX)
app.include_router(research_donation_router, prefix=settings.API_V1_PREFIX)
app.include_router(production_router, prefix=settings.API_V1_PREFIX)
app.include_router(pilot_service_router, prefix=settings.API_V1_PREFIX)
app.include_router(insurance_company_router, prefix=settings.API_V1_PREFIX)
app.include_router(material_router, prefix=settings.API_V1_PREFIX)
app.include_router(storage_release_review_router, prefix=settings.API_V1_PREFIX)
app.include_router(sample_sms_notification_router, prefix=settings.API_V1_PREFIX)
app.include_router(storage_entry_sms_notification_router, prefix=settings.API_V1_PREFIX)

logger = logging.getLogger("sqxp")


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"}