src/web/app.py

"""FastAPI application entrypoint for the web UI."""

import hashlib
import hmac
import logging
import secrets
import sys
from pathlib import Path
from typing import Optional

from fastapi import FastAPI, Form, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates

from ..config.settings import get_settings
from .routes import api_router
from .routes.websocket import router as ws_router
from .task_manager import task_manager

logger = logging.getLogger(__name__)

if getattr(sys, "frozen", False):
    resource_root = Path(sys._MEIPASS)
else:
    resource_root = Path(__file__).parent.parent.parent

STATIC_DIR = resource_root / "static"
TEMPLATES_DIR = resource_root / "templates"


def _build_static_asset_version(static_dir: Path) -> str:
    latest_mtime = 0
    if static_dir.exists():
        for path in static_dir.rglob("*"):
            if path.is_file():
                latest_mtime = max(latest_mtime, int(path.stat().st_mtime))
    return str(latest_mtime or 1)


def create_app() -> FastAPI:
    settings = get_settings()

    app = FastAPI(
        title=settings.app_name,
        version=settings.app_version,
        description="OpenAI/Codex CLI 自动注册系统 Web UI",
        docs_url="/api/docs" if settings.debug else None,
        redoc_url="/api/redoc" if settings.debug else None,
    )

    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    if not STATIC_DIR.exists():
        STATIC_DIR.mkdir(parents=True, exist_ok=True)
        logger.info("Created static directory: %s", STATIC_DIR)
    app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")

    if not TEMPLATES_DIR.exists():
        TEMPLATES_DIR.mkdir(parents=True, exist_ok=True)
        logger.info("Created templates directory: %s", TEMPLATES_DIR)

    app.include_router(api_router, prefix="/api")
    app.include_router(ws_router, prefix="/api")

    templates = Jinja2Templates(directory=str(TEMPLATES_DIR))
    templates.env.globals["static_version"] = _build_static_asset_version(STATIC_DIR)

    def _auth_token(password: str) -> str:
        secret = get_settings().webui_secret_key.get_secret_value().encode("utf-8")
        return hmac.new(secret, password.encode("utf-8"), hashlib.sha256).hexdigest()

    def _is_authenticated(request: Request) -> bool:
        cookie = request.cookies.get("webui_auth")
        expected = _auth_token(get_settings().webui_access_password.get_secret_value())
        return bool(cookie) and secrets.compare_digest(cookie, expected)

    def _redirect_to_login(request: Request) -> RedirectResponse:
        return RedirectResponse(url=f"/login?next={request.url.path}", status_code=302)

    @app.get("/login", response_class=HTMLResponse)
    async def login_page(request: Request, next: Optional[str] = "/"):
        return templates.TemplateResponse(
            request,
            "login.html",
            {"request": request, "error": "", "next": next or "/"},
        )

    @app.post("/login")
    async def login_submit(request: Request, password: str = Form(...), next: Optional[str] = "/"):
        expected = get_settings().webui_access_password.get_secret_value()
        if not secrets.compare_digest(password, expected):
            return templates.TemplateResponse(
                request,
                "login.html",
                {"request": request, "error": "密码错误", "next": next or "/"},
                status_code=401,
            )

        response = RedirectResponse(url=next or "/", status_code=302)
        response.set_cookie("webui_auth", _auth_token(expected), httponly=True, samesite="lax")
        return response

    @app.get("/logout")
    async def logout(request: Request, next: Optional[str] = "/login"):
        response = RedirectResponse(url=next or "/login", status_code=302)
        response.delete_cookie("webui_auth")
        return response

    @app.get("/", response_class=HTMLResponse)
    async def index(request: Request):
        if not _is_authenticated(request):
            return _redirect_to_login(request)
        return templates.TemplateResponse(request, "index.html", {"request": request})

    @app.get("/accounts", response_class=HTMLResponse)
    async def accounts_page(request: Request):
        if not _is_authenticated(request):
            return _redirect_to_login(request)
        return templates.TemplateResponse(request, "accounts.html", {"request": request})

    @app.get("/email-services", response_class=HTMLResponse)
    async def email_services_page(request: Request):
        if not _is_authenticated(request):
            return _redirect_to_login(request)
        return templates.TemplateResponse(request, "email_services.html", {"request": request})

    @app.get("/settings", response_class=HTMLResponse)
    async def settings_page(request: Request):
        if not _is_authenticated(request):
            return _redirect_to_login(request)
        return templates.TemplateResponse(request, "settings.html", {"request": request})

    @app.get("/payment", response_class=HTMLResponse)
    async def payment_page(request: Request):
        return templates.TemplateResponse(request, "payment.html", {"request": request})

    @app.on_event("startup")
    async def startup_event():
        import asyncio
        from ..database.init_db import initialize_database

        try:
            initialize_database()
        except Exception as exc:
            logger.warning("Database initialization during startup raised: %s", exc)

        task_manager.set_loop(asyncio.get_running_loop())

        logger.info("=" * 50)
        logger.info("%s v%s starting", settings.app_name, settings.app_version)
        logger.info("Debug mode: %s", settings.debug)
        logger.info("Configured database URL: %s", settings.database_url)
        logger.info("=" * 50)

    @app.on_event("shutdown")
    async def shutdown_event():
        logger.info("Application shutdown")

    return app


app = create_app()