ykxmz/_build/patch_selfcheck_request.py

# -*- coding: utf-8 -*-
"""手表上的「连不上？」-> 让电脑端弹出网络自检窗口。

为什么走 UDP 而不是 TCP：
    「连不上」的典型场景就是 TCP 8756 被防火墙拦掉，但 UDP 广播仍然通
    （这正是"能搜到但连不上"的原因）。所以用 UDP 广播一个请求，
    只要广播能到电脑，电脑就能把自检窗口弹出来。

改动：
  discovery.py —— 新增 SELFCHECK_MAGIC / ACK_MAGIC，响应器识别请求并回调
  server.py    —— 收到请求时启动网络自检（开一个独立控制台窗口）
"""
from __future__ import annotations

import sys
from pathlib import Path

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

# ------------------------------------------------------------- discovery.py

OLD_MAGIC = '''DISCOVERY_PORT = 8766
PROBE_MAGIC = "LANREMOTE_DISCOVER"
REPLY_MAGIC = "LANREMOTE_REPLY"
PROTOCOL = 1
'''
NEW_MAGIC = '''DISCOVERY_PORT = 8766
PROBE_MAGIC = "LANREMOTE_DISCOVER"
REPLY_MAGIC = "LANREMOTE_REPLY"
# 手表点「连不上？」时广播这个，电脑收到就弹出网络自检窗口
SELFCHECK_MAGIC = "LANREMOTE_SELFCHECK"
ACK_MAGIC = "LANREMOTE_ACK"
PROTOCOL = 1
'''

OLD_HELPERS = '''def parse_probe(data: bytes) -> dict | None:'''
NEW_HELPERS = '''def build_selfcheck_ack(name: str) -> bytes:
    payload = {"lanremote": ACK_MAGIC, "name": name}
    return json.dumps(payload, ensure_ascii=False).encode("utf-8")


def build_selfcheck_request() -> bytes:
    payload = {"lanremote": SELFCHECK_MAGIC, "proto": PROTOCOL}
    return json.dumps(payload, ensure_ascii=False).encode("utf-8")


def parse_selfcheck(data: bytes) -> dict | None:
    try:
        obj = json.loads(data.decode("utf-8", errors="replace"))
    except ValueError:
        return None
    if not isinstance(obj, dict):
        return None
    if obj.get("lanremote") != SELFCHECK_MAGIC:
        return None
    return obj


def parse_probe(data: bytes) -> dict | None:'''

OLD_INIT = '''        self.on_probe = on_probe          # 可选回调，用于界面提示"有人在找这台电脑"
'''
NEW_INIT = '''        self.on_probe = on_probe          # 可选回调，用于界面提示"有人在找这台电脑"
        # 收到「请打开网络自检」的请求时调用；由上层决定怎么弹窗口
        self.on_selfcheck = None
        self.selfcheck_requests = 0
'''

OLD_LOOP = '''            probe = parse_probe(data)
            if probe is None:
                continue
'''
NEW_LOOP = '''            # 先看是不是「请打开网络自检」的请求
            if parse_selfcheck(data) is not None:
                self.selfcheck_requests += 1
                try:
                    self._sock.sendto(
                        build_selfcheck_ack(str(self.store.get("server_name", ""))),
                        addr)
                except OSError:
                    pass
                if self.on_selfcheck:
                    try:
                        self.on_selfcheck(addr[0])
                    except Exception:
                        pass
                continue

            probe = parse_probe(data)
            if probe is None:
                continue
'''

OLD_STATUS = '''            "answered": self.probes_answered,
            "last_probe": self.last_probe_at,
            "error": self.last_error,
'''
NEW_STATUS = '''            "answered": self.probes_answered,
            "last_probe": self.last_probe_at,
            "selfcheck_requests": self.selfcheck_requests,
            "error": self.last_error,
'''

# ---------------------------------------------------------------- server.py

OLD_START = '''        if self._discovery is not None:
            self._discovery.stop()
            self._discovery = None
'''
NEW_START = '''        if self._discovery is not None:
            self._discovery.stop()
            self._discovery = None
'''

OLD_DISCOVERY_INIT = '''            self._discovery = DiscoveryResponder('''
NEW_DISCOVERY_INIT = '''            self._discovery = DiscoveryResponder('''

# 在 start() 里挂上 on_selfcheck
OLD_ON_PROBE = '''        self._discovery.on_probe = self._on_probe'''
NEW_ON_PROBE = '''        self._discovery.on_probe = self._on_probe
        self._discovery.on_selfcheck = self._on_selfcheck_request'''


def main() -> int:
    text = DISCOVERY.read_text(encoding="utf-8")
    ok = True
    for old, new, tag in ((OLD_MAGIC, NEW_MAGIC, "常量"),
                          (OLD_HELPERS, NEW_HELPERS, "构造/解析"),
                          (OLD_INIT, NEW_INIT, "初始化"),
                          (OLD_LOOP, NEW_LOOP, "主循环"),
                          (OLD_STATUS, NEW_STATUS, "状态")):
        if old not in text:
            print(f"  [失败] discovery.py {tag} 锚点未匹配")
            ok = False
        else:
            text = text.replace(old, new, 1)
            print(f"  discovery.py / {tag}: OK")
    if ok:
        DISCOVERY.write_text(text, encoding="utf-8")

    text = SERVER.read_text(encoding="utf-8")
    if OLD_ON_PROBE not in text:
        print("  [失败] server.py on_probe 锚点未匹配")
        ok = False
    else:
        text = text.replace(OLD_ON_PROBE, NEW_ON_PROBE, 1)
        print("  server.py / 挂载回调: OK")

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

    print("\n语法检查：")
    import ast
    for name, path in (("discovery.py", DISCOVERY), ("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())