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, 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".*?", re.IGNORECASE | re.DOTALL) ICON_LINK_RE = re.compile(r"]*\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 "ChatGPT 号池管理"), 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'' f'' ) next_html = TITLE_RE.sub(f"{title}", html_text, count=1) if next_html == html_text and "" in next_html: next_html = next_html.replace("", f"{title}", 1) next_html = ICON_LINK_RE.sub("", next_html) if "" in next_html: next_html = next_html.replace("", f"{icon_markup}", 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="chatgpt2api", 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(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