# -*- coding: utf-8 -*- """Bug 1：配对码被占用时，要求在**装防火墙之前**就发现。 用户原话： "如果配对码被占用了，你输入进去以后你不应该是先检查没有占用， 然后有占用去改，然后再去放行防火墙吗？你现在如果占用的话，你改不了" 现在的流程（错）： 设置窗口 -> 存码 -> 开自启 -> 修防火墙 -> 静默运行 ^ ^ 这里没检查 等转静默才发现码被占， 但防火墙已经改了，用户也没法改码 要改成： 设置窗口 -> 检查码 -> 可用就继续；被占就弹框让他换，重新开设置窗口 -> 通过之后才 开自启 / 修防火墙 / 静默 关键点：**检查不能真的去注册那个码** —— 注册就等于占用了它。 所以给中转加一个只读接口： GET /v1/code/check?code=123456 -> {"ok": true, "taken": false} -> {"ok": true, "taken": true, "deviceName": "某台电脑"} """ from __future__ import annotations import sys from pathlib import Path ROOT = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote") RELAY_SERVER = ROOT / "relay" / "relay.py" RELAY_CLIENT = ROOT / "windows" / "lanremote" / "relay.py" RUN = ROOT / "windows" / "run_agent.py" # ================= 1. 中转：只读查询接口 ================= SRV_ANCHOR = ''' if path in ("/v1/health", "/health"):''' SRV_NEW = ''' # 只读查询：这个配对码有没有被别的电脑占？ # 安装时要"先检查再决定要不要改防火墙"，而检查不能真去注册 # （一注册就等于自己占了它）。所以单独开一个只读接口。 if path in ("/v1/code/check", "/v1/check"): query = urllib.parse.urlparse(self.path).query wanted = (urllib.parse.parse_qs(query).get("code") or [""])[0] wanted = wanted.strip() if not CODE_RE.match(wanted): self._json(400, {"ok": False, "error": "配对码要是 6 位数字"}) return with RELAY.lock: owner = RELAY.codes.get(wanted) device = RELAY.devices.get(owner) if owner else None self._json(200, { "ok": True, "code": wanted, "taken": bool(owner), "deviceName": device.name if device else "", "online": bool(device and device.online()), }) return if path in ("/v1/health", "/health"):''' # ================= 2. 被控端：调用这个接口 ================= CLIENT_ANCHOR = ''' def _ensure_device_id(self) -> str:''' CLIENT_NEW = ''' def _check_code(self, code: str) -> dict: """问中转：这个码被占了吗？（只读，不会占用它） 返回 {"ok": True, "taken": bool, "deviceName": str} 连不上中转就返回 {"ok": False}，调用方自己决定怎么办。 """ bases = list(self.candidates) or ([self.url] if self.url else []) if not bases: return {"ok": False, "error": "还没配置中转地址"} quoted = urllib.parse.quote(code) for base in bases: try: request = urllib.request.Request( f"{base}/v1/code/check?code={quoted}", headers={"User-Agent": USER_AGENT}) context = (_ssl_context() if base.startswith("https://") else None) with urllib.request.urlopen(request, timeout=10, context=context) as response: return json.loads(response.read(4096).decode( "utf-8", "replace")) except Exception as exc: # noqa: BLE001 last = f"{type(exc).__name__}: {exc}" continue return {"ok": False, "error": last if "last" in dir() else "连不上中转"} def _ensure_device_id(self) -> str:''' # ================= 3. run_setup：先检查再改防火墙 ================= RUN_OLD = ''' result = SetupWindow(store).show() if result is None: return 0 # 用户关掉了窗口 store.set_pairing_code(result["code"]) store.set("setup_done", True)''' RUN_NEW = ''' # 先检查配对码能不能用，**通过了才去改防火墙**。 # # 踩过的坑：以前是先存码、开自启、修防火墙，最后才转静默—— # 而"码被别的电脑占了"这件事要等转静默之后才发现。 # 那时防火墙已经改完了，用户在设置窗口里也没机会换码。 result = None while True: window = SetupWindow(store).show() if window is None: return 0 # 用户关掉了窗口（放弃安装） code = window.get("code", "") problem = check_code_available(store, code) if problem is None: result = window break # 被占了：说清楚，让他回去换一个 _tell_code_problem(problem) store.set_pairing_code(result["code"]) store.set("setup_done", True)''' # 在 run_setup 前面插入两个辅助函数 HELPER_ANCHOR = '''def run_setup(store) -> int:''' HELPER_NEW = '''def check_code_available(store, code: str): """这个配对码能用吗？ 能用（或没法判断）返回 None；被别的电脑占了返回一段说明文字。 检查走中转的只读接口 /v1/code/check —— 不注册，所以不会误占。 连不上中转时返回 None（放行）：不能因为服务器连不上就装不了， 真有问题等转静默之后还会再提示一次。 """ if not code or len(code) != 6 or not code.isdigit(): return "配对码必须是 6 位数字" target = str(store.get("relay_url") or "") or DEFAULT_RELAY_URL if not target.strip(): return None # 没配中转，没法检查 try: from lanremote.relay import RelayLink link = RelayLink(store, log=lambda _m: None) # start() 会去连中转并登记 —— 但我们只用它的 candidate 解析， # 不用它登记（登记会占用配对码）。 link.candidates = [] for piece in target.replace(",", ";").split(";"): piece = piece.strip().rstrip("/") if not piece: continue if not piece.startswith(("http://", "https://")): from lanremote.relay import _guess_scheme piece = _guess_scheme(piece) + piece link.candidates.append(piece) link.url = link.candidates[0] if link.candidates else "" reply = link._check_code(code) except Exception: # noqa: BLE001 return None # 检查失败就当没问题 if not reply.get("ok"): return None # 服务器没回应，放行 if not reply.get("taken"): return None # 没人用，可以用 where = reply.get("deviceName") or "另一台电脑" return ( f"配对码 {code} 已经被「{where}」占用了。\\n\\n" f"一个配对码同时只能给一台电脑用。\\n" f"请换一个别的 6 位数字。\\n\\n" f"（想继续用这个码的话，可以去中转管理页把占用它的那台设备删掉）" ) def _tell_code_problem(message: str) -> None: """把配对码的问题告诉用户。有界面就弹窗，没有就打印。""" try: import tkinter as tk from tkinter import messagebox root = tk.Tk() root.withdraw() root.attributes("-topmost", True) messagebox.showerror("配对码不能用", message) root.destroy() return except Exception: # noqa: BLE001 pass print("=" * 56) print("配对码不能用") print(message) print("=" * 56) def run_setup(store) -> int:''' def main() -> int: ok = True # ---- 中转 ---- text = RELAY_SERVER.read_text(encoding="utf-8") if "/v1/code/check" in text: print(" 中转：已经有 /v1/code/check") elif SRV_ANCHOR in text: text = text.replace(SRV_ANCHOR, SRV_NEW, 1) RELAY_SERVER.write_text(text, encoding="utf-8", newline="\n") print(" 中转：加了只读查询 /v1/code/check") else: print(" [失败] 中转锚点没找到") ok = False # 确认 urllib.parse 已导入 if "import urllib.parse" not in text: text = text.replace("import urllib.request", "import urllib.parse\nimport urllib.request", 1) RELAY_SERVER.write_text(text, encoding="utf-8", newline="\n") print(" 中转：补了 import urllib.parse") # ---- 被控端 ---- text = RELAY_CLIENT.read_text(encoding="utf-8") if "_check_code" in text: print(" 被控端：已经有 _check_code") elif CLIENT_ANCHOR in text: text = text.replace(CLIENT_ANCHOR, CLIENT_NEW, 1) RELAY_CLIENT.write_text(text, encoding="utf-8", newline="\n") print(" 被控端：加了 _check_code()") else: print(" [失败] 被控端锚点没找到") ok = False if "import urllib.parse" not in text: text = text.replace("import urllib.request", "import urllib.parse\nimport urllib.request", 1) RELAY_CLIENT.write_text(text, encoding="utf-8", newline="\n") print(" 被控端：补了 import urllib.parse") # ---- run_agent ---- text = RUN.read_text(encoding="utf-8") if "check_code_available" in text: print(" run_agent：已经有检查逻辑") else: if HELPER_ANCHOR in text: text = text.replace(HELPER_ANCHOR, HELPER_NEW, 1) print(" run_agent：加了 check_code_available / _tell_code_problem") else: print(" [失败] run_agent 锚点没找到") ok = False if RUN_OLD in text: text = text.replace(RUN_OLD, RUN_NEW, 1) print(" run_agent：run_setup 改成先检查再改防火墙") else: print(" [失败] 找不到 run_setup 的那一段") ok = False RUN.write_text(text, encoding="utf-8", newline="\n") # ---- 语法 ---- import ast for label, path in (("中转", RELAY_SERVER), ("被控端", RELAY_CLIENT), ("run_agent", RUN)): try: ast.parse(path.read_text(encoding="utf-8")) print(f" {label} 语法 OK") except SyntaxError as exc: print(f" {label} 语法错误: 行 {exc.lineno}: {exc.msg}") ok = False return 0 if ok else 1 if __name__ == "__main__": sys.exit(main())