api/system.py

from __future__ import annotations

import asyncio
import functools
import json
from concurrent.futures import ThreadPoolExecutor
from typing import Any
from urllib.parse import quote, urlparse

import httpx
from fastapi import APIRouter, Header, HTTPException, Query, Request
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import HTMLResponse, Response, StreamingResponse
from pydantic import BaseModel, ConfigDict, Field

from api.support import require_admin, require_identity, resolve_image_base_url
from api.image_inputs import _download_image_url
from services.backup_service import BackupError, backup_service
from services.config import config
from services.content_filter import preview_local_review
from services.credit_ledger_service import credit_ledger_service
from services.error_message_service import (
    ERROR_SUMMARY_SYSTEM_PROMPT,
    error_message_service,
    extract_keyword_candidates,
    parse_ai_suggestion,
)
from services.log_service import LoggedCall, log_service
from services.protocol import openai_v1_chat_complete
from services.auth_service import auth_service
from services.image_service import delete_images, download_images_zip, get_image_download_response, get_image_response, get_thumbnail_response, list_images
from services.image_storage_service import ImageStorageError, image_storage_service
from services.image_tags_service import delete_tag, get_all_tags, set_tags

# ⚠️⚠️ 图片读写必须走**独立线程池**，不能和 StaticFiles / 其他 run_in_threadpool 共用
# AnyIO 的默认线程池（本服务设成 50）。原因：图片存在**网络挂载**的 Storage Bucket 上，
# 单张 2.8MB 在桶上读一次可能要几十秒；几十人同时刷一屏图（每人 6~9 张）会把 50 个共享
# 线程全部占满，后面排队的 HTML、JS、/health 全部超时 —— 表现是「连接被关闭 / 一个字节
# 都不回」（2026-10-19 全站无响应事故，后台线程日志一切正常，就是 HTTP 不响应）。
# 独立池 = 图片再慢也只排队在自己这里，页面和接口照常可用。
_IMAGE_IO_POOL = ThreadPoolExecutor(max_workers=8, thread_name_prefix="image-io")
# 单张图片在桶上读不完时的上限：超时直接 503 + Retry-After，让浏览器/CF 稍后重试。
# ⚠️ 不能无限排队：几百个请求排在 8 个 worker 后面，会变成十几分钟的积压，
# 用户看到的就是「页面一直在转」，而 CF 也会一直保持连接。
_IMAGE_IO_TIMEOUT_SECONDS = 25.0


async def _run_image_io(func, *args):
    """在图片专用线程池里跑阻塞的图片读写，隔离出共享线程池。"""
    loop = asyncio.get_running_loop()
    future = loop.run_in_executor(_IMAGE_IO_POOL, functools.partial(func, *args))
    try:
        return await asyncio.wait_for(future, timeout=_IMAGE_IO_TIMEOUT_SECONDS)
    except asyncio.TimeoutError:
        raise HTTPException(
            status_code=503,
            detail={"error": "image io timeout"},
            headers={"Retry-After": "10"},
        )
from services.proxy_service import proxy_settings, test_clearance, test_proxy
from services.share_evidence_service import share_evidence_service
from services.pool_pricing_service import pool_pricing_service
from services.presence_service import presence_service, view_for_role
from services.third_party_image_service import third_party_image_service
from services.thread_pool import apply_thread_pool_tokens


def _validate_pool_pricing_groups(validation: dict[str, Any]) -> dict[str, Any]:
    valid_group_ids = {str(item.get("id") or "") for item in third_party_image_service.list_groups()}
    errors = list(validation.get("errors") or [])
    for rule in validation.get("config", {}).get("rules", []):
        if rule.get("target_type") != "third_party_group":
            continue
        group_id = str(rule.get("third_party_group_id") or "")
        if group_id and group_id not in valid_group_ids:
            errors.append({"rule_id": str(rule.get("id") or ""), "message": "所选第三方生图 API 分组不存在"})
    return {**validation, "errors": errors, "valid": not errors}


class SettingsUpdateRequest(BaseModel):
    model_config = ConfigDict(extra="allow")


class UnconfiguredErrorClearRequest(BaseModel):
    """清空「未配置报错」待办；不传 fingerprints 表示全部清空。"""

    fingerprints: list[str] = Field(default_factory=list)


class ErrorSummarizeRequest(BaseModel):
    """把一段报错原文交给模型归纳成「关键词 + 大白话」。"""

    text: str = Field(..., min_length=1, max_length=8000)


class ContentReviewPreviewRequest(BaseModel):
    text: str = ""
    sensitive_words: list[str] | None = None
    local_review: dict[str, object] | None = None


class PoolPricingPreviewRequest(BaseModel):
    at: str = ""
    quantity: int = 1
    config: dict[str, Any] | None = None
    target_type: str = "pool"
    third_party_group_id: str = ""
    base_unit_price: float = 1


class WechatLoginRequest(BaseModel):
    code: str = ""
    invite_code: str = ""


class NormalUserRegisterRequest(BaseModel):
    name: str = ""
    invite_code: str = ""


class ProxyTestRequest(BaseModel):
    url: str = ""


class ClearanceTestRequest(BaseModel):
    target_url: str = "https://chatgpt.com"


class UpstreamProbeRequest(BaseModel):
    access_token: str = ""
    routes: list[str] = []


class ImageDeleteRequest(BaseModel):
    paths: list[str] = []
    start_date: str = ""
    end_date: str = ""
    all_matching: bool = False

class ImageDownloadRequest(BaseModel):
    paths: list[str]

class ImageTagsRequest(BaseModel):
    path: str
    tags: list[str]

class LogDeleteRequest(BaseModel):
    ids: list[str] = []
class BackupDeleteRequest(BaseModel):
    key: str = ""


class UserStorageImportRequest(BaseModel):
    payload: Any
    dry_run: bool = True


def _parse_json_object(value: object) -> dict[str, Any]:
    if isinstance(value, dict):
        return dict(value)
    if isinstance(value, str):
        try:
            parsed = json.loads(value)
        except json.JSONDecodeError:
            return {}
        return dict(parsed) if isinstance(parsed, dict) else {}
    return {}


def _parse_json_list(value: object) -> list[dict[str, Any]]:
    if isinstance(value, list):
        return [dict(item) for item in value if isinstance(item, dict)]
    if isinstance(value, str):
        try:
            parsed = json.loads(value)
        except json.JSONDecodeError:
            return []
        return [dict(item) for item in parsed if isinstance(item, dict)] if isinstance(parsed, list) else []
    return []


def _normalize_auth_key_import(payload: dict[str, Any]) -> list[dict[str, Any]]:
    raw_auth_keys = payload.get("auth_keys")
    rows = raw_auth_keys if isinstance(raw_auth_keys, list) else []
    items: list[dict[str, Any]] = []
    for row in rows:
        if not isinstance(row, dict):
            continue
        item = _parse_json_object(row.get("data")) if "data" in row else dict(row)
        item_id = str(item.get("id") or row.get("key_id") or "").strip()
        if item_id:
            item["id"] = item_id
            items.append(item)
    return items


def _normalize_shop_state_import(payload: dict[str, Any]) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]:
    state: dict[str, Any] = {}
    named_states: dict[str, dict[str, Any]] = {}
    raw_shop_state = payload.get("shop_state")

    if isinstance(raw_shop_state, dict):
        state = dict(raw_shop_state)
    elif isinstance(raw_shop_state, list):
        for row in raw_shop_state:
            if not isinstance(row, dict):
                continue
            key = str(row.get("key") or "default").strip() or "default"
            item_state = _parse_json_object(row.get("data")) if "data" in row else dict(row)
            if key == "default":
                state = item_state
            else:
                named_states[key] = item_state

    imported_codes = _parse_json_list(payload.get("redeem_codes"))
    if imported_codes:
        existing_codes = state.get("codes") if isinstance(state.get("codes"), list) else []
        merged: dict[str, dict[str, Any]] = {}
        for item in [*existing_codes, *imported_codes]:
            if not isinstance(item, dict):
                continue
            data = _parse_json_object(item.get("data")) if "data" in item else dict(item)
            data.setdefault("id", item.get("code_id") or item.get("id"))
            data.setdefault("code_hash", item.get("code_hash"))
            data.setdefault("status", item.get("status"))
            if item.get("batch_id") and not data.get("batch_id"):
                data["batch_id"] = item.get("batch_id")
            if item.get("redeemed_by") and not data.get("redeemed_by"):
                data["redeemed_by"] = item.get("redeemed_by")
            identity = str(data.get("code_hash") or data.get("id") or "").strip()
            if identity:
                merged[identity] = data
        state["codes"] = list(merged.values())

    if not isinstance(state.get("ledger"), list):
        state["ledger"] = []
    if not isinstance(state.get("codes"), list):
        state["codes"] = []
    return state, named_states


def _normalize_user_storage_import(payload: object) -> tuple[list[dict[str, Any]], dict[str, Any], dict[str, dict[str, Any]]]:
    parsed = _parse_json_object(payload)
    if "full_backup" in parsed:
        parsed = _parse_json_object(parsed.get("full_backup"))
    auth_keys = _normalize_auth_key_import(parsed)
    shop_state, named_states = _normalize_shop_state_import(parsed)
    return auth_keys, shop_state, named_states


def create_router(app_version: str) -> APIRouter:
    router = APIRouter()

    def image_memory_button_enabled(identity: dict[str, object]) -> bool:
        return bool(
            config.image_memory_button_enabled
            and (
                identity.get("role") == "admin"
                or identity.get("image_memory_button_enabled", False)
            )
        )

    def login_response(identity: dict[str, object]) -> dict[str, object]:
        return {
            "ok": True,
            "version": app_version,
            "role": identity.get("role"),
            "subject_id": identity.get("id"),
            "name": identity.get("name"),
            "account_pool_enabled": bool(identity.get("role") == "admin" or identity.get("account_pool_enabled")),
            "image_memory_button_enabled": image_memory_button_enabled(identity),
            "image_share_evidence_enabled": config.image_share_evidence_enabled,
            "login_session_duration_hours": config.login_session_duration_hours_for_role(identity.get("role")),
        }

    @router.post("/auth/login")
    async def login(authorization: str | None = Header(default=None)):
        identity = require_identity(authorization)
        return login_response(identity)

    @router.post("/auth/session")
    async def refresh_session(authorization: str | None = Header(default=None)):
        identity = require_identity(authorization, touch=False)
        return login_response(identity)

    @router.post("/api/presence/ping")
    async def presence_ping(authorization: str | None = Header(default=None)):
        # touch=False：心跳每 25s 一次，不能污染密钥的「最后使用时间」（那是「最后调用」的语义）。
        identity = require_identity(authorization, touch=False)
        return presence_service.ping(identity)

    @router.post("/api/presence/leave")
    async def presence_leave(authorization: str | None = Header(default=None)):
        # 页面关闭（pagehide）时上报，避免「人已经走了，数字还挂满一个窗口」。
        identity = require_identity(authorization, touch=False)
        return presence_service.leave(identity)

    @router.get("/api/presence/online")
    async def presence_online(authorization: str | None = Header(default=None)):
        identity = require_identity(authorization, touch=False)
        return view_for_role(presence_service.snapshot(), identity.get("role"))

    @router.get("/api/dashboard/credits")
    async def dashboard_credits(
        range_key: str = Query(default="today", alias="range"),
        role: str = Query(default=""),
        model: str = Query(default=""),
        top: int = Query(default=10, ge=1, le=100),
        authorization: str | None = Header(default=None),
    ):
        """生图用量统计，仅管理员可见。

        口径说明（前端文案必须跟着这个来）：
          - 只有**普通用户**消耗积分。内部用户走模型每日额度、管理员免费，
            这两类角色在流水里 credits 恒为 0，只能用「调用次数」衡量。
          - `equiv_credits` 是按**普通用户同时段等效单价**把调用量折算出来的
            参考值，用来横向感受内部用量规模，**不是实际扣费**。
        """
        require_admin(authorization)
        normalized_role = str(role or "").strip().lower()
        if normalized_role not in {"normal", "user", "admin"}:
            normalized_role = ""
        data = credit_ledger_service.summary(
            range_key=range_key,
            role=normalized_role,
            model=str(model or "").strip(),
            top_limit=top,
        )
        # 补上昵称，否则管理员只能看到一串 id，排查「谁把额度用满了」无从下手。
        names: dict[str, str] = {}
        try:
            for item in auth_service.list_keys():
                key_id = str(item.get("id") or "").strip()
                if key_id:
                    names[key_id] = str(item.get("name") or "").strip()
        except Exception:
            names = {}
        for entry in data.get("top_users") or []:
            identity_id = str(entry.get("identity_id") or "")
            entry["name"] = names.get(identity_id) or identity_id[:8]
        return data

    @router.get("/api/auth/wechat/status")
    async def wechat_login_status():
        return {
            "enabled": config.wechat_login_enabled,
            "auto_register_enabled": config.wechat_auto_register_enabled,
            "login_session_duration_hours": config.normal_user_login_session_duration_hours,
        }

    @router.get("/api/branding")
    async def get_branding(response: Response):
        response.headers["Cache-Control"] = "no-store, max-age=0"
        return {"branding": config.get_branding()}

    @router.get("/api/navigation")
    async def get_navigation(authorization: str | None = Header(default=None)):
        identity = require_identity(authorization)
        role = str(identity.get("role") or "")
        if role == "normal":
            items = config.get_normal_user_navigation()
        elif role == "admin":
            items = config.get_admin_user_navigation()
        else:
            items = config.get_internal_user_navigation()
        return {"items": items}

    @router.post("/api/auth/wechat/login")
    async def wechat_login(body: WechatLoginRequest):
        if not config.wechat_login_enabled:
            raise HTTPException(status_code=403, detail={"error": "微信验证码登录已关闭，请使用登录密钥。"})
        code = str(body.code or "").strip()
        if not code:
            raise HTTPException(status_code=400, detail={"error": "请输入微信验证码"})

        try:
            async with httpx.AsyncClient(timeout=15.0) as client:
                response = await client.post("https://wx.z-l.top/api/auth/verify", json={"code": code})
        except httpx.HTTPError as exc:
            raise HTTPException(status_code=502, detail={"error": "微信登录服务暂时不可用，请稍后再试"}) from exc

        try:
            payload = response.json()
        except ValueError as exc:
            raise HTTPException(status_code=502, detail={"error": "微信登录服务返回异常"}) from exc

        if not response.is_success or not payload.get("success"):
            message = str(payload.get("message") or "微信验证码无效或已过期")
            raise HTTPException(status_code=401, detail={"error": message})

        user = payload.get("user") if isinstance(payload.get("user"), dict) else {}
        openid = str(user.get("openid") or "").strip()
        if not openid:
            raise HTTPException(status_code=502, detail={"error": "微信登录服务未返回 openid"})

        try:
            item, raw_key, created = auth_service.get_or_create_wechat_normal_user(
                openid=openid,
                nickname=str(user.get("nickname") or "").strip(),
                avatar=str(user.get("avatar") or "").strip(),
                allow_create=config.wechat_auto_register_enabled,
            )
            if str(body.invite_code or "").strip():
                auth_service.attach_invite(str(item.get("id") or ""), body.invite_code)
                item = auth_service.get_key(str(item.get("id") or ""), role="normal") or item
        except ValueError as exc:
            raise HTTPException(status_code=403, detail={"error": str(exc)}) from exc
        except Exception as exc:
            print(f"[wechat-login] failed to save user: {exc}")
            raise HTTPException(status_code=503, detail={"error": f"微信登录已验证，但保存用户失败：{exc}"}) from exc
        return {
            "ok": True,
            "version": app_version,
            "role": "normal",
            "subject_id": item.get("id"),
            "name": item.get("name"),
            "account_pool_enabled": bool(item.get("account_pool_enabled")),
            "image_memory_button_enabled": image_memory_button_enabled(item),
            "key": raw_key,
            "created": created,
            "login_session_duration_hours": config.normal_user_login_session_duration_hours,
            "credits": auth_service.credit_summary(item),
            "wechat": {
                "openid": openid,
                "nickname": item.get("wechat_nickname") or user.get("nickname") or "",
                "avatar": item.get("wechat_avatar") or user.get("avatar") or "",
            },
        }

    @router.get("/version")
    async def get_version():
        return {"version": app_version}

    @router.get("/api/settings")
    async def get_settings(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return {"config": config.get()}

    @router.get("/api/announcements")
    async def get_announcements(authorization: str | None = Header(default=None)):
        identity = require_identity(authorization)
        return {"items": config.get_announcements(role=str(identity.get("role") or ""))}

    @router.post("/api/settings")
    async def save_settings(body: SettingsUpdateRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            update = body.model_dump(mode="python")
            for pricing_key in ("normal_user_pool_pricing", "normal_user_third_party_image_pricing"):
                if pricing_key not in update:
                    continue
                validation = _validate_pool_pricing_groups(pool_pricing_service.validate(update.get(pricing_key)))
                if not validation["valid"]:
                    message = "；".join(item["message"] for item in validation["errors"][:3])
                    raise ValueError(message or "活动价格规则无效")
                update[pricing_key] = validation["config"]
            previous_valid_days = int(config.get().get("checkin_credit_valid_days") or 0)
            # 用归一后的值（范围 1-720、缺省 12）做对比，避免「存的是字符串/缺键」导致
            # 每次保存都误判成「改过」而重算一遍台账。
            previous_share_hours = config.image_share_retention_hours
            saved = config.update(update)
            if "thread_pool_tokens" in update:
                # The anyio limiter is a plain in-place object, so a resize takes
                # effect immediately for every queued and future request.
                tokens = apply_thread_pool_tokens(saved.get("thread_pool_tokens"))
                print(f"[thread-pool] worker thread capacity set to {tokens}")
            # 「签到积分可用天数」变了 → 立刻按新值重算**已发放**的签到积分形态。
            # 不改的话设置只对下一次签到生效，管理员会觉得「改了却没生效」。
            if "checkin_credit_valid_days" in update:
                next_valid_days = int(saved.get("checkin_credit_valid_days") or 0)
                if next_valid_days != previous_valid_days:
                    try:
                        await run_in_threadpool(auth_service.reconcile_checkin_credit_expiry, next_valid_days)
                    except Exception as exc:  # 重算失败不应让「保存设置」本身失败
                        print(f"[checkin] 签到积分有效期重算失败：{exc}")
            # 「链接保留时长」变了 → 立刻按新值重算**已登记**的官网分享链接到期时刻。
            # 不重算的话旧记录会继续按上一个时长活着，用户会觉得「设了 3 小时，链接过了
            # 3 小时还能打开」；重算后已过期的会马上被回收。
            if "image_share_retention_hours" in update:
                next_hours = config.image_share_retention_hours
                if next_hours != previous_share_hours:
                    try:
                        await run_in_threadpool(share_evidence_service.reconcile_retention, next_hours)
                    except Exception as exc:  # 同上：重算失败不该让保存设置失败
                        print(f"[share-evidence] 分享链接保留时长重算失败：{exc}")
            return {"config": saved}
        except ValueError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc

    @router.post("/api/settings/content-review/preview")
    async def preview_content_review(body: ContentReviewPreviewRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return preview_local_review(
            body.text,
            words=body.sensitive_words,
            settings=body.local_review,
        )

    @router.post("/api/settings/pool-pricing/preview")
    async def preview_pool_pricing(body: PoolPricingPreviewRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        pricing_config = body.config
        if pricing_config is None:
            pricing_config = (
                config.normal_user_third_party_image_pricing
                if body.target_type == "third_party_group"
                else config.normal_user_pool_pricing
            )
        validation = _validate_pool_pricing_groups(pool_pricing_service.validate(pricing_config))
        if not validation["valid"]:
            return {"validation": validation, "quote": None, "calendar": pool_pricing_service.calendar_status()}
        quote = pool_pricing_service.quote(
            {"role": "normal"},
            body.quantity,
            at=body.at,
            pricing_config=validation["config"],
            target_type=body.target_type,
            third_party_group_id=body.third_party_group_id,
            base_unit_price=body.base_unit_price,
        )
        return {"validation": validation, "quote": quote, "calendar": pool_pricing_service.calendar_status()}

    @router.post("/api/settings/pool-pricing/calendar/refresh")
    async def refresh_pool_pricing_calendar(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return {"calendar": await run_in_threadpool(pool_pricing_service.refresh_calendar)}

    @router.get("/api/images")
    async def get_images(
        request: Request,
        start_date: str = "",
        end_date: str = "",
        limit: int = 12,
        offset: int = 0,
        tags: str = "",
        authorization: str | None = Header(default=None),
    ):
        require_admin(authorization)
        # ⚠️ list_images 会读整份 image_index.json（线上已是百 MB 级）并扫目录，
        # 直接在事件循环上跑，一次调用就能让全站所有请求排队几十秒 —— 实测正是如此。
        return await run_in_threadpool(
            list_images,
            resolve_image_base_url(request),
            start_date=start_date.strip(),
            end_date=end_date.strip(),
            limit=limit,
            offset=offset,
            tags=[tag.strip() for tag in tags.split(",") if tag.strip()],
        )

    @router.get("/images/{image_path:path}", include_in_schema=False)
    async def get_image(image_path: str):
        # ⚠️⚠️ 必须进线程池，不要改回直接调用。
        # get_image_response 里既有同步 stat，本地未命中时还会把整张图（1~4MB）从**网络挂载**
        # 的 Storage Bucket 整个读进内存。在 async 路由里直接调 = 把这段阻塞 IO 放在事件循环
        # 线程上跑；而图片是全站最热的请求，几十人同时刷图会把事件循环彻底堵死 ⇒ 连 /health、
        # 静态资源和页面 SSR 都一起排队超时（2026-10-16 线上实测：/health 连续 20~25s 无响应，
        # 而同一 commit 的空闲容器 1s 返回 200）。
        return await _run_image_io(get_image_response, image_path)

    @router.get("/api/images/proxy")
    async def proxy_image(url: str, authorization: str | None = Header(default=None)):
        require_identity(authorization)
        parsed = urlparse(str(url or "").strip())
        if parsed.scheme not in {"http", "https"} or not parsed.netloc:
            raise HTTPException(status_code=400, detail={"error": "invalid image url"})
        data, _filename, content_type = await _run_image_io(_download_image_url, url)
        return Response(
            content=data,
            media_type=content_type,
            headers={"Cache-Control": "private, max-age=3600"},
        )

    @router.get("/image-thumbnails/{image_path:path}", include_in_schema=False)
    async def get_image_thumbnail(image_path: str):
        # ⚠️ 同上：ensure_thumbnail 会读原图 + 用 PIL 做 LANCZOS 重采样并 PNG 编码，
        # 是纯 CPU 密集的阻塞活，放事件循环上跑会把所有请求一起拖死。
        return await _run_image_io(get_thumbnail_response, image_path)

    @router.post("/api/images/delete")
    async def delete_images_endpoint(body: ImageDeleteRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return await _run_image_io(delete_images, body.paths, start_date=body.start_date.strip(), end_date=body.end_date.strip(), all_matching=body.all_matching)

    @router.post("/api/images/download")
    async def download_images_endpoint(body: ImageDownloadRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        # 打包下载要逐张从网络 Bucket 读原图，阻塞时间比单张更长 —— 更不能占共享线程池。
        buf = await _run_image_io(download_images_zip, body.paths)
        return StreamingResponse(
            buf,
            media_type="application/zip",
            headers={"Content-Disposition": 'attachment; filename="images.zip"'},
        )

    @router.get("/api/images/download/{image_path:path}")
    async def download_single_image_endpoint(image_path: str, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        # ⚠️ 同样会把整张图从网络 Bucket 读进内存，必须进**图片专用**线程池。
        return await _run_image_io(get_image_download_response, image_path)

    @router.get("/api/logs")
    async def get_logs(type: str = "", start_date: str = "", end_date: str = "", limit: int = 200, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return {"items": log_service.list(type=type.strip(), start_date=start_date.strip(), end_date=end_date.strip(), limit=limit)}

    @router.post("/api/logs/delete")
    async def delete_logs(body: LogDeleteRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return log_service.delete(body.ids)

    @router.get("/api/error-messages/unconfigured")
    async def get_unconfigured_errors(authorization: str | None = Header(default=None)):
        """前台撞上但还没配文案的报错，供管理员照着补规则。"""
        require_admin(authorization)
        return {"items": await run_in_threadpool(error_message_service.list_unconfigured)}

    @router.post("/api/error-messages/unconfigured/clear")
    async def clear_unconfigured_errors(body: UnconfiguredErrorClearRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        removed = await run_in_threadpool(error_message_service.clear_unconfigured, body.fingerprints or None)
        return {"removed": removed, "items": await run_in_threadpool(error_message_service.list_unconfigured)}

    @router.post("/api/error-messages/summarize")
    async def summarize_error_message(body: ErrorSummarizeRequest, authorization: str | None = Header(default=None)):
        """管理员粘贴一条报错，让模型归纳出可用的规则（关键词 + 大白话）。

        **永不硬失败**：模型不可用 / 超时 / 返回的不是 JSON / 返回了原文里不存在的
        关键词，都会退化成「本地抽取候选关键词」，保证这个按钮总有东西可用。
        """
        identity = require_admin(authorization)
        raw = body.text.strip()
        local_keywords = await run_in_threadpool(extract_keyword_candidates, raw)
        local_result = {"keywords": local_keywords, "display": "", "source": "local"}

        model = (config.error_message_ai_model or "").strip() or config.default_upstream_model_name
        payload = {
            "model": model,
            "stream": False,
            "messages": [
                {"role": "system", "content": ERROR_SUMMARY_SYSTEM_PROMPT},
                {"role": "user", "content": raw},
            ],
        }
        call = LoggedCall(identity, "/api/error-messages/summarize", model, "报错文案归纳", request_text=raw[:200])
        try:
            result = await call.run(openai_v1_chat_complete.handle, payload, identity)
        except Exception as exc:
            return {**local_result, "reason": str(exc)[:300]}

        from api.ai import _chat_completion_text

        answer = _chat_completion_text(result) if isinstance(result, dict) else ""
        suggestion = await run_in_threadpool(parse_ai_suggestion, answer, raw=raw)
        if not suggestion.get("keywords") and not suggestion.get("display"):
            return {**local_result, "reason": "模型未返回可解析的 JSON"}
        # 模型漏了关键词时，用本地抽取的结果补齐，别让这一行是空的。
        return {
            "keywords": suggestion.get("keywords") or local_keywords,
            "display": suggestion.get("display") or "",
            "source": "ai",
            "model": model,
        }

    @router.post("/api/proxy/test")
    async def test_proxy_endpoint(body: ProxyTestRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return {"result": await run_in_threadpool(test_proxy, (body.url or "").strip())}

    @router.post("/api/upstream/probe")
    async def upstream_probe_endpoint(body: UpstreamProbeRequest, authorization: str | None = Header(default=None)):
        """诊断用：从本容器出口直连 chatgpt.com 的若干端点，回报状态/响应头/响应体。

        用于定位「本地好、线上坏」类问题（如 accounts/check 返回 403）。
        仅管理员可用；不写库、不改状态。
        """
        require_admin(authorization)

        def _probe() -> dict[str, Any]:
            import time as _time

            from services.openai_backend_api import OpenAIBackendAPI

            token = (body.access_token or "").strip()
            routes = body.routes or [
                "/backend-api/accounts/check/v4-2023-04-27",
                "/backend-api/me",
            ]
            out: list[dict[str, Any]] = []
            api = OpenAIBackendAPI(access_token=token) if token else None
            try:
                if api is not None:
                    try:
                        api._bootstrap()
                    except Exception as exc:  # noqa: BLE001
                        out.append({"route": "<bootstrap>", "error": f"{type(exc).__name__}: {exc}"})
                for route in routes:
                    started = _time.perf_counter()
                    entry: dict[str, Any] = {"route": route}
                    try:
                        if api is None:
                            from curl_cffi import requests as cffi_requests

                            sess = cffi_requests.Session(impersonate="chrome146", timeout=20)
                            resp = sess.get("https://chatgpt.com" + route)
                        else:
                            resp = api.session.get(
                                api.base_url + route,
                                headers=api._headers(route),
                                timeout=20,
                            )
                        body_text = (getattr(resp, "text", "") or "")
                        entry.update({
                            "status": int(resp.status_code),
                            "content_type": str(resp.headers.get("content-type") or ""),
                            "cf_ray": str(resp.headers.get("cf-ray") or ""),
                            "server": str(resp.headers.get("server") or ""),
                            "body_head": body_text[:500],
                            "body_len": len(body_text),
                            "looks_like_cdn_block": bool(
                                api is not None and api._looks_like_cdn_block(resp)
                            ),
                        })
                    except Exception as exc:  # noqa: BLE001
                        entry["error"] = f"{type(exc).__name__}: {exc}"
                    entry["latency_ms"] = int((_time.perf_counter() - started) * 1000)
                    out.append(entry)
            finally:
                if api is not None:
                    try:
                        api.close()
                    except Exception:
                        pass
            return {
                "egress": "direct",
                "proxy_runtime": proxy_settings.get_runtime_status(),
                "results": out,
            }

        return await run_in_threadpool(_probe)

    @router.get("/api/proxy/runtime")
    async def get_proxy_runtime_endpoint(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return {
            "runtime": config.get_public_proxy_runtime_settings(),
            "status": proxy_settings.get_runtime_status(),
        }

    @router.post("/api/proxy/runtime")
    async def save_proxy_runtime_endpoint(body: SettingsUpdateRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            config.update({"proxy_runtime": body.model_dump(mode="python")})
        except ValueError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc
        return {
            "runtime": config.get_public_proxy_runtime_settings(),
            "status": proxy_settings.get_runtime_status(),
        }

    @router.post("/api/proxy/clearance/test")
    async def test_proxy_clearance_endpoint(body: ClearanceTestRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return {"result": await run_in_threadpool(test_clearance, body.target_url)}

    @router.get("/api/storage/info")
    async def get_storage_info(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        storage = config.get_storage_backend()
        user_storage = config.get_user_storage_backend()
        account_pool_storage = config.get_account_pool_storage_backend()
        return {
            "backend": storage.get_backend_info(),
            "health": storage.health_check(),
            "user_backend": user_storage.get_backend_info(),
            "user_health": user_storage.health_check(),
            "account_pool_backend": account_pool_storage.get_backend_info(),
            "account_pool_health": account_pool_storage.health_check(),
        }

    @router.get("/health", response_model=None)
    async def health_dashboard(format: str = Query(default="html")):
        from services.account_service import account_service as acct_svc

        # ⚠️ 健康检查要保证「容器越忙也测得出来」：这两个都是同步读库/探活的阻塞调用，
        # 直接放事件循环上跑的话，图片请求一多，/health 自己就先被堵死，反而失去判活能力
        # （2026-10-16 线上现象：/health 20~25s 超时，让人误以为容器已死）。
        # 再加一层硬超时：线程池被占满时也必须秒回，否则连"容器是否活着"都测不出来。
        try:
            stats = await asyncio.wait_for(run_in_threadpool(acct_svc.get_stats), timeout=3.0)
        except Exception:
            stats = {"active": 0, "total": 0, "unavailable_stats": True}
        storage = config.get_storage_backend()
        try:
            storage_health = await asyncio.wait_for(run_in_threadpool(storage.health_check), timeout=3.0)
        except Exception:
            storage_health = {"ok": False, "error": "timeout"}
        healthy = int(stats.get("active") or 0) > 0
        payload = {
            "status": "ok" if healthy else "degraded",
            "healthy": healthy,
            "version": app_version,
            "storage": {"backend": storage.get_backend_info(), "health": storage_health},
            "proxy_runtime": proxy_settings.get_runtime_status(),
            "accounts": stats,
        }
        if format == "json":
            return payload
        return HTMLResponse(
            "<!doctype html><html lang='zh'><meta charset='utf-8'>"
            f"<title>号池健康状态</title><body><h1>{payload['status']}</h1>"
            f"<p>version {app_version}</p><pre>{json.dumps(payload, ensure_ascii=False, indent=2)}</pre></body></html>"
        )

    @router.post("/api/admin/import-user-storage")
    async def import_user_storage(body: UserStorageImportRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        auth_keys, shop_state, named_states = _normalize_user_storage_import(body.payload)
        code_count = len(shop_state.get("codes") or []) if isinstance(shop_state, dict) else 0
        ledger_count = len(shop_state.get("ledger") or []) if isinstance(shop_state, dict) else 0
        if body.dry_run:
            return {
                "ok": True,
                "dry_run": True,
                "target": "user_storage",
                "auth_keys": len(auth_keys),
                "codes": code_count,
                "ledger": ledger_count,
                "named_states": sorted(named_states.keys()),
            }

        storage = config.get_user_storage_backend()
        try:
            await run_in_threadpool(storage.save_auth_keys, auth_keys)
            await run_in_threadpool(storage.save_shop_state, shop_state)
            save_named_state = getattr(storage, "save_named_state", None)
            if callable(save_named_state):
                for key, state in named_states.items():
                    await run_in_threadpool(save_named_state, key, state)
        except Exception as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc
        return {
            "ok": True,
            "dry_run": False,
            "target": "user_storage",
            "auth_keys": len(auth_keys),
            "codes": code_count,
            "ledger": ledger_count,
            "named_states": sorted(named_states.keys()),
        }

    @router.post("/auth/register")
    async def register_normal_user(body: NormalUserRegisterRequest):
        try:
            item, raw_key = auth_service.create_invited_normal_user(name=body.name, invite_code=body.invite_code)
        except ValueError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc
        return {
            "ok": True,
            "key": raw_key,
            "role": "normal",
            "subject_id": item.get("id"),
            "name": item.get("name"),
            "account_pool_enabled": False,
            "image_memory_button_enabled": image_memory_button_enabled(item),
            "login_session_duration_hours": config.normal_user_login_session_duration_hours,
            "invite_code": item.get("invite_code"),
        }

    @router.post("/api/backup/test")
    async def test_backup_connection(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            return {"result": await run_in_threadpool(backup_service.test_connection)}
        except BackupError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc

    @router.post("/api/image-storage/test")
    async def test_image_storage_endpoint(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return {"result": await run_in_threadpool(image_storage_service.test_webdav)}

    @router.post("/api/image-storage/sync")
    async def sync_image_storage_endpoint(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            return {"result": await run_in_threadpool(image_storage_service.sync_all)}
        except ImageStorageError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc

    @router.get("/api/backups")
    async def get_backups(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            return {
                "items": await run_in_threadpool(backup_service.list_backups),
                "state": backup_service.get_status(),
                "settings": backup_service.get_settings(),
            }
        except BackupError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc

    @router.post("/api/backups/run")
    async def run_backup_endpoint(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            return {"result": await run_in_threadpool(backup_service.run_backup)}
        except BackupError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc

    @router.post("/api/backups/delete")
    async def delete_backup_endpoint(body: BackupDeleteRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            await run_in_threadpool(backup_service.delete_backup, body.key)
            return {"ok": True}
        except BackupError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc

    @router.get("/api/backups/detail")
    async def get_backup_detail(key: str = "", authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            return {"item": await run_in_threadpool(backup_service.get_backup_detail, key)}
        except BackupError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc

    @router.get("/api/backups/download")
    async def download_backup_endpoint(key: str = "", authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            item = await run_in_threadpool(backup_service.download_backup, key)
        except BackupError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc
        filename = str(item.get("name") or "backup.bin")
        quoted = quote(filename)
        headers = {
            "Content-Disposition": f"attachment; filename*=UTF-8''{quoted}",
            "Content-Length": str(int(item.get("size") or 0)),
        }
        return Response(
            content=bytes(item.get("payload") or b""),
            media_type=str(item.get("content_type") or "application/octet-stream"),
            headers=headers,
        )


    @router.get("/api/images/tags")
    async def list_image_tags(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return {"tags": get_all_tags()}

    @router.post("/api/images/tags")
    async def update_image_tags(body: ImageTagsRequest, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        rel = body.path.strip().lstrip("/")
        if not rel:
            raise HTTPException(status_code=400, detail={"error": "path is required"})
        tags = set_tags(rel, body.tags)
        return {"ok": True, "tags": tags}

    @router.delete("/api/images/tags/{tag}")
    async def delete_image_tag(tag: str, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        count = delete_tag(tag)
        return {"ok": True, "removed_from": count}

    return router