study

Paused

study / services /error_message_service.py

xiaolongmr

报错文案：新增「规则测试」，并把「未匹配兜底」开关真正落地

8d002eb 15 days ago

21.3 kB

"""把上游/后端的技术报错翻译成普通用户看得懂的大白话。
设计要点：
- 规则由管理员在后台配置，形如「关键词列表 -> 展示文案」；命中任意关键词即替换。
- **只影响前台展示**：后台日志（LogService）始终保留原始报错，方便排查。
- 未命中任何规则时保留原始报错，同时把这条「未配置报错」记下来，
  后台会看到一个待办列表，照着补规则即可 —— 规则库能自我完善。
"""
from __future__ import annotations

import json
import re
import threading
import time
from pathlib import Path
from typing import Any

from services.config import DATA_DIR


# 未配置报错的收集文件。与业务数据放在一起，随持久卷保留。
UNCONFIGURED_ERRORS_FILE = DATA_DIR / "error-message-unconfigured.json"

# 同一个报错模板最多保留多少条样本；超出后只累加计数，避免文件无限膨胀。
_MAX_UNCONFIGURED_ENTRIES = 200
# 单条样本的原文上限，防止把整段 HTML/JSON 塞进去。
_MAX_SAMPLE_CHARS = 600
# 关键词的最短长度：太短的关键词（如 "a"）会误伤一大片正常报错。
_MIN_KEYWORD_CHARS = 2
# 归一化后用于去重的指纹长度。
_FINGERPRINT_CHARS = 16

# 未命中任何规则、且管理员打开了「未匹配时报错也兜底成通用文案」时，前台显示这句话。
# ⚠️ 只影响前台文案：原始报错照样进「未配置报错」台账，规则库仍能自我完善。
FALLBACK_DISPLAY = "操作失败，请稍后重试"


class ErrorMessageRuleError(ValueError):
    """规则校验失败。"""


def _clean(value: object) -> str:
    return str(value or "").strip()


def _normalize_keywords(value: object) -> list[str]:
    """把关键词字段归一化成去重、非空、长度达标的列表。
    支持列表或换行/逗号/中文顿号分隔的字符串，方便后台直接粘贴多行。
    """
    raw: list[str] = []
    if isinstance(value, (list, tuple)):
        raw = [str(item or "") for item in value]
    elif isinstance(value, str):
        text = value.replace("，", "\n").replace(",", "\n").replace("、", "\n")
        raw = text.splitlines()
    result: list[str] = []
    seen: set[str] = set()
    for item in raw:
        keyword = item.strip()
        if len(keyword) < _MIN_KEYWORD_CHARS:
            continue
        lowered = keyword.lower()
        if lowered in seen:
            continue
        seen.add(lowered)
        result.append(keyword)
    return result


def normalize_rule(value: object, *, fallback_index: int = 0) -> dict[str, Any] | None:
    """归一化单条规则；缺少关键词或文案则返回 None（视为无效，直接丢弃）。"""
    if not isinstance(value, dict):
        return None
    rule_id = _clean(value.get("id")) or f"rule-{fallback_index + 1}"
    keywords = _normalize_keywords(value.get("keywords"))
    display = _clean(value.get("display"))
    if not keywords or not display:
        return None
    enabled = value.get("enabled")
    if isinstance(enabled, str):
        enabled = enabled.strip().lower() in {"1", "true", "yes", "on"}
    elif enabled is None:
        enabled = True
    else:
        enabled = bool(enabled)
    return {
        "id": rule_id,
        "keywords": keywords,
        "display": display,
        "enabled": enabled,
        "note": _clean(value.get("note")),
    }


def normalize_rules(value: object) -> list[dict[str, Any]]:
    """归一化规则列表，丢弃无效项并保证 id 唯一。"""
    if not isinstance(value, (list, tuple)):
        return []
    rules: list[dict[str, Any]] = []
    seen_ids: set[str] = set()
    for index, item in enumerate(value):
        rule = normalize_rule(item, fallback_index=index)
        if rule is None:
            continue
        base_id = rule["id"]
        suffix = 1
        while rule["id"] in seen_ids:
            suffix += 1
            rule["id"] = f"{base_id}-{suffix}"
        seen_ids.add(rule["id"])
        rules.append(rule)
    return rules


# --------------------------------------------------------------- AI 归纳规则

# 「AI 归纳」的 system prompt：让模型只回 JSON，并禁止它编造原文里没有的词。
ERROR_SUMMARY_SYSTEM_PROMPT = (
    "你是这个上游中转服务的运维助手，负责把一条报错归纳成「报错文案映射规则」。\n"
    "用户会给你一段报错原文，请输出两部分：\n"
    "1. keywords：2-4 个能**唯一标识这一类**报错的关键词，"
    "必须是原文里**原样出现**的子串，不要翻译、不要改写、不要造词。\n"
    "【最关键的一条】本服务的匹配是「原文包含**任意一个**关键词即命中」，"
    "不是全部命中 —— 关键词越多，这条规则覆盖的报错就越宽。"
    "所以宁可少而准，每个关键词都要单凭它自己就能定位到这一类报错。\n"
    "本服务的报错格式固定为 `{接口路径} failed: status={状态码}, body={上游返回体}`。"
    "接口路径是上游接口，同一个前缀会对应多种完全不同的操作"
    "（例如 /backend-api/files 下面既有创建、上传确认，也有下载和文件库），"
    "所以取路径时要**带上能区分「哪一步」的片段**（如 /uploaded、/download、/library、/conversation），"
    "只写 /backend-api/files 这样的前缀会误伤一大片其它报错。\n"
    "优先选：错误标识符（如 upload_protection_limit_exceeded）、"
    "「步骤 + 状态码」组合（如 /uploaded failed: status=500）、"
    "状态码（如 status=429）、curl 错误号（如 curl: (16)）。\n"
    "严禁选过于通用、会误伤正常报错的词（如 error、failed、exception、status、body、detail、"
    "只出现一次的 file_id / uuid）。\n"
    "2. display：一句给普通用户看的中文说明，不超过 20 个字。"
    "不要出现英文术语、状态码、接口路径、账号邮箱、网址或联系方式。\n"
    "只返回 JSON，不要解释、不要 markdown 代码块，格式固定为："
    '{"keywords": ["..."], "display": "..."}'
)

# 抽取候选关键词时要忽略的「太通用」片段：拿它们当规则会误伤一片正常报错。
# 其中一批是 JSON 字段名（error_code、reason…）——它们是报错的「骨架」，
# 几乎每条报错都有，当关键词等于没筛。
_GENERIC_TOKENS = frozenset({
    "error", "errors", "failed", "failure", "exception", "status", "body",
    "detail", "details", "message", "msg", "type", "code", "request",
    "response", "http", "https", "true", "false", "null", "none", "data",
    "error_code", "error_message", "error_type", "error_detail",
})

# 每次请求都会变的高熵片段（uuid / 上游 file_id / 长十六进制）。
# 把它们当关键词等于给单次请求打标，下一次必然命中不了。
_VOLATILE_SEGMENT = re.compile(
    r"^(?:[0-9a-f]{8,}|file_[0-9a-z]{6,}|[0-9]{6,})$",
    re.IGNORECASE,
)


def _strip_volatile_path(path: str) -> str:
    """把路径截断到第一个高熵片段之前。
    `/backend-api/files/file_00000000ab.../uploaded` → `/backend-api/files`：
    前两段才代表链路，后面是单次请求的 id。
    """
    kept: list[str] = []
    for segment in str(path or "").split("/"):
        if segment and _VOLATILE_SEGMENT.match(segment):
            break
        kept.append(segment)
    trimmed = "/".join(kept).rstrip("/")
    # 只留单段的路径太泛（/files 之类），不值得当关键词。
    return trimmed if trimmed.count("/") >= 2 else ""


# 形如 `"reason": "xxx"` 的 JSON 键值片段。
_JSON_FRAGMENT = re.compile(r'"\s*:\s*"')


def _sanitize_keyword(value: object) -> str:
    """剥掉关键词两端的 JSON 语法碎片，只留下「值」本身。
    归纳（模型或本地抽取）偶尔会把 `reason": "upload_protection_limit_exceeded`
    这种从报文里直接截出来的片段整条当关键词。它虽然能把当前这条报错匹配上，
    但可读性差，而且上游 JSON 的字段顺序/空格一变就失效 —— 例如相邻字段换成
    `"a": "reason": "upload_...` 时，`reason` 就跑到中间、不再是行首了。
    ⚠️ 只用于「归纳」链路；后台手输的关键词走 _normalize_keywords 原样保留，
    避免把管理员明确想匹配的带引号子串改掉。
    """
    text = str(value or "").strip()
    if not text:
        return ""
    parts = _JSON_FRAGMENT.split(text)
    if len(parts) > 1:
        text = parts[-1]
    return text.strip("\"':,{}[]\\").strip()


def _dedupe_ordered(items: list[str]) -> list[str]:
    """按首次出现顺序去重，同时过滤空值和过短的项。"""
    result: list[str] = []
    seen: set[str] = set()
    for item in items:
        text = str(item or "").strip()
        if len(text) < _MIN_KEYWORD_CHARS:
            continue
        lowered = text.lower()
        if lowered in seen:
            continue
        seen.add(lowered)
        result.append(text)
    return result


def extract_keyword_candidates(message: object) -> list[str]:
    """从报错原文里抽出可直接用作关键词的片段（不依赖模型）。
    这是「AI 归纳」的兜底：模型不可用、超时或被限流时，仍然能给出一组
    高辨识度的候选词，管理员补一句文案即可。抽取顺序即优先级，越靠前
    越能唯一标识这类报错。
    """
    import re

    text = str(message or "")
    if not text.strip():
        return []
    candidates: list[str] = []

    # 1) 结构化字段值：reason / error_code / code —— 这些是上游给的精确标识
    for pattern in (
        r'"reason"\s*:\s*"([^"]+)"',
        r'"error_code"\s*:\s*"([^"]+)"',
        r'"code"\s*:\s*"([^"]+)"',
    ):
        candidates.extend(re.findall(pattern, text, flags=re.IGNORECASE))

    # 2) 接口路径：/backend-api/xxx 这类前缀最能区分链路。
    #    路径里常混着每次请求都变的 file_id / uuid，必须截断，
    #    否则规则只对这一次报错生效，下次换 id 就再也命中不了。
    candidates.extend(
        trimmed
        for trimmed in (
            _strip_volatile_path(item)
            for item in re.findall(r"/[a-z0-9][a-z0-9._/-]{3,}", text, flags=re.IGNORECASE)
        )
        if trimmed
    )

    # 3) 状态码：status=429
    candidates.extend(re.findall(r"status=\d{3}", text, flags=re.IGNORECASE))

    # 4) curl 错误号：curl: (16)
    candidates.extend(re.findall(r"curl:\s*\(\d+\)", text, flags=re.IGNORECASE))

    # 5) 长标识符（snake_case / 驼峰之外的连续字母数字串）；高熵的不收。
    candidates.extend(
        item
        for item in re.findall(r"[a-z][a-z0-9_]{7,}", text, flags=re.IGNORECASE)
        if not _VOLATILE_SEGMENT.match(item)
    )

    # 路径会带出 /backend-api、/files 之类碎片，去掉只有斜杠的单段噪声
    cleaned: list[str] = []
    for item in candidates:
        fixed = _sanitize_keyword(item)
        if len(fixed) < _MIN_KEYWORD_CHARS or fixed == "/":
            continue
        if fixed.lower() in _GENERIC_TOKENS:
            continue
        cleaned.append(fixed)
    return _dedupe_ordered(cleaned)[:12]


def parse_ai_suggestion(value: object, *, raw: str = "") -> dict[str, Any]:
    """把模型返回的文本解析成 {keywords, display}。
    - 容忍 markdown 代码块、前后解释文字（取第一个 JSON 对象）。
    - **只保留在原文里真实出现的关键词**：匹配是子串匹配，编造的词永远命中不了。
    """
    import json
    import re

    text = str(value or "").strip()
    if not text:
        return {"keywords": [], "display": ""}
    if text.startswith("```"):
        text = re.sub(r"^```[a-zA-Z]*\s*", "", text)
        text = re.sub(r"\s*```$", "", text).strip()
    match = re.search(r"\{.*\}", text, flags=re.DOTALL)
    if match:
        text = match.group(0)
    try:
        data = json.loads(text)
    except ValueError:
        return {"keywords": [], "display": ""}
    if not isinstance(data, dict):
        return {"keywords": [], "display": ""}
    # 先剥 JSON 语法碎片（模型会把 `reason": "upload_xxx` 整条当关键词），再去重。
    keywords = _normalize_keywords(
        [_sanitize_keyword(item) for item in _normalize_keywords(data.get("keywords"))]
    )
    # 通用 JSON 字段名（error_code / reason / message…）几乎每条报错都有，
    # 当关键词等于把所有报错都命中，必须剔掉（本地抽取走的是同一份黑名单）。
    keywords = [keyword for keyword in keywords if keyword.lower() not in _GENERIC_TOKENS]
    lowered = str(raw or "").lower()
    if lowered:
        keywords = [keyword for keyword in keywords if keyword.lower() in lowered]
    return {"keywords": keywords, "display": _clean(data.get("display"))}


def _fingerprint(message: str) -> str:
    """把报错原文压成一个稳定指纹，用于同模板去重。
    同一类报错每次的 uuid / 时间戳 / 文件名都不同，直接按原文去重会得到
    成百上千条噪声；这里剥掉明显的可变片段（十六进制串、数字、引号内容）
    后再算指纹。
    """
    import hashlib
    import re

    text = str(message or "")
    # UUID（8-4-4-4-12）整段归一，避免只吃掉前半段、剩余片段被 \d+ 打成
    # 不同长度的 `#` 序列，导致同类报错算出不同指纹。
    normalized = re.sub(
        r"\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b",
        "#",
        text,
        flags=re.IGNORECASE,
    )
    # 其余长十六进制串（hash / token / 文件名）同样归一。
    normalized = re.sub(r"[0-9a-f]{8,}", "#", normalized, flags=re.IGNORECASE)
    normalized = re.sub(r"\d+", "#", normalized)
    normalized = re.sub(r"\s+", " ", normalized).strip().lower()
    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:_FINGERPRINT_CHARS]


class ErrorMessageService:
    def __init__(self, store_path: Path | None = None) -> None:
        self._path = store_path or UNCONFIGURED_ERRORS_FILE
        self._lock = threading.RLock()
        self._unconfigured_cache: dict[str, dict[str, Any]] | None = None

    # ---------------------------------------------------------------- 规则

    def _rules(self) -> list[dict[str, Any]]:
        # 延迟导入：config 在导入期会读取 DATA_DIR，避免循环依赖。
        from services.config import config

        return normalize_rules(config.data.get("error_message_rules"))

    def translate(self, message: object, *, channel: str = "") -> str:
        """把一条原始报错翻译成前台文案。
        命中规则 -> 返回配置的大白话；
        未命中   -> 开了「兜底」时返回通用文案，否则原样返回，并记录一条「未配置报错」待办。
        """
        text = _clean(message)
        if not text:
            return text
        lowered = text.lower()
        for rule in self._rules():
            if not rule.get("enabled", True):
                continue
            if any(keyword.lower() in lowered for keyword in rule["keywords"]):
                return rule["display"]
        self._record_unconfigured(text, channel=channel)
        # 兜底只换前台文案；原文此时已记进「未配置报错」台账，管理员照旧能照着补规则。
        return FALLBACK_DISPLAY if self.fallback_enabled() else text

    @staticmethod
    def fallback_enabled() -> bool:
        """「未匹配时报错也兜底成通用文案」开关（后台配置，默认关）。"""
        from services.config import config

        try:
            return bool(config.error_message_fallback_enabled)
        except Exception:  # noqa: BLE001 - 配置读不到时按「不兜底」处理，别把原文吞掉
            return False

    def translate_detail(self, detail: object, *, channel: str = "") -> object:
        """对 FastAPI 的 detail 结构做就地翻译，保持原有结构。"""
        if isinstance(detail, str):
            return self.translate(detail, channel=channel)
        if isinstance(detail, list):
            return [self.translate_detail(item, channel=channel) for item in detail]
        if isinstance(detail, dict):
            translated: dict[Any, Any] = {}
            for key, value in detail.items():
                # 只翻译真正承载文案的字段，避免把 code / type 也换掉。
                if key in {"message", "error", "detail", "msg"}:
                    translated[key] = self.translate_detail(value, channel=channel)
                else:
                    translated[key] = value
            return translated
        return detail

    # ------------------------------------------------------- 未配置报错收集

    def _load_unconfigured(self) -> dict[str, dict[str, Any]]:
        with self._lock:
            if self._unconfigured_cache is not None:
                return self._unconfigured_cache
            data: dict[str, dict[str, Any]] = {}
            try:
                raw = json.loads(self._path.read_text(encoding="utf-8"))
            except (OSError, ValueError):
                raw = {}
            items = raw.get("items") if isinstance(raw, dict) else None
            if isinstance(items, list):
                for item in items:
                    if not isinstance(item, dict):
                        continue
                    fingerprint = _clean(item.get("fingerprint"))
                    if fingerprint:
                        data[fingerprint] = item
            self._unconfigured_cache = data
            return data

    def _save_unconfigured(self, data: dict[str, dict[str, Any]]) -> None:
        with self._lock:
            payload = {"items": sorted(data.values(), key=lambda item: str(item.get("last_seen") or ""), reverse=True)}
            try:
                self._path.parent.mkdir(parents=True, exist_ok=True)
                temp = self._path.with_suffix(self._path.suffix + ".tmp")
                temp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
                temp.replace(self._path)
            except OSError:
                # 收集失败绝不能影响正常请求链路。
                return
            self._unconfigured_cache = data

    def _record_unconfigured(self, message: str, *, channel: str = "") -> None:
        fingerprint = _fingerprint(message)
        sample = message[:_MAX_SAMPLE_CHARS]
        with self._lock:
            data = self._load_unconfigured()
            existing = data.get(fingerprint)
            now = time.strftime("%Y-%m-%d %H:%M:%S")
            if existing is None and len(data) >= _MAX_UNCONFIGURED_ENTRIES:
                self._prune_locked(data)
            if existing is None:
                data[fingerprint] = {
                    "fingerprint": fingerprint,
                    "sample": sample,
                    "count": 1,
                    "channel": _clean(channel),
                    "first_seen": now,
                    "last_seen": now,
                }
            else:
                entry = dict(existing)
                entry["count"] = int(entry.get("count") or 0) + 1
                entry["last_seen"] = now
                entry["sample"] = sample
                if channel and not entry.get("channel"):
                    entry["channel"] = _clean(channel)
                data[fingerprint] = entry
            self._save_unconfigured(data)

    @staticmethod
    def _prune_locked(data: dict[str, dict[str, Any]]) -> None:
        """超出上限时先淘汰最久未出现、且出现次数最少的条目。"""
        ordered = sorted(
            data.values(),
            key=lambda item: (int(item.get("count") or 0), str(item.get("last_seen") or "")),
        )
        for item in ordered[: max(1, len(ordered) // 10)]:
            data.pop(_clean(item.get("fingerprint")), None)

    def list_unconfigured(self) -> list[dict[str, Any]]:
        data = self._load_unconfigured()
        items = sorted(
            (dict(item) for item in data.values()),
            key=lambda item: (
                -int(item.get("count") or 0),
                str(item.get("last_seen") or ""),
            ),
        )
        # 顺带给出可直接用作关键词的候选片段，省得管理员从整段报错里手抠。
        for item in items:
            item["keywords"] = extract_keyword_candidates(item.get("sample"))
        return items

    def clear_unconfigured(self, fingerprints: list[str] | None = None) -> int:
        with self._lock:
            data = self._load_unconfigured()
            if not fingerprints:
                removed = len(data)
                data.clear()
            else:
                removed = 0
                for raw in fingerprints:
                    fingerprint = _clean(raw)
                    if fingerprint in data:
                        data.pop(fingerprint)
                        removed += 1
            self._save_unconfigured(data)
            return removed


error_message_service = ErrorMessageService()

	"""把上游/后端的技术报错翻译成普通用户看得懂的大白话。

	设计要点：
	- 规则由管理员在后台配置，形如「关键词列表 -> 展示文案」；命中任意关键词即替换。
	- 只影响前台展示：后台日志（LogService）始终保留原始报错，方便排查。
	- 未命中任何规则时保留原始报错，同时把这条「未配置报错」记下来，
	后台会看到一个待办列表，照着补规则即可 —— 规则库能自我完善。
	"""
	from __future__ import annotations

	import json
	import re
	import threading
	import time
	from pathlib import Path
	from typing import Any

	from services.config import DATA_DIR


	# 未配置报错的收集文件。与业务数据放在一起，随持久卷保留。
	UNCONFIGURED_ERRORS_FILE = DATA_DIR / "error-message-unconfigured.json"

	# 同一个报错模板最多保留多少条样本；超出后只累加计数，避免文件无限膨胀。
	_MAX_UNCONFIGURED_ENTRIES = 200
	# 单条样本的原文上限，防止把整段 HTML/JSON 塞进去。
	_MAX_SAMPLE_CHARS = 600
	# 关键词的最短长度：太短的关键词（如 "a"）会误伤一大片正常报错。
	_MIN_KEYWORD_CHARS = 2
	# 归一化后用于去重的指纹长度。
	_FINGERPRINT_CHARS = 16

	# 未命中任何规则、且管理员打开了「未匹配时报错也兜底成通用文案」时，前台显示这句话。
	# ⚠️ 只影响前台文案：原始报错照样进「未配置报错」台账，规则库仍能自我完善。
	FALLBACK_DISPLAY = "操作失败，请稍后重试"


	class ErrorMessageRuleError(ValueError):
	"""规则校验失败。"""


	def _clean(value: object) -> str:
	return str(value or "").strip()


	def _normalize_keywords(value: object) -> list[str]:
	"""把关键词字段归一化成去重、非空、长度达标的列表。

	支持列表或换行/逗号/中文顿号分隔的字符串，方便后台直接粘贴多行。
	"""
	raw: list[str] = []
	if isinstance(value, (list, tuple)):
	raw = [str(item or "") for item in value]
	elif isinstance(value, str):
	text = value.replace("，", "\n").replace(",", "\n").replace("、", "\n")
	raw = text.splitlines()
	result: list[str] = []
	seen: set[str] = set()
	for item in raw:
	keyword = item.strip()
	if len(keyword) < _MIN_KEYWORD_CHARS:
	continue
	lowered = keyword.lower()
	if lowered in seen:
	continue
	seen.add(lowered)
	result.append(keyword)
	return result


	def normalize_rule(value: object, *, fallback_index: int = 0) -> dict[str, Any] \| None:
	"""归一化单条规则；缺少关键词或文案则返回 None（视为无效，直接丢弃）。"""
	if not isinstance(value, dict):
	return None
	rule_id = _clean(value.get("id")) or f"rule-{fallback_index + 1}"
	keywords = _normalize_keywords(value.get("keywords"))
	display = _clean(value.get("display"))
	if not keywords or not display:
	return None
	enabled = value.get("enabled")
	if isinstance(enabled, str):
	enabled = enabled.strip().lower() in {"1", "true", "yes", "on"}
	elif enabled is None:
	enabled = True
	else:
	enabled = bool(enabled)
	return {
	"id": rule_id,
	"keywords": keywords,
	"display": display,
	"enabled": enabled,
	"note": _clean(value.get("note")),
	}


	def normalize_rules(value: object) -> list[dict[str, Any]]:
	"""归一化规则列表，丢弃无效项并保证 id 唯一。"""
	if not isinstance(value, (list, tuple)):
	return []
	rules: list[dict[str, Any]] = []
	seen_ids: set[str] = set()
	for index, item in enumerate(value):
	rule = normalize_rule(item, fallback_index=index)
	if rule is None:
	continue
	base_id = rule["id"]
	suffix = 1
	while rule["id"] in seen_ids:
	suffix += 1
	rule["id"] = f"{base_id}-{suffix}"
	seen_ids.add(rule["id"])
	rules.append(rule)
	return rules


	# --------------------------------------------------------------- AI 归纳规则

	# 「AI 归纳」的 system prompt：让模型只回 JSON，并禁止它编造原文里没有的词。
	ERROR_SUMMARY_SYSTEM_PROMPT = (
	"你是这个上游中转服务的运维助手，负责把一条报错归纳成「报错文案映射规则」。\n"
	"用户会给你一段报错原文，请输出两部分：\n"
	"1. keywords：2-4 个能唯一标识这一类报错的关键词，"
	"必须是原文里原样出现的子串，不要翻译、不要改写、不要造词。\n"
	"【最关键的一条】本服务的匹配是「原文包含任意一个关键词即命中」，"
	"不是全部命中 —— 关键词越多，这条规则覆盖的报错就越宽。"
	"所以宁可少而准，每个关键词都要单凭它自己就能定位到这一类报错。\n"
	"本服务的报错格式固定为 `{接口路径} failed: status={状态码}, body={上游返回体}`。"
	"接口路径是上游接口，同一个前缀会对应多种完全不同的操作"
	"（例如 /backend-api/files 下面既有创建、上传确认，也有下载和文件库），"
	"所以取路径时要带上能区分「哪一步」的片段（如 /uploaded、/download、/library、/conversation），"
	"只写 /backend-api/files 这样的前缀会误伤一大片其它报错。\n"
	"优先选：错误标识符（如 upload_protection_limit_exceeded）、"
	"「步骤 + 状态码」组合（如 /uploaded failed: status=500）、"
	"状态码（如 status=429）、curl 错误号（如 curl: (16)）。\n"
	"严禁选过于通用、会误伤正常报错的词（如 error、failed、exception、status、body、detail、"
	"只出现一次的 file_id / uuid）。\n"
	"2. display：一句给普通用户看的中文说明，不超过 20 个字。"
	"不要出现英文术语、状态码、接口路径、账号邮箱、网址或联系方式。\n"
	"只返回 JSON，不要解释、不要 markdown 代码块，格式固定为："
	'{"keywords": ["..."], "display": "..."}'
	)

	# 抽取候选关键词时要忽略的「太通用」片段：拿它们当规则会误伤一片正常报错。
	# 其中一批是 JSON 字段名（error_code、reason…）——它们是报错的「骨架」，
	# 几乎每条报错都有，当关键词等于没筛。
	_GENERIC_TOKENS = frozenset({
	"error", "errors", "failed", "failure", "exception", "status", "body",
	"detail", "details", "message", "msg", "type", "code", "request",
	"response", "http", "https", "true", "false", "null", "none", "data",
	"error_code", "error_message", "error_type", "error_detail",
	})

	# 每次请求都会变的高熵片段（uuid / 上游 file_id / 长十六进制）。
	# 把它们当关键词等于给单次请求打标，下一次必然命中不了。
	_VOLATILE_SEGMENT = re.compile(
	r"^(?:[0-9a-f]{8,}\|file_[0-9a-z]{6,}\|[0-9]{6,})$",
	re.IGNORECASE,
	)


	def _strip_volatile_path(path: str) -> str:
	"""把路径截断到第一个高熵片段之前。

	`/backend-api/files/file_00000000ab.../uploaded` → `/backend-api/files`：
	前两段才代表链路，后面是单次请求的 id。
	"""
	kept: list[str] = []
	for segment in str(path or "").split("/"):
	if segment and _VOLATILE_SEGMENT.match(segment):
	break
	kept.append(segment)
	trimmed = "/".join(kept).rstrip("/")
	# 只留单段的路径太泛（/files 之类），不值得当关键词。
	return trimmed if trimmed.count("/") >= 2 else ""


	# 形如 `"reason": "xxx"` 的 JSON 键值片段。
	_JSON_FRAGMENT = re.compile(r'"\s:\s"')


	def _sanitize_keyword(value: object) -> str:
	"""剥掉关键词两端的 JSON 语法碎片，只留下「值」本身。

	归纳（模型或本地抽取）偶尔会把 `reason": "upload_protection_limit_exceeded`
	这种从报文里直接截出来的片段整条当关键词。它虽然能把当前这条报错匹配上，
	但可读性差，而且上游 JSON 的字段顺序/空格一变就失效 —— 例如相邻字段换成
	`"a": "reason": "upload_...` 时，`reason` 就跑到中间、不再是行首了。

	⚠️ 只用于「归纳」链路；后台手输的关键词走 _normalize_keywords 原样保留，
	避免把管理员明确想匹配的带引号子串改掉。
	"""
	text = str(value or "").strip()
	if not text:
	return ""
	parts = _JSON_FRAGMENT.split(text)
	if len(parts) > 1:
	text = parts[-1]
	return text.strip("\"':,{}[]\\").strip()


	def _dedupe_ordered(items: list[str]) -> list[str]:
	"""按首次出现顺序去重，同时过滤空值和过短的项。"""
	result: list[str] = []
	seen: set[str] = set()
	for item in items:
	text = str(item or "").strip()
	if len(text) < _MIN_KEYWORD_CHARS:
	continue
	lowered = text.lower()
	if lowered in seen:
	continue
	seen.add(lowered)
	result.append(text)
	return result


	def extract_keyword_candidates(message: object) -> list[str]:
	"""从报错原文里抽出可直接用作关键词的片段（不依赖模型）。

	这是「AI 归纳」的兜底：模型不可用、超时或被限流时，仍然能给出一组
	高辨识度的候选词，管理员补一句文案即可。抽取顺序即优先级，越靠前
	越能唯一标识这类报错。
	"""
	import re

	text = str(message or "")
	if not text.strip():
	return []
	candidates: list[str] = []

	# 1) 结构化字段值：reason / error_code / code —— 这些是上游给的精确标识
	for pattern in (
	r'"reason"\s:\s"([^"]+)"',
	r'"error_code"\s:\s"([^"]+)"',
	r'"code"\s:\s"([^"]+)"',
	):
	candidates.extend(re.findall(pattern, text, flags=re.IGNORECASE))

	# 2) 接口路径：/backend-api/xxx 这类前缀最能区分链路。
	# 路径里常混着每次请求都变的 file_id / uuid，必须截断，
	# 否则规则只对这一次报错生效，下次换 id 就再也命中不了。
	candidates.extend(
	trimmed
	for trimmed in (
	_strip_volatile_path(item)
	for item in re.findall(r"/[a-z0-9][a-z0-9._/-]{3,}", text, flags=re.IGNORECASE)
	)
	if trimmed
	)

	# 3) 状态码：status=429
	candidates.extend(re.findall(r"status=\d{3}", text, flags=re.IGNORECASE))

	# 4) curl 错误号：curl: (16)
	candidates.extend(re.findall(r"curl:\s*\(\d+\)", text, flags=re.IGNORECASE))

	# 5) 长标识符（snake_case / 驼峰之外的连续字母数字串）；高熵的不收。
	candidates.extend(
	item
	for item in re.findall(r"[a-z][a-z0-9_]{7,}", text, flags=re.IGNORECASE)
	if not _VOLATILE_SEGMENT.match(item)
	)

	# 路径会带出 /backend-api、/files 之类碎片，去掉只有斜杠的单段噪声
	cleaned: list[str] = []
	for item in candidates:
	fixed = _sanitize_keyword(item)
	if len(fixed) < _MIN_KEYWORD_CHARS or fixed == "/":
	continue
	if fixed.lower() in _GENERIC_TOKENS:
	continue
	cleaned.append(fixed)
	return _dedupe_ordered(cleaned)[:12]


	def parse_ai_suggestion(value: object, *, raw: str = "") -> dict[str, Any]:
	"""把模型返回的文本解析成 {keywords, display}。

	- 容忍 markdown 代码块、前后解释文字（取第一个 JSON 对象）。
	- 只保留在原文里真实出现的关键词：匹配是子串匹配，编造的词永远命中不了。
	"""
	import json
	import re

	text = str(value or "").strip()
	if not text:
	return {"keywords": [], "display": ""}
	if text.startswith("```"):
	text = re.sub(r"^```[a-zA-Z]\s", "", text)
	text = re.sub(r"\s*```$", "", text).strip()
	match = re.search(r"\{.*\}", text, flags=re.DOTALL)
	if match:
	text = match.group(0)
	try:
	data = json.loads(text)
	except ValueError:
	return {"keywords": [], "display": ""}
	if not isinstance(data, dict):
	return {"keywords": [], "display": ""}
	# 先剥 JSON 语法碎片（模型会把 `reason": "upload_xxx` 整条当关键词），再去重。
	keywords = _normalize_keywords(
	[_sanitize_keyword(item) for item in _normalize_keywords(data.get("keywords"))]
	)
	# 通用 JSON 字段名（error_code / reason / message…）几乎每条报错都有，
	# 当关键词等于把所有报错都命中，必须剔掉（本地抽取走的是同一份黑名单）。
	keywords = [keyword for keyword in keywords if keyword.lower() not in _GENERIC_TOKENS]
	lowered = str(raw or "").lower()
	if lowered:
	keywords = [keyword for keyword in keywords if keyword.lower() in lowered]
	return {"keywords": keywords, "display": _clean(data.get("display"))}


	def _fingerprint(message: str) -> str:
	"""把报错原文压成一个稳定指纹，用于同模板去重。

	同一类报错每次的 uuid / 时间戳 / 文件名都不同，直接按原文去重会得到
	成百上千条噪声；这里剥掉明显的可变片段（十六进制串、数字、引号内容）
	后再算指纹。
	"""
	import hashlib
	import re

	text = str(message or "")
	# UUID（8-4-4-4-12）整段归一，避免只吃掉前半段、剩余片段被 \d+ 打成
	# 不同长度的 `#` 序列，导致同类报错算出不同指纹。
	normalized = re.sub(
	r"\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b",
	"#",
	text,
	flags=re.IGNORECASE,
	)
	# 其余长十六进制串（hash / token / 文件名）同样归一。
	normalized = re.sub(r"[0-9a-f]{8,}", "#", normalized, flags=re.IGNORECASE)
	normalized = re.sub(r"\d+", "#", normalized)
	normalized = re.sub(r"\s+", " ", normalized).strip().lower()
	return hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:_FINGERPRINT_CHARS]


	class ErrorMessageService:
	def __init__(self, store_path: Path \| None = None) -> None:
	self._path = store_path or UNCONFIGURED_ERRORS_FILE
	self._lock = threading.RLock()
	self._unconfigured_cache: dict[str, dict[str, Any]] \| None = None

	# ---------------------------------------------------------------- 规则

	def _rules(self) -> list[dict[str, Any]]:
	# 延迟导入：config 在导入期会读取 DATA_DIR，避免循环依赖。
	from services.config import config

	return normalize_rules(config.data.get("error_message_rules"))

	def translate(self, message: object, *, channel: str = "") -> str:
	"""把一条原始报错翻译成前台文案。

	命中规则 -> 返回配置的大白话；
	未命中 -> 开了「兜底」时返回通用文案，否则原样返回，并记录一条「未配置报错」待办。
	"""
	text = _clean(message)
	if not text:
	return text
	lowered = text.lower()
	for rule in self._rules():
	if not rule.get("enabled", True):
	continue
	if any(keyword.lower() in lowered for keyword in rule["keywords"]):
	return rule["display"]
	self._record_unconfigured(text, channel=channel)
	# 兜底只换前台文案；原文此时已记进「未配置报错」台账，管理员照旧能照着补规则。
	return FALLBACK_DISPLAY if self.fallback_enabled() else text

	@staticmethod
	def fallback_enabled() -> bool:
	"""「未匹配时报错也兜底成通用文案」开关（后台配置，默认关）。"""
	from services.config import config

	try:
	return bool(config.error_message_fallback_enabled)
	except Exception: # noqa: BLE001 - 配置读不到时按「不兜底」处理，别把原文吞掉
	return False

	def translate_detail(self, detail: object, *, channel: str = "") -> object:
	"""对 FastAPI 的 detail 结构做就地翻译，保持原有结构。"""
	if isinstance(detail, str):
	return self.translate(detail, channel=channel)
	if isinstance(detail, list):
	return [self.translate_detail(item, channel=channel) for item in detail]
	if isinstance(detail, dict):
	translated: dict[Any, Any] = {}
	for key, value in detail.items():
	# 只翻译真正承载文案的字段，避免把 code / type 也换掉。
	if key in {"message", "error", "detail", "msg"}:
	translated[key] = self.translate_detail(value, channel=channel)
	else:
	translated[key] = value
	return translated
	return detail

	# ------------------------------------------------------- 未配置报错收集

	def _load_unconfigured(self) -> dict[str, dict[str, Any]]:
	with self._lock:
	if self._unconfigured_cache is not None:
	return self._unconfigured_cache
	data: dict[str, dict[str, Any]] = {}
	try:
	raw = json.loads(self._path.read_text(encoding="utf-8"))
	except (OSError, ValueError):
	raw = {}
	items = raw.get("items") if isinstance(raw, dict) else None
	if isinstance(items, list):
	for item in items:
	if not isinstance(item, dict):
	continue
	fingerprint = _clean(item.get("fingerprint"))
	if fingerprint:
	data[fingerprint] = item
	self._unconfigured_cache = data
	return data

	def _save_unconfigured(self, data: dict[str, dict[str, Any]]) -> None:
	with self._lock:
	payload = {"items": sorted(data.values(), key=lambda item: str(item.get("last_seen") or ""), reverse=True)}
	try:
	self._path.parent.mkdir(parents=True, exist_ok=True)
	temp = self._path.with_suffix(self._path.suffix + ".tmp")
	temp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
	temp.replace(self._path)
	except OSError:
	# 收集失败绝不能影响正常请求链路。
	return
	self._unconfigured_cache = data

	def _record_unconfigured(self, message: str, *, channel: str = "") -> None:
	fingerprint = _fingerprint(message)
	sample = message[:_MAX_SAMPLE_CHARS]
	with self._lock:
	data = self._load_unconfigured()
	existing = data.get(fingerprint)
	now = time.strftime("%Y-%m-%d %H:%M:%S")
	if existing is None and len(data) >= _MAX_UNCONFIGURED_ENTRIES:
	self._prune_locked(data)
	if existing is None:
	data[fingerprint] = {
	"fingerprint": fingerprint,
	"sample": sample,
	"count": 1,
	"channel": _clean(channel),
	"first_seen": now,
	"last_seen": now,
	}
	else:
	entry = dict(existing)
	entry["count"] = int(entry.get("count") or 0) + 1
	entry["last_seen"] = now
	entry["sample"] = sample
	if channel and not entry.get("channel"):
	entry["channel"] = _clean(channel)
	data[fingerprint] = entry
	self._save_unconfigured(data)

	@staticmethod
	def _prune_locked(data: dict[str, dict[str, Any]]) -> None:
	"""超出上限时先淘汰最久未出现、且出现次数最少的条目。"""
	ordered = sorted(
	data.values(),
	key=lambda item: (int(item.get("count") or 0), str(item.get("last_seen") or "")),
	)
	for item in ordered[: max(1, len(ordered) // 10)]:
	data.pop(_clean(item.get("fingerprint")), None)

	def list_unconfigured(self) -> list[dict[str, Any]]:
	data = self._load_unconfigured()
	items = sorted(
	(dict(item) for item in data.values()),
	key=lambda item: (
	-int(item.get("count") or 0),
	str(item.get("last_seen") or ""),
	),
	)
	# 顺带给出可直接用作关键词的候选片段，省得管理员从整段报错里手抠。
	for item in items:
	item["keywords"] = extract_keyword_candidates(item.get("sample"))
	return items

	def clear_unconfigured(self, fingerprints: list[str] \| None = None) -> int:
	with self._lock:
	data = self._load_unconfigured()
	if not fingerprints:
	removed = len(data)
	data.clear()
	else:
	removed = 0
	for raw in fingerprints:
	fingerprint = _clean(raw)
	if fingerprint in data:
	data.pop(fingerprint)
	removed += 1
	self._save_unconfigured(data)
	return removed


	error_message_service = ErrorMessageService()