Spaces:
Runtime error
Runtime error
| """ | |
| FastAPI application entry point. | |
| DeepFake Detector API - Milestone 1: Hugging Face hosted dummy models. | |
| """ | |
| import time | |
| from datetime import datetime, timezone | |
| from contextlib import asynccontextmanager | |
| from typing import AsyncGenerator | |
| MAIN_IMPORT_T0 = time.perf_counter() | |
| print( | |
| f"{datetime.now(timezone.utc).isoformat()} | INFO | app.main | phase3 module_import_start", | |
| flush=True, | |
| ) | |
| from fastapi import FastAPI, Request | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import JSONResponse | |
| from app.api import routes_health, routes_models, routes_predict | |
| from app.core.config import settings | |
| from app.core.errors import DeepFakeDetectorError | |
| from app.core.logging import setup_logging, get_logger | |
| from app.services.model_registry import get_model_registry | |
| # Set up logging | |
| setup_logging() | |
| logger = get_logger(__name__) | |
| logger.info( | |
| "phase3 module_import_complete duration_seconds=%.3f", | |
| time.perf_counter() - MAIN_IMPORT_T0, | |
| ) | |
| async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: | |
| """ | |
| Application lifespan manager. | |
| Handles startup and shutdown events: | |
| - Startup: Load models from Hugging Face | |
| - Shutdown: Cleanup resources | |
| """ | |
| startup_t0 = time.perf_counter() | |
| logger.info("phase3 startup_lifespan_begin") | |
| # Startup | |
| logger.info("Starting DeepFake Detector API...") | |
| logger.info(f"Configuration: HF_FUSION_REPO_ID={settings.HF_FUSION_REPO_ID}") | |
| logger.info(f"Configuration: HF_CACHE_DIR={settings.HF_CACHE_DIR}") | |
| # Load models from Hugging Face | |
| try: | |
| model_load_t0 = time.perf_counter() | |
| registry = get_model_registry() | |
| await registry.load_from_fusion_repo(settings.HF_FUSION_REPO_ID) | |
| logger.info( | |
| "phase3 startup_model_load_duration_seconds=%.3f", | |
| time.perf_counter() - model_load_t0, | |
| ) | |
| logger.info("Models loaded successfully!") | |
| except Exception as e: | |
| logger.error(f"Failed to load models on startup: {e}") | |
| logger.warning("API will start but /ready will report not_ready until models are loaded") | |
| logger.info( | |
| "phase3 startup_lifespan_total_duration_seconds=%.3f", | |
| time.perf_counter() - startup_t0, | |
| ) | |
| yield # Application runs here | |
| # Shutdown | |
| logger.info("Shutting down DeepFake Detector API...") | |
| # Create FastAPI application | |
| app = FastAPI( | |
| title=settings.PROJECT_NAME, | |
| version=settings.VERSION, | |
| description=""" | |
| DeepFake Detector API - Analyze images to detect AI-generated content. | |
| ## Features | |
| - **Fusion prediction**: Combines multiple model predictions using majority vote | |
| - **Individual model prediction**: Run specific submodels directly | |
| - **Timing information**: Detailed performance metrics for each request | |
| ## Milestone 1 | |
| This is the initial milestone using dummy random models hosted on Hugging Face | |
| for testing the API infrastructure. | |
| """, | |
| lifespan=lifespan, | |
| debug=settings.ENABLE_DEBUG | |
| ) | |
| # Add CORS middleware | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=settings.cors_origins_list, | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| logger.info(f"CORS enabled for origins: {settings.cors_origins_list}") | |
| # Global exception handler for custom errors | |
| async def deepfake_error_handler(request: Request, exc: DeepFakeDetectorError): | |
| """Handle custom DeepFakeDetector exceptions.""" | |
| return JSONResponse( | |
| status_code=500, | |
| content={ | |
| "error": type(exc).__name__, | |
| "message": exc.message, | |
| "details": exc.details | |
| } | |
| ) | |
| # Include routers | |
| app.include_router(routes_health.router) | |
| app.include_router(routes_models.router) | |
| app.include_router(routes_predict.router) | |
| # Root endpoint | |
| async def root(): | |
| """Root endpoint with API information.""" | |
| return { | |
| "name": settings.PROJECT_NAME, | |
| "version": settings.VERSION, | |
| "docs": "/docs", | |
| "health": "/health", | |
| "ready": "/ready" | |
| } | |
| if __name__ == "__main__": | |
| import uvicorn | |
| uvicorn.run( | |
| "app.main:app", | |
| host=settings.HOST, | |
| port=settings.PORT, | |
| reload=settings.ENABLE_DEBUG | |
| ) | |