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