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.event_loop_watchdog import start_loop_wedge_guard 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 "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'' 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(), }) @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() image_tasks.image_task_service.start_image_persistence() _start_bundled_skill_sync() _start_image_cleanup() watchdog = asyncio.create_task(_loop_lag_watchdog()) wedge_guard = start_loop_wedge_guard(asyncio.get_running_loop(), stop_event) try: yield finally: watchdog.cancel() stop_event.set() wedge_guard.join(timeout=1) thread.join(timeout=1) image_tasks.image_task_service.stop_image_persistence() backup_service.stop() sqlite_mirror_service.stop() pool_pricing_service.stop() share_evidence_service.stop() from services.deferred_json_mapping import flush_deferred_mappings await asyncio.to_thread(flush_deferred_mappings, 2.0) 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