"""plugin-tips-cn — 把 Hermes 发出来的英文系统提示，就地换成随机的日常中文说法。 三层网（缺一层就会漏）： 1. ``transform_tool_result`` —— 工具结果里的英文（超时、权限、路径、接口报错…） 2. ``transform_llm_output`` —— 助手最终回复里的同类英文 3. **适配器出口包装** —— 网关自己发到聊天的通知（忙时回执、进度心跳、自主进化复盘、 后台任务收工、cron 失败…）。**这些不走任何插件钩子**（源码里直接 ``adapter.send(...)``），所以只能把适配器的文本发送方法包起来。这是本插件 最关键的一层，少了它，聊天里看到的还是英文。 词库在 ``phrases.json``，每类可放任意多条中文候选，命中随机挑一条；改完不用重启 （按 mtime 热加载）。 开关（环境变量，新旧名字都认）： PLUGIN_TIPS_CN_DISABLE=1 / CN_VOICE_DISABLE=1 整个插件静默（默认开） PLUGIN_TIPS_CN_LOG=1 / CN_VOICE_LOG=1 每次替换打日志 PLUGIN_TIPS_CN_NO_SEND=1 只关「适配器出口」那一层 """ from __future__ import annotations import inspect import json import logging import os import random import re from pathlib import Path from typing import Any, Dict, List, Optional, Tuple logger = logging.getLogger(__name__) _HERE = Path(__file__).resolve().parent _PHRASES = _HERE / "phrases.json" # 太大的文本（多 MB 的工具输出）跳过：收益低、还拖慢循环。 _MAX_LEN = 200_000 _TRUTHY = {"1", "true", "yes", "on"} # 适配器上「发文本」的方法名。send/edit_message 覆盖绝大多数； # send_stream_frame 是原生流式（企微走这条）；send_exec_approval/send_clarify/ # send_slash_confirm 是审批与提问；send_image 等带 caption；emit_warning 是诊断推送。 _TEXT_METHODS = ( "send", "edit_message", "send_draft", "send_stream_frame", "send_private_notice", "send_slash_confirm", "send_clarify", "send_exec_approval", "retire_clarify_card", "send_image", "send_image_file", "send_animation", "send_voice", "send_video", "send_document", "send_multiple_images", "emit_warning", "emit_media_warning", "send_final_ledgered", "_send_final_text", "_send_plain_fallback", "_send_with_retry", ) # 参数名命中这些就替换（一个方法可能有多个，比如 send_slash_confirm 的 title+message） _TEXT_PARAMS = ( "content", "text", "message", "caption", "title", "notice", "question", "text_content", "prompt", ) # 这些方法「返回值」才是要发出去的文本（文本在函数内部拼，包参数没用） _FORMAT_METHODS = ("_format_exec_approval", "_format_clarify_text") def _flag(*names: str) -> bool: for name in names: if os.environ.get(name, "").strip().lower() in _TRUTHY: return True return False def _disabled() -> bool: return _flag("PLUGIN_TIPS_CN_DISABLE", "CN_VOICE_DISABLE") # ────────────────────────── 词库 ────────────────────────── def _load_categories() -> List[Dict[str, Any]]: """读词库并预编译正则；坏规则只跳过自己，不拖垮插件。""" try: with open(_PHRASES, encoding="utf-8") as fh: data = json.load(fh) except Exception as exc: # pragma: no cover - 配置问题 logger.warning("plugin-tips-cn: 词库读取失败 %s: %s", _PHRASES, exc) return [] cats: List[Dict[str, Any]] = [] for raw in data.get("categories", []) or []: pats = [] for pat in raw.get("patterns", []) or []: try: pats.append(re.compile(pat, re.IGNORECASE)) except re.error as exc: logger.warning("plugin-tips-cn: 跳过非法正则 %r (%s): %s", pat, raw.get("id"), exc) variants = [v for v in (raw.get("variants") or []) if isinstance(v, str) and v.strip()] if pats and variants: cats.append({"id": raw.get("id", "?"), "pats": pats, "variants": variants}) return cats _CATEGORIES: List[Dict[str, Any]] = _load_categories() _CATS_MTIME: float = 0.0 try: _CATS_MTIME = _PHRASES.stat().st_mtime except OSError: pass def _categories() -> List[Dict[str, Any]]: """词库改了就热加载（一次 stat，开销可忽略），省得每改一句话都重启网关。""" global _CATEGORIES, _CATS_MTIME try: mtime = _PHRASES.stat().st_mtime except OSError: return _CATEGORIES if mtime != _CATS_MTIME: _CATEGORIES = _load_categories() _CATS_MTIME = mtime logger.info("plugin-tips-cn: 词库已热加载，共 %d 类", len(_CATEGORIES)) return _CATEGORIES def localize(text: str) -> Tuple[str, List[str]]: """返回 (替换后的文本, 命中的类别 id 列表)。没命中就原样返回。""" if not isinstance(text, str) or not text or _disabled(): return text, [] if len(text) > _MAX_LEN: return text, [] hits: List[str] = [] out = text for cat in _categories(): variants = cat["variants"] def _replace(match: "re.Match[str]", _variants: List[str] = variants, _cid: str = cat["id"]) -> str: phrase = random.choice(_variants) groups = {k: v for k, v in (match.groupdict() or {}).items() if v is not None} if groups: try: phrase = phrase.format(**groups) except (KeyError, IndexError, ValueError): pass hits.append(_cid) return phrase for pat in cat["pats"]: out = pat.sub(_replace, out) if hits and _flag("PLUGIN_TIPS_CN_LOG", "CN_VOICE_LOG"): logger.info("plugin-tips-cn: 命中 %s", ", ".join(sorted(set(hits)))) return out, hits def _loc(value: Any) -> Any: """只处理字符串；命中才换。""" if isinstance(value, str): new, hits = localize(value) return new if hits else value return value # ─────────────────── 第三层：适配器出口包装 ─────────────────── def _wrap_method(cls: type, name: str) -> int: """包住 cls 自己的 name 方法，把它的**所有**文本参数过一遍词库。幂等。""" fn = cls.__dict__.get(name) if fn is None or getattr(fn, "_tips_cn_wrapped", False): return 0 if not inspect.iscoroutinefunction(fn): return 0 try: params = list(inspect.signature(fn).parameters)[1:] # 去掉 self except (TypeError, ValueError): return 0 # 一个方法可能不止一个文本参数（send_slash_confirm 的 title + message） names = {i: p for i, p in enumerate(params) if p in _TEXT_PARAMS} if not names: return 0 async def wrapper(self, *args, **kwargs): try: a = list(args) for i, pname in names.items(): if len(a) > i: a[i] = _loc(a[i]) elif pname in kwargs: kwargs[pname] = _loc(kwargs[pname]) args = tuple(a) except Exception: pass # 替换失败绝不拦发送 return await fn(self, *args, **kwargs) wrapper._tips_cn_wrapped = True try: wrapper.__name__ = getattr(fn, "__name__", name) wrapper.__doc__ = getattr(fn, "__doc__", None) except Exception: pass setattr(cls, name, wrapper) return 1 def _wrap_formatter(cls: type, name: str) -> int: """包住「返回值就是文本」的方法：审批提示这类文案是在函数内部拼的，包参数没用。""" fn = cls.__dict__.get(name) if fn is None or getattr(fn, "_tips_cn_wrapped", False): return 0 if inspect.iscoroutinefunction(fn): return 0 def wrapper(*args, **kwargs): out = fn(*args, **kwargs) try: return _loc(out) except Exception: return out wrapper._tips_cn_wrapped = True try: wrapper.__name__ = getattr(fn, "__name__", name) wrapper.__doc__ = getattr(fn, "__doc__", None) except Exception: pass setattr(cls, name, wrapper) return 1 def patch_adapters() -> int: """把出口焊死：所有平台适配器（含子类）的文本发送方法都过一遍词库。幂等，可反复调。""" if _disabled() or _flag("PLUGIN_TIPS_CN_NO_SEND"): return 0 try: from gateway.platforms.base import BasePlatformAdapter except Exception as exc: logger.debug("plugin-tips-cn: 拿不到 BasePlatformAdapter（%s）", exc) return 0 total = 0 seen, stack = set(), [BasePlatformAdapter] while stack: cls = stack.pop() if cls in seen: continue seen.add(cls) for name in _TEXT_METHODS: try: total += _wrap_method(cls, name) except Exception as exc: logger.debug("plugin-tips-cn: 包 %s.%s 失败：%s", cls.__name__, name, exc) for name in _FORMAT_METHODS: try: total += _wrap_formatter(cls, name) except Exception as exc: logger.debug("plugin-tips-cn: 包 %s.%s 失败：%s", cls.__name__, name, exc) try: stack.extend(cls.__subclasses__()) except Exception: pass if total: logger.info("plugin-tips-cn: 已包住 %d 个适配器出站方法（覆盖 %d 个类）", total, len(seen)) return total # ────────────────────────── 钩子 ────────────────────────── def _on_transform_tool_result(tool_name: str = "", result: Any = None, **_: Any) -> Optional[str]: if not isinstance(result, str): return None new, hits = localize(result) return new if hits else None def _on_transform_llm_output(response_text: Any = None, **_: Any) -> Optional[str]: if not isinstance(response_text, str): return None new, hits = localize(response_text) return new if hits else None def _on_pre_gateway_dispatch(**_: Any): """每条入站消息都顺手补一次适配器包装（适配器可能是插件加载之后才导入的）。""" try: patch_adapters() except Exception: pass return None def register(ctx) -> None: ctx.register_hook("transform_tool_result", _on_transform_tool_result) ctx.register_hook("transform_llm_output", _on_transform_llm_output) ctx.register_hook("pre_gateway_dispatch", _on_pre_gateway_dispatch) try: patch_adapters() except Exception as exc: logger.warning("plugin-tips-cn: 适配器出口包装失败：%s", exc)