| """Main application entry point""" |
| import uvicorn |
| from fastapi import FastAPI |
| from fastapi.responses import FileResponse, HTMLResponse |
| from fastapi.staticfiles import StaticFiles |
| from fastapi.middleware.cors import CORSMiddleware |
| from pathlib import Path |
|
|
| |
| from .core.config import config |
| from .core.database import Database |
| from .services.token_manager import TokenManager |
| from .services.proxy_manager import ProxyManager |
| from .services.load_balancer import LoadBalancer |
| from .services.sora_client import SoraClient |
| from .services.generation_handler import GenerationHandler |
| from .api import routes as api_routes |
| from .api import admin as admin_routes |
|
|
| |
| app = FastAPI( |
| title="Sora2API", |
| description="OpenAI compatible API for Sora", |
| version="1.0.0" |
| ) |
|
|
| |
| app.add_middleware( |
| CORSMiddleware, |
| allow_origins=["*"], |
| allow_credentials=True, |
| allow_methods=["*"], |
| allow_headers=["*"], |
| ) |
|
|
| |
| db = Database() |
| token_manager = TokenManager(db) |
| proxy_manager = ProxyManager(db) |
| load_balancer = LoadBalancer(token_manager) |
| sora_client = SoraClient(proxy_manager) |
| generation_handler = GenerationHandler(sora_client, token_manager, load_balancer, db, proxy_manager) |
|
|
| |
| api_routes.set_generation_handler(generation_handler) |
| admin_routes.set_dependencies(token_manager, proxy_manager, db, generation_handler) |
|
|
| |
| app.include_router(api_routes.router) |
| app.include_router(admin_routes.router) |
|
|
| |
| static_dir = Path(__file__).parent.parent / "static" |
| static_dir.mkdir(exist_ok=True) |
| app.mount("/static", StaticFiles(directory=str(static_dir)), name="static") |
|
|
| |
| tmp_dir = Path(__file__).parent.parent / "tmp" |
| tmp_dir.mkdir(exist_ok=True) |
| app.mount("/tmp", StaticFiles(directory=str(tmp_dir)), name="tmp") |
|
|
| |
| @app.get("/", response_class=HTMLResponse) |
| async def root(): |
| """Redirect to login page""" |
| return """ |
| <!DOCTYPE html> |
| <html> |
| <head> |
| <meta http-equiv="refresh" content="0; url=/login"> |
| </head> |
| <body> |
| <p>Redirecting to login...</p> |
| </body> |
| </html> |
| """ |
|
|
| @app.get("/login", response_class=FileResponse) |
| async def login_page(): |
| """Serve login page""" |
| return FileResponse(str(static_dir / "login.html")) |
|
|
| @app.get("/manage", response_class=FileResponse) |
| async def manage_page(): |
| """Serve management page""" |
| return FileResponse(str(static_dir / "manage.html")) |
|
|
| @app.on_event("startup") |
| async def startup_event(): |
| """Initialize database on startup""" |
| |
| config_dict = config.get_raw_config() |
|
|
| |
| is_first_startup = not db.db_exists() |
|
|
| |
| await db.init_db() |
|
|
| |
| if is_first_startup: |
| print("🎉 First startup detected. Initializing database and configuration from setting.toml...") |
| await db.init_config_from_toml(config_dict, is_first_startup=True) |
| print("✓ Database and configuration initialized successfully.") |
| else: |
| print("🔄 Existing database detected. Checking for missing tables and columns...") |
| await db.check_and_migrate_db(config_dict) |
| print("✓ Database migration check completed.") |
|
|
| |
| admin_config = await db.get_admin_config() |
| config.set_admin_username_from_db(admin_config.admin_username) |
| config.set_admin_password_from_db(admin_config.admin_password) |
|
|
| |
| cache_config = await db.get_cache_config() |
| config.set_cache_enabled(cache_config.cache_enabled) |
| config.set_cache_timeout(cache_config.cache_timeout) |
| config.set_cache_base_url(cache_config.cache_base_url or "") |
|
|
| |
| generation_config = await db.get_generation_config() |
| config.set_image_timeout(generation_config.image_timeout) |
| config.set_video_timeout(generation_config.video_timeout) |
|
|
| |
| token_refresh_config = await db.get_token_refresh_config() |
| config.set_at_auto_refresh_enabled(token_refresh_config.at_auto_refresh_enabled) |
|
|
| |
| await generation_handler.file_cache.start_cleanup_task() |
|
|
| @app.on_event("shutdown") |
| async def shutdown_event(): |
| """Cleanup on shutdown""" |
| await generation_handler.file_cache.stop_cleanup_task() |
|
|
| if __name__ == "__main__": |
| uvicorn.run( |
| "src.main:app", |
| host=config.server_host, |
| port=config.server_port, |
| reload=False |
| ) |
|
|
