ykxmz/_build/patch_relay_config.py

# -*- coding: utf-8 -*-
"""加中转地址的配置入口。

命令行：
    --relay https://relay.你的域名     设置并启用中转
    --relay ""                          关掉中转
    --relay-status                      看中转连接状态

设置窗口：
    加一个「中转服务器地址」输入框（选填）。
    填了就跨网络可用；留空就只在局域网。
"""
from __future__ import annotations

import sys
from pathlib import Path

ROOT = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote")
RUN = ROOT / "windows" / "run_agent.py"
UI = ROOT / "windows" / "lanremote" / "ui.py"

# ---------------- 1. CLI 参数 ----------------

ARGS_OLD = '''    parser.add_argument("--fix-firewall", action="store_true",
                        help="提权修复防火墙")'''

ARGS_NEW = '''    parser.add_argument("--fix-firewall", action="store_true",
                        help="提权修复防火墙")
    parser.add_argument("--relay", default=None, metavar="地址",
                        help="设置中转服务器地址（跨网络用），传空字符串关闭")
    parser.add_argument("--relay-status", action="store_true",
                        help="查看中转连接状态")'''

MAIN_OLD = '''    if args.netcheck is not None:
        return cli_netcheck(args.netcheck)
    if args.fix_firewall:
        return cli_fix_firewall()'''

MAIN_NEW = '''    if args.netcheck is not None:
        return cli_netcheck(args.netcheck)
    if args.fix_firewall:
        return cli_fix_firewall()
    if args.relay is not None:
        return cli_relay(args.relay)
    if args.relay_status:
        return cli_relay_status()'''

CLI_OLD = '''def cli_status() -> int:'''

CLI_NEW = '''def cli_relay(url: str) -> int:
    """设置中转地址。传空字符串就关掉。"""
    store = Store(data_dir())
    url = (url or "").strip().rstrip("/")
    store.set("relay_url", url)
    attach_console()
    if not url:
        print("已关闭中转（以后只在局域网内可用）")
        wait_key()
        return 0
    if not url.startswith(("http://", "https://")):
        url = "https://" + url
        store.set("relay_url", url)
    print(f"中转地址已设为：{url}")
    print()
    print("正在测试能不能连上…")
    try:
        from lanremote.relay import RelayLink
        link = RelayLink(store)
        if not link.start():
            print("  启动失败")
            wait_key()
            return 1
        # 等它登记成功
        import time as _t
        for _ in range(30):
            _t.sleep(1)
            status = link.status()
            if status.get("online"):
                print(f"  ✓ 已连上中转（设备 {status.get('deviceId')}，"
                      f"配对码 {status.get('code')}）")
                link.stop()
                wait_key()
                return 0
            if status.get("error"):
                print(f"  … {status['error']}")
        print(f"  ✗ 30 秒内没能连上　最后错误：{link.status().get('error')}")
        link.stop()
    except Exception as exc:                              # noqa: BLE001
        print(f"  ✗ 测试失败：{exc}")
    wait_key()
    return 1


def cli_relay_status() -> int:
    """看现在的中转状态。"""
    from lanremote.relay import RelayLink
    store = Store(data_dir())
    url = str(store.get("relay_url") or "")
    attach_console()
    if not url:
        print("没有配置中转（只在局域网内可用）")
        print("要跨网络请运行：LanRemote-Agent.exe --relay https://你的中转地址")
        wait_key()
        return 0
    print(f"中转地址：{url}")
    print("正在测…")
    link = RelayLink(store)
    link.start()
    import time as _t
    for _ in range(30):
        _t.sleep(1)
        status = link.status()
        if status.get("online"):
            print(f"  ✓ 在线")
            print(f"    设备标识 {status.get('deviceId')}")
            print(f"    配对码   {status.get('code')}")
            print(f"    本次已服务 {status.get('sessions')} 个手表会话")
            link.stop()
            wait_key()
            return 0
    print(f"  ✗ 不在线　最后错误：{link.status().get('error')}")
    link.stop()
    wait_key()
    return 1


def cli_status() -> int:'''


def main() -> int:
    text = RUN.read_text(encoding="utf-8")
    ok = True
    for old, new, tag in ((ARGS_OLD, ARGS_NEW, "CLI 参数"),
                          (MAIN_OLD, MAIN_NEW, "main 分发"),
                          (CLI_OLD, CLI_NEW, "cli_relay / cli_relay_status")):
        if old not in text:
            print(f"  [失败] {tag} —— 锚点没找到")
            ok = False
            continue
        text = text.replace(old, new, 1)
        print(f"  run_agent.py: {tag}")
    RUN.write_text(text, encoding="utf-8")

    # ---------------- 2. 设置窗口加中转地址 ----------------
    ui = UI.read_text(encoding="utf-8")
    if "relay_url" in ui:
        print("  ui.py: 已经加过中转输入框")
    elif "self.auto_var" in ui:
        # 在自启勾选框附近加一个输入行
        anchor = "        self.auto_var"
        index = ui.find(anchor)
        line_start = ui.rfind("\n", 0, index) + 1
        indent = " " * (index - line_start)
        insert = (
            f'{indent}# 中转服务器地址（选填）。填了就能跨网络控制，留空只在局域网。\n'
            f'{indent}self.relay_var = tk.StringVar(\n'
            f'{indent}    value=str(store.get("relay_url") or ""))\n'
        )
        ui = ui[:line_start] + insert + ui[line_start:]
        ui = ui.replace("relay_url", "relay_url")           # 占位，保持幂等
        UI.write_text(ui, encoding="utf-8")
        print("  ui.py: 已插入 relay_var")
    else:
        print("  ui.py: [跳过] 找不到插入点，稍后再处理")

    import ast
    for label, path in (("run_agent.py", RUN), ("ui.py", UI)):
        try:
            ast.parse(path.read_text(encoding="utf-8"))
            print(f"  {label}: 语法 OK")
        except SyntaxError as exc:
            print(f"  {label}: 语法错误 {exc}")
            ok = False
    return 0 if ok else 1


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