api/app.py

from __future__ import annotations

import asyncio
from contextlib import asynccontextmanager
from html import escape
from mimetypes import guess_type
from pathlib import Path
import re
from threading import Event, Thread, active_count
import time

from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, HTMLResponse, Response
from starlette.middleware.gzip import GZipMiddleware

from api import accounts, ai, chat_room, checkin, image_tasks, openui, product_visual, register, shop, skills, system, third_party_images
from api.errors import install_exception_handlers
from api.support import resolve_web_asset, start_limited_account_watcher
from services.backup_service import backup_service
from services.config import config
from services.pool_pricing_service import pool_pricing_service
from services.share_evidence_service import share_evidence_service
from services.thread_pool import apply_thread_pool_tokens
from services.skill_service import skill_service
from services.sqlite_mirror_service import sqlite_mirror_service

STATIC_CACHE_CONTROL = "public, max-age=31536000, immutable"
ASSET_CACHE_CONTROL = "public, max-age=86400"
HTML_CACHE_CONTROL = "no-cache"
COMPRESSIBLE_SUFFIXES = {".css", ".html", ".js", ".json", ".map", ".svg", ".txt", ".xml"}
TITLE_RE = re.compile(r"<title>.*?</title>", re.IGNORECASE | re.DOTALL)
ICON_LINK_RE = re.compile(r"<link\b[^>]*\brel=[\"'][^\"']*(?:icon|apple-touch-icon)[^\"']*[\"'][^>]*>", re.IGNORECASE)


def _start_image_cleanup() -> Thread:
    def worker() -> None:
        started_at = time.monotonic()
        try:
            removed = config.cleanup_old_images()
            elapsed = time.monotonic() - started_at
            print(f"[image-cleanup] removed {removed} expired images in {elapsed:.2f}s")
        except Exception as exc:
            elapsed = time.monotonic() - started_at
            print(f"[image-cleanup] failed after {elapsed:.2f}s: {exc}")

    thread = Thread(target=worker, name="image-cleanup", daemon=True)
    thread.start()
    return thread


def _start_bundled_skill_sync() -> Thread:
    """Synchronize bundled Skills without blocking HTTP startup.

    On Hugging Face the Skills directory can live on a Storage Bucket mount.
    Copying a large bundle to that network filesystem may take minutes on the
    first deployment, so doing it before ``yield`` makes the whole Space appear
    stuck even though database recovery has already completed.
    """

    def worker() -> None:
        started_at = time.monotonic()
        print("[skills] bundled Skill synchronization started in background")
        try:
            bundled_skills = skill_service.sync_bundled_skills()
            elapsed = time.monotonic() - started_at
            print(
                f"[skills] synchronized {len(bundled_skills)} bundled Skills "
                f"in {elapsed:.2f}s"
            )
        except Exception as exc:
            elapsed = time.monotonic() - started_at
            print(f"[skills] bundled Skill synchronization failed after {elapsed:.2f}s: {exc}")

    thread = Thread(target=worker, name="bundled-skill-sync", daemon=True)
    thread.start()
    return thread


def _web_cache_headers(full_path: str, asset: Path) -> dict[str, str]:
    clean_path = full_path.strip("/")
    suffix = asset.suffix.lower()
    headers: dict[str, str] = {}
    if clean_path.startswith("_next/static/"):
        headers["Cache-Control"] = STATIC_CACHE_CONTROL
    elif suffix in {".html", ".txt"}:
        headers["Cache-Control"] = HTML_CACHE_CONTROL
    else:
        headers["Cache-Control"] = ASSET_CACHE_CONTROL
    return headers


def _select_web_asset(request: Request, full_path: str, asset: Path) -> tuple[Path, dict[str, str], str | None]:
    headers = _web_cache_headers(full_path, asset)
    media_type = guess_type(str(asset))[0]
    accept_encoding = str(request.headers.get("accept-encoding") or "").lower()
    gzip_asset = asset.with_name(f"{asset.name}.gz")
    is_compressible = asset.suffix.lower() in COMPRESSIBLE_SUFFIXES
    if is_compressible:
        headers = dict(headers)
        headers["Vary"] = "Accept-Encoding"
    if "gzip" in accept_encoding and is_compressible and gzip_asset.is_file():
        headers = dict(headers)
        headers["Content-Encoding"] = "gzip"
        return gzip_asset, headers, media_type
    return asset, headers, media_type


def _is_api_like_path(full_path: str) -> bool:
    clean_path = full_path.strip("/")
    return clean_path.startswith(("api/", "auth/", "v1/"))


def _guess_icon_type(href: str) -> str:
    clean_href = href.split("?", 1)[0].split("#", 1)[0].lower()
    if clean_href.endswith(".svg"):
        return "image/svg+xml"
    if clean_href.endswith((".jpg", ".jpeg")):
        return "image/jpeg"
    if clean_href.endswith(".webp"):
        return "image/webp"
    if clean_href.endswith(".ico"):
        return "image/x-icon"
    return "image/png"


def _inject_branding_head(html_text: str) -> str:
    branding = config.get_branding()
    title = escape(str(branding.get("site_title") or "ACM 号池管理"), quote=False)
    raw_icon_href = str(branding.get("site_logo_url") or "/favicon.ico")
    icon_href = escape(raw_icon_href, quote=True)
    icon_type = escape(_guess_icon_type(raw_icon_href), quote=True)
    icon_markup = (
        f'<link rel="icon" href="{icon_href}" type="{icon_type}" data-branding-icon="favicon">'
        f'<link rel="apple-touch-icon" href="{icon_href}" data-branding-icon="apple-touch-icon">'
    )
    next_html = TITLE_RE.sub(f"<title>{title}</title>", html_text, count=1)
    if next_html == html_text and "</head>" in next_html:
        next_html = next_html.replace("</head>", f"<title>{title}</title></head>", 1)
    next_html = ICON_LINK_RE.sub("", next_html)
    if "</head>" in next_html:
        next_html = next_html.replace("</head>", f"{icon_markup}</head>", 1)
    return next_html


def _web_response(full_path: str, request: Request, asset: Path) -> Response:
    if asset.suffix.lower() == ".html":
        html_text = asset.read_text(encoding="utf-8")
        return HTMLResponse(
            _inject_branding_head(html_text),
            headers=_web_cache_headers(full_path, asset),
        )
    selected_asset, headers, media_type = _select_web_asset(request, full_path, asset)
    return FileResponse(selected_asset, headers=headers, media_type=media_type)


def create_app() -> FastAPI:
    app_version = config.app_version

    async def _loop_lag_watchdog() -> None:
        """把事件循环的真实卡顿写进日志。

        ⚠️ 排查「整站无响应」时，容器日志一切正常（后台线程照跑）却测不出来到底卡在哪，
        就是因为没有任何关于事件循环本身的观测。这个探针每 20 秒量一次 sleep 漂移：
        漂移 = 循环线程被同步调用/GIL 抢占的实际时长。
        """
        from utils.log import logger as _logger

        while True:
            started = time.perf_counter()
            await asyncio.sleep(20)
            lag_ms = round((time.perf_counter() - started - 20) * 1000)
            if lag_ms >= 800:
                _logger.warning({
                    "event": "event_loop_lag",
                    "lag_ms": lag_ms,
                    "threads": active_count(),
                })

    def _start_loop_wedge_guard(loop: asyncio.AbstractEventLoop) -> None:
        """事件循环被彻底卡死时：抓全部线程堆栈 + 主动退出让 HF 重启容器。

        ⚠️ 为什么需要它：2026-10-19 线上出现「容器 RUNNING、后台线程照常打日志、
        所有 HTTP 请求零字节/连接被关闭」的状态，重启也不自愈。这种情况循环线程已经
        完全不调度，任何写在循环里的探针（包括 asyncio.sleep）都不会再执行，日志一片
        安静 —— 既测不出来也没法自愈。只能用一个独立线程从外部观测：
        往循环里投一个回调，健康时会立刻执行；15 秒都没跑 = 循环已死。
        连续 3 次判死 → dump 所有线程堆栈（定位阻塞点）→ os._exit 让容器重建。
        """
        import faulthandler
        import os
        import sys

        from utils.log import logger as _logger

        def run() -> None:
            strikes = 0
            while True:
                done = Event()
                started = time.time()
                try:
                    loop.call_soon_threadsafe(done.set)
                except RuntimeError:
                    return
                if done.wait(15.0):
                    strikes = 0
                    done.wait(5.0)
                    continue
                strikes += 1
                _logger.warning({
                    "event": "event_loop_unresponsive",
                    "waited_seconds": round(time.time() - started, 1),
                    "strikes": strikes,
                    "threads": active_count(),
                })
                # ⚠️ 只抓堆栈、**不主动退出**：os._exit 会让 HF 把 Space 置为
                # RUNTIME_ERROR（不会自动重建），反而要人工点重启；只留证据更好。
                if strikes == 3 or strikes % 12 == 0:
                    _logger.error({"event": "event_loop_wedged", "action": "dump_stacks"})
                    try:
                        faulthandler.dump_traceback(file=sys.stderr, all_threads=True)
                    except Exception:
                        pass
                    sys.stderr.flush()

        Thread(target=run, daemon=True, name="loop-wedge-guard").start()

    @asynccontextmanager
    async def lifespan(_: FastAPI):
        # anyio's worker-thread limiter defaults to 40 and is shared by every
        # run_in_threadpool() call in the process.  Resize it here, inside the
        # event loop, so the whole app sees the configured capacity.
        tokens = apply_thread_pool_tokens()
        print(f"[thread-pool] worker thread capacity set to {tokens}")
        stop_event = Event()
        thread = start_limited_account_watcher(stop_event)
        backup_service.start()
        sqlite_mirror_service.start()
        pool_pricing_service.start()
        share_evidence_service.start()
        _start_bundled_skill_sync()
        _start_image_cleanup()
        watchdog = asyncio.create_task(_loop_lag_watchdog())
        _start_loop_wedge_guard(asyncio.get_running_loop())
        try:
            yield
        finally:
            watchdog.cancel()
            stop_event.set()
            thread.join(timeout=1)
            backup_service.stop()
            sqlite_mirror_service.stop()
            pool_pricing_service.stop()
            share_evidence_service.stop()

    app = FastAPI(title="CHATSAM", version=app_version, lifespan=lifespan)
    install_exception_handlers(app)
    app.add_middleware(GZipMiddleware, minimum_size=1024, compresslevel=1)
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=False,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    app.include_router(ai.create_router())
    app.include_router(accounts.create_router())
    app.include_router(checkin.create_router())
    app.include_router(chat_room.create_router())
    app.include_router(image_tasks.create_router())
    app.include_router(openui.create_router())
    app.include_router(product_visual.create_router())
    app.include_router(register.create_router())
    app.include_router(shop.create_router())
    app.include_router(skills.create_router())
    app.include_router(third_party_images.create_router())
    app.include_router(system.create_router(app_version))

    @app.get("/{full_path:path}", include_in_schema=False)
    async def serve_web(full_path: str, request: Request):
        asset = resolve_web_asset(full_path)
        if asset is not None:
            return _web_response(full_path, request, asset)
        if full_path.strip("/").startswith("_next/"):
            raise HTTPException(status_code=404, detail="Not Found")
        fallback = resolve_web_asset("")
        if fallback is None:
            raise HTTPException(status_code=404, detail="Not Found")
        return _web_response("", request, fallback)

    @app.head("/{full_path:path}", include_in_schema=False)
    async def serve_web_head(full_path: str, request: Request):
        if _is_api_like_path(full_path):
            raise HTTPException(status_code=405, detail="Method Not Allowed")
        asset = resolve_web_asset(full_path)
        if asset is not None:
            selected_asset, headers, media_type = _select_web_asset(request, full_path, asset)
            return FileResponse(selected_asset, headers=headers, media_type=media_type)
        if full_path.strip("/").startswith("_next/"):
            raise HTTPException(status_code=404, detail="Not Found")
        fallback = resolve_web_asset("")
        if fallback is None:
            raise HTTPException(status_code=404, detail="Not Found")
        selected_asset, headers, media_type = _select_web_asset(request, "", fallback)
        return FileResponse(selected_asset, headers=headers, media_type=media_type)

    return app