from __future__ import annotations import json import math from dataclasses import asdict from typing import Any from fastapi import APIRouter, Request from fastapi.responses import HTMLResponse, JSONResponse from app.core.request_id import get_or_create_request_id, set_request_id_header, set_request_id_on_state from app.core.time import iso_utc_ms from app.db.session import with_sqlite_busy_retry router = APIRouter() def _clamp_int(value: Any) -> int: try: return int(value or 0) except Exception: return 0 async def _query_gallery_stats(engine) -> dict[str, Any]: async def _op() -> dict[str, Any]: async with engine.connect() as conn: images_total = int((await conn.exec_driver_sql("SELECT COUNT(*) FROM images;")).scalar_one()) images_enabled = int((await conn.exec_driver_sql("SELECT COUNT(*) FROM images WHERE status=1;")).scalar_one()) illust_total = int((await conn.exec_driver_sql("SELECT COUNT(DISTINCT illust_id) FROM images;")).scalar_one()) authors_total = int( ( await conn.exec_driver_sql( "SELECT COUNT(DISTINCT user_id) FROM images WHERE user_id IS NOT NULL;" ) ).scalar_one() ) r18_count = int((await conn.exec_driver_sql("SELECT COUNT(*) FROM images WHERE status=1 AND x_restrict=1;")).scalar_one()) safe_count = int((await conn.exec_driver_sql("SELECT COUNT(*) FROM images WHERE status=1 AND x_restrict=0;")).scalar_one()) r18_unknown = int( (await conn.exec_driver_sql("SELECT COUNT(*) FROM images WHERE status=1 AND x_restrict IS NULL;")).scalar_one() ) ai_count = int((await conn.exec_driver_sql("SELECT COUNT(*) FROM images WHERE status=1 AND ai_type=1;")).scalar_one()) non_ai_count = int((await conn.exec_driver_sql("SELECT COUNT(*) FROM images WHERE status=1 AND ai_type=0;")).scalar_one()) ai_unknown = int( (await conn.exec_driver_sql("SELECT COUNT(*) FROM images WHERE status=1 AND ai_type IS NULL;")).scalar_one() ) return { "gallery": { "images_total": images_total, "images_enabled": images_enabled, "illust_total": illust_total, "authors_total": authors_total, }, "breakdown": { "r18": {"r18": r18_count, "safe": safe_count, "unknown": r18_unknown}, "ai": {"ai": ai_count, "non_ai": non_ai_count, "unknown": ai_unknown}, }, } return await with_sqlite_busy_retry(_op) def _build_status_html(*, base_url: str, status_code: int, payload: dict[str, Any]) -> str: base = (base_url or "").rstrip("/") if not base: base = "" def u(path: str) -> str: path_norm = (path or "").strip() if not path_norm.startswith("/"): path_norm = "/" + path_norm return f"{base}{path_norm}" api_status = str(payload.get("api_status") or "unknown") api_status_code = int(payload.get("api_status_code") or status_code) updated_at = str(payload.get("updated_at") or "") gallery = payload.get("gallery") if isinstance(payload.get("gallery"), dict) else {} images_total = _clamp_int(gallery.get("images_total")) images_enabled = _clamp_int(gallery.get("images_enabled")) illust_total = _clamp_int(gallery.get("illust_total")) authors_total = _clamp_int(gallery.get("authors_total")) random_stats = payload.get("random") if isinstance(payload.get("random"), dict) else {} random_total = _clamp_int(random_stats.get("total_requests")) random_in_flight = _clamp_int(random_stats.get("in_flight")) last_window_requests = _clamp_int(random_stats.get("last_window_requests")) last_window_success_rate = float(random_stats.get("last_window_success_rate") or 0.0) if not math.isfinite(last_window_success_rate): last_window_success_rate = 0.0 last_window_success_rate = max(0.0, min(float(last_window_success_rate), 1.0)) json_url = u("/status.json") docs_url = u("/docs") random_url = u("/random") wtf_url = u("/wtf") return f""" Status · Random Mage

Status · Random Mage {api_status.upper()} {api_status_code}

最后更新：{updated_at} · JSON：{json_url}
/random 随机出图 /docs 使用文档 /wtf 瀑布流

图库概览

总图片数
{images_total}
images
总作品数
{illust_total}
DISTINCT illust_id
总作者数
{authors_total}
DISTINCT user_id
可用图片数
{images_enabled}
status=1

请求概览（/random）

总请求数
{random_total}
跨重启持久化
实时并发
{random_in_flight}
in_flight
近 60 秒请求
{last_window_requests}
window
近 60 秒成功率
{last_window_success_rate*100:.1f}%
2xx/3xx

占比图

选择维度：
说明：占比统计基于 status=1 的图片；未补全字段会落到 unknown。
/status 只读展示，不需要登录。
""" @router.get("/status.json", include_in_schema=False) async def status_json(request: Request) -> JSONResponse: rid = get_or_create_request_id(request) set_request_id_on_state(request, rid) engine = request.app.state.engine api_status = "ok" api_status_code = 200 payload: dict[str, Any] = {"api_status": api_status, "api_status_code": api_status_code, "updated_at": iso_utc_ms()} stats = getattr(request.app.state, "random_request_stats", None) if stats is not None: payload["random"] = asdict(await stats.snapshot()) else: payload["random"] = { "total_requests": 0, "total_ok": 0, "total_error": 0, "in_flight": 0, "window_seconds": 60, "last_window_requests": 0, "last_window_ok": 0, "last_window_error": 0, "last_window_success_rate": 0.0, } try: payload.update(await _query_gallery_stats(engine)) except Exception as exc: api_status = "degraded" api_status_code = 503 payload["api_status"] = api_status payload["api_status_code"] = api_status_code payload["error"] = {"type": type(exc).__name__, "message": str(exc)} resp = JSONResponse(status_code=int(api_status_code), content={"ok": api_status_code == 200, "data": payload, "request_id": rid}) set_request_id_header(resp, rid) return resp @router.get("/status", include_in_schema=False) async def status_page(request: Request) -> HTMLResponse: rid = get_or_create_request_id(request) set_request_id_on_state(request, rid) engine = request.app.state.engine api_status = "ok" api_status_code = 200 payload: dict[str, Any] = {"api_status": api_status, "api_status_code": api_status_code, "updated_at": iso_utc_ms()} stats = getattr(request.app.state, "random_request_stats", None) if stats is not None: payload["random"] = asdict(await stats.snapshot()) try: payload.update(await _query_gallery_stats(engine)) except Exception as exc: api_status = "degraded" api_status_code = 503 payload["api_status"] = api_status payload["api_status_code"] = api_status_code payload["error"] = {"type": type(exc).__name__, "message": str(exc)} html = _build_status_html( base_url=str(getattr(request, "base_url", "") or "").rstrip("/"), status_code=int(api_status_code), payload=payload, ) resp = HTMLResponse(content=html, status_code=int(api_status_code), headers={"Cache-Control": "no-store"}) set_request_id_header(resp, rid) return resp