plugin-tips-cn/__init__.py

"""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)