| 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(), |
| }) |
| |
| |
| 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): |
| |
| |
| |
| 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 |
|
|
