""" Web application entry point """ from fastapi import FastAPI, Request from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from fastapi.responses import RedirectResponse from starlette.middleware.sessions import SessionMiddleware from pathlib import Path from web.admin.routes import router as admin_router from shared.config import settings from web.utils.auth import start_session_cleanup_task import logging import secrets logger = logging.getLogger(__name__) app = FastAPI( title="TGLoader Admin Panel", description="Web interface for managing TGLoader Telegram bot", version="1.0.0", docs_url="/api/docs", redoc_url="/api/redoc", openapi_url="/api/openapi.json" ) # Check and generate SECRET_KEY if not settings.WEB_SECRET_KEY or settings.WEB_SECRET_KEY == "your-secret-key-change-in-production": logger.warning( "⚠️ WEB_SECRET_KEY not set or using default value!\n" "⚠️ This is unsafe for production!\n" "⚠️ Generate a random key and add it to .env file:\n" f" WEB_SECRET_KEY={secrets.token_urlsafe(32)}" ) # Generate temporary key for development (in production this should be an error) secret_key = secrets.token_urlsafe(32) logger.warning(f"⚠️ Using temporary key (DO NOT USE IN PRODUCTION!): {secret_key[:20]}...") else: secret_key = settings.WEB_SECRET_KEY # Session middleware app.add_middleware( SessionMiddleware, secret_key=secret_key, max_age=86400 * 7 # 7 days ) # Mount static files and templates static_dir = Path("web/admin/static") static_dir.mkdir(parents=True, exist_ok=True) app.mount("/static", StaticFiles(directory=str(static_dir)), name="static") templates_dir = Path("web/admin/templates") templates = Jinja2Templates(directory=str(templates_dir)) # Include routers app.include_router(admin_router, prefix="/admin", tags=["admin"]) @app.on_event("startup") async def startup_event(): """Start background tasks on application startup""" try: # Initialize database (create tables if they don't exist) from shared.database.session import init_db try: await init_db() logger.info("Database initialized for web application") except Exception as db_error: logger.error(f"Error initializing database: {db_error}", exc_info=True) # Don't interrupt startup, database might already be initialized start_session_cleanup_task() # Start cleanup of old OTP codes on startup try: from web.utils.otp import cleanup_expired_otp_codes from web.utils.database import get_db async for db in get_db(): await cleanup_expired_otp_codes(db) break except Exception as otp_cleanup_error: logger.warning(f"Failed to cleanup old OTP codes on startup: {otp_cleanup_error}") # Start background task for updating user information try: import asyncio from bot.utils.user_info_updater import update_users_without_info_periodically loop = asyncio.get_running_loop() user_info_task = loop.create_task(update_users_without_info_periodically()) logger.info("User information update task started") except Exception as e: logger.warning(f"Failed to start user information update task: {e}") except Exception as e: logger.error(f"Error starting background tasks: {e}", exc_info=True) @app.get("/") async def root(): """Redirect to admin panel""" return RedirectResponse(url="/admin/login") if __name__ == "__main__": import uvicorn from shared.config import settings uvicorn.run( app, host=settings.WEB_HOST, port=settings.WEB_PORT )