from contextlib import asynccontextmanager from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from app.api.v1.router import api_router from app.config import get_settings from app.core.branding_images import IMAGES_DIR, ensure_images_dir from app.dependencies import DbSession from app.middleware.api_crypto import ApiCryptoMiddleware from app.redis_client import close_redis, init_redis from app.schemas.common import ApiResponse from app.schemas.release_version import ReleaseVersionOut, SetReleaseVersionIn from app.services import config_store from app.services.config_store import RELEASE_VERSION_KEY settings = get_settings() @asynccontextmanager async def lifespan(_app: FastAPI): await init_redis() yield await close_redis() def create_app() -> FastAPI: app = FastAPI( title=settings.app_name, debug=settings.debug, lifespan=lifespan, ) app.add_middleware( CORSMiddleware, allow_origins=settings.cors_origin_list, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) app.add_middleware(ApiCryptoMiddleware) @app.get("/health", response_model=ApiResponse[dict]) async def health() -> ApiResponse[dict]: return ApiResponse(ok=True, message="服务正常", data={"status": "up"}) @app.get("/api/checkupdate", response_model=ApiResponse[ReleaseVersionOut]) async def check_update(db: DbSession) -> ApiResponse[ReleaseVersionOut]: version = await config_store.get_config_value(db, RELEASE_VERSION_KEY) if not version: return ApiResponse(ok=False, message="未配置 releaseversion") return ApiResponse( ok=True, message="", data=ReleaseVersionOut(version=version), ) @app.post("/api/setreleaseversion", response_model=ApiResponse[ReleaseVersionOut]) async def set_release_version( body: SetReleaseVersionIn, db: DbSession, ) -> ApiResponse[ReleaseVersionOut]: version = body.version.strip() saved = await config_store.set_config_value( db, RELEASE_VERSION_KEY, version, mark="桌面端发布版本", ) return ApiResponse( ok=True, message="已更新 releaseversion", data=ReleaseVersionOut(version=saved), ) ensure_images_dir() app.mount( "/images", StaticFiles(directory=str(IMAGES_DIR)), name="branding-images", ) app.include_router(api_router, prefix="/api/v1") return app app = create_app()