113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
"""
|
|
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
|
|
)
|
|
|