ykxmz/_build/patch_features_wire.py

# -*- coding: utf-8 -*-
"""把功能注册表接到服务端。

  * 新增 feature_list / feature_run 两条命令
  * RemoteServer 里挂一个 Features 实例
  * 顺手补一个洞：_resolve_received 原来接受**任意绝对路径**，
    等于能拉取电脑上任何文件。现在限定在接收目录 + 允许的用户目录内。
"""
from __future__ import annotations

import sys
from pathlib import Path

ROOT = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote\windows\lanremote")
SERVER = ROOT / "server.py"
PROTOCOL = ROOT / "protocol.py"

PROTO_OLD = 'CMD_SAY_CLIP = "say_clip"               # 朗读剪贴板\n'
PROTO_NEW = ('CMD_SAY_CLIP = "say_clip"               # 朗读剪贴板\n'
             'CMD_FEATURE_LIST = "feature_list"       # 电脑端能做哪些功能\n'
             'CMD_FEATURE_RUN = "feature_run"         # 执行某个功能\n')

IMPORT_OLD = "from .discovery import DISCOVERY_PORT, DiscoveryResponder\n"
IMPORT_NEW = ("from .discovery import DISCOVERY_PORT, DiscoveryResponder\n"
              "from .features import Features\n")

INIT_OLD = "        self._total_connections = 0\n"
INIT_NEW = ("        self._total_connections = 0\n"
            "        # 功能注册表（夜间模式 / 亮度 / 文件浏览…）\n"
            "        self.features: Features | None = None\n")

# 在 actions 赋值处挂上 features —— 找 server 里 actions 的设置点
ACTIONS_ANCHOR = "        self.actions = actions\n"
ACTIONS_NEW = ("        self.actions = actions\n"
               "        self.features = Features(actions) if actions else None\n")

CMD_ANCHOR = '        if cmd == P.CMD_SET_VOICE:'
CMD_NEW = '''        if cmd == P.CMD_FEATURE_LIST:
            if self.features is None:
                raise ActionError("功能不可用")
            return {"ok": True, "features": self.features.defs(),
                    "extra_apps": self.features.extra_apps(),
                    "extra_commands": self.features.extra_commands(),
                    "roots": [str(p) for p in self.features.allowed_roots()]}

        if cmd == P.CMD_FEATURE_RUN:
            if self.features is None:
                raise ActionError("功能不可用")
            name = str(msg.get("name") or "")
            params = msg.get("params") or {}
            if not isinstance(params, dict):
                params = {}
            try:
                note = self.features.run(name, params)
            except ValueError as exc:
                raise ActionError(str(exc)) from exc
            self._log(f"[功能] {name} -> {note}")
            return {"ok": True, "note": str(note)[:4000]}

        if cmd == P.CMD_SET_VOICE:'''

# 收紧 _resolve_received 的绝对路径
RESOLVE_OLD = '''        raw = Path(path)
        if raw.is_absolute() and raw.is_file():
            return raw
'''
RESOLVE_NEW = '''        raw = Path(path)
        if raw.is_absolute() and raw.is_file():
            # 只允许拉取「接收目录」或「用户自己的目录」里的文件。
            # 否则这条命令等于"能下载电脑上任意文件"，是个大洞。
            try:
                resolved = raw.resolve()
            except OSError:
                return None
            allowed = [self.store.received_dir()]
            if self.features is not None:
                allowed.extend(self.features.allowed_roots())
            for root in allowed:
                try:
                    resolved.relative_to(root.resolve())
                    return resolved
                except (ValueError, OSError):
                    continue
            self._log(f"[文件] 拒绝拉取允许范围外的路径: {raw}")
            return None
'''


def main() -> int:
    ok = True

    text = PROTOCOL.read_text(encoding="utf-8")
    if PROTO_OLD not in text:
        print("  [失败] protocol.py 锚点未匹配")
        ok = False
    else:
        text = text.replace(PROTO_OLD, PROTO_NEW, 1)
        PROTOCOL.write_text(text, encoding="utf-8")
        print("  protocol.py: 新增 2 条命令")

    stext = SERVER.read_text(encoding="utf-8")
    for old, new, tag in ((IMPORT_OLD, IMPORT_NEW, "导入"),
                          (INIT_OLD, INIT_NEW, "初始化"),
                          (CMD_ANCHOR, CMD_NEW, "命令处理"),
                          (RESOLVE_OLD, RESOLVE_NEW, "收紧文件路径")):
        if old not in stext:
            print(f"  [失败] server.py {tag} 锚点未匹配")
            ok = False
        else:
            stext = stext.replace(old, new, 1)
            print(f"  server.py / {tag}: OK")

    if ACTIONS_ANCHOR in stext:
        stext = stext.replace(ACTIONS_ANCHOR, ACTIONS_NEW, 1)
        print("  server.py / 挂载 Features: OK")
    else:
        print("  [警告] 没找到 actions 赋值点，features 需要在 run_agent 里挂")

    if ok:
        SERVER.write_text(stext, encoding="utf-8")

    print("\n语法检查：")
    import ast
    for name, path in (("protocol.py", PROTOCOL), ("server.py", SERVER)):
        try:
            ast.parse(path.read_text(encoding="utf-8"))
            print(f"  {name}: OK")
        except SyntaxError as exc:
            print(f"  {name}: 语法错误 {exc}")
            ok = False
    return 0 if ok else 1


if __name__ == "__main__":
    sys.exit(main())