# -*- coding: utf-8 -*- """被控端 TCP 服务：配对、鉴权、命令分发、文件收发。""" from __future__ import annotations import os import re import secrets import socket import threading import time from pathlib import Path import tempfile import sys from . import protocol as P from . import updater as U from . import input as IN from .actions import ActionError from .config import Store from .discovery import DISCOVERY_PORT, DiscoveryResponder from .features import Features from .screen import FrameStreamer, ScreenCapturer, ScreenError def _truthy(value) -> bool: """字符串 "false"/"0"/"off" 不能被当成真（bool("false") 是 True）。""" if isinstance(value, bool): return value if isinstance(value, (int, float)): return value != 0 if isinstance(value, str): return value.strip().lower() in ("1", "true", "on", "yes", "是", "开") return bool(value) class Session: """一条客户端连接。""" def __init__(self, conn: socket.socket, addr, server: "RemoteServer"): self.conn = conn self.addr = addr self.server = server self.device_id = "" self.device_name = "" self.authed = False self.wants_screen = False # 是否正在观看屏幕镜像 self.connected_at = time.time() self.last_seen = time.time() self.buffers = { P.CMD_IMAGE: P.ChunkAssembler(), P.CMD_AUDIO: P.ChunkAssembler(), P.CMD_FILE_PUSH: P.ChunkAssembler(), } self._send_lock = threading.Lock() def send(self, obj: dict) -> bool: try: with self._send_lock: self.conn.sendall(P.dumps(obj)) return True except OSError: return False def send_chunks(self, event: str, name: str, blob: bytes) -> bool: """把二进制分块发给客户端。""" for seq, last, piece in P.chunk_bytes(blob): ok = self.send({ "event": event, "name": name, "seq": seq, "last": last, "data": P.b64encode(piece), }) if not ok: return False return True def close(self) -> None: try: self.conn.shutdown(socket.SHUT_RDWR) except OSError: pass try: self.conn.close() except OSError: pass # 内置的中转地址：**默认留空**，由用户自己在设置里填。 # # 为什么不再内置一个： # 原来写死了一台随身 WiFi 的地址 + 一个公网隧道域名。问题是 # 那台设备的 IP 会变（DHCP），写死的地址很快就过期，用户开箱 # 就一直在连一个不存在的主机，表现成"一直转圈/连不上"。 # 与其猜一个迟早会错的地址，不如留空 —— 局域网本身完全不受影响。 # # 空字符串的含义：**只走局域网**。 # * UDP 广播发现、TCP 直连、配对、命令下发，全都照常工作 # * 不会去连任何中转，不报错、不卡住（RelayLink.start("") 会直接跳过） # * 用户想要跨网络时，在「中转设置」里填自己的地址即可 # * 多个地址可用分号隔开，按顺序尝试，例如： # http://192.168.1.10:8080;https://你自己的域名 DEFAULT_RELAY_URL = "" class RemoteServer: """局域网被控端服务。""" def __init__(self, store: Store, actions, ui=None): self.store = store self.actions = actions self.features = Features(actions) if actions else None self.ui = ui self.sessions: list[Session] = [] # 配对失败记录：{来源 IP: [时间戳]}，用于限速防枚举 self._pair_failures: dict = {} self._lock = threading.RLock() # 跨网络中转链路（配了 relay_url 才会启用） self._relay = None self._sock: socket.socket | None = None self._accept_thread: threading.Thread | None = None # 累计收到过多少次 TCP 连接 —— 自检用它判断"手表的包有没有到" self._total_connections = 0 self._clip_thread: threading.Thread | None = None self._running = False self._last_clip = "" self.pairing_enabled = True # 屏幕镜像 self._capturer: ScreenCapturer | None = None # 建采集器时用的 (分辨率, 画质)。变了就重建，不然改设置不生效。 self._capturer_signature = None self._streamer: FrameStreamer | None = None self._screen_indicator_on = False # 局域网自动发现（手表不用输 IP） self._discovery: DiscoveryResponder | None = None # ---------------- 生命周期 ---------------- @property def port(self) -> int: return int(self.store.get("port", P.DEFAULT_PORT)) def start(self) -> None: if self._running: return sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # Windows 上 SO_REUSEADDR 允许第二个进程抢占同一端口， # 会让"端口被占用"检测失效。改用独占绑定。 if hasattr(socket, "SO_EXCLUSIVEADDRUSE"): try: sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1) except OSError: pass else: sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind(("0.0.0.0", self.port)) sock.listen(16) sock.settimeout(1.0) self._sock = sock self._running = True # ── 先起局域网发现 ── # # 顺序很重要：enable_relay() 会挨个试中转地址， # 每个候选都要等到超时才算"不通"，地址越烂等得越久（实测 10 秒+）。 # 原来这一步排在发现服务前面，导致： # 电脑刚开机的十几秒里，手表广播搜不到它。 # 局域网发现是纯本机能力，不该被"连不连得上外网"拖住， # 所以挪到前面，开机即可被搜到。 self._discovery = DiscoveryResponder(self.store, self.port) # 手表点「连不上？」会广播请求，这里就把网络自检结论算出来 self._discovery.on_selfcheck = self._on_selfcheck_request if self._discovery.start(): self._log(f"自动发现已开启（UDP {DISCOVERY_PORT}）") else: self._log(f"自动发现未开启：{self._discovery.last_error}") # ── 再起收发线程 ── self._accept_thread = threading.Thread(target=self._accept_loop, name="accept", daemon=True) self._accept_thread.start() self._clip_thread = threading.Thread(target=self._clip_loop, name="clip", daemon=True) self._clip_thread.start() # ── 最后才连中转（会阻塞，所以放最后）── # 配了中转地址就同时启用跨网络通道（和局域网监听并存，不冲突） try: self.enable_relay() except Exception as exc: # noqa: BLE001 self._log(f"启用中转失败：{exc}") self._log(f"服务已启动，监听 0.0.0.0:{self.port}") def _selfcheck_report(self, from_ip: str) -> str: """生成一段给手表看的诊断结论（不弹任何窗口）。 用户要的是「电脑上什么都不出现」，所以结论直接回传给手表显示。 """ try: from .diagnostics import collect, verdict data = collect(self, self.store) _level, text = verdict(data) ips = data.get("ips") or [] port = self.port blocked = (data.get("firewall") or {}).get("blocked_count", 0) lines = [ f"电脑：{data.get('name') or '被控端'}", f"本机地址：{', '.join(ips[:2]) if ips else '读不到'}", f"监听端口：{port}", f"收到过的连接：{data.get('tcp_connections', 0)} 次", f"阻止规则：{blocked} 条", "", f"结论：{text}", ] return "\n".join(lines) except Exception as exc: # noqa: BLE001 return f"电脑端诊断失败：{exc}" def _on_selfcheck_request(self, from_ip: str) -> str: """手表说「我连不上」→ 生成诊断结论，随 ACK 回给手表。 走 UDP 是有意的：连不上的典型原因就是 TCP 被防火墙拦了， 而 UDP 广播往往还通（那正是"能搜到但连不上"的原因）。 注意：**不往电脑上弹窗口**。结论直接回传给手表， 用户看的就是手表，电脑上保持干净。 """ note = self._selfcheck_report(from_ip) first = note.splitlines()[0] if note else "" self._log(f"[自检] 收到来自 {from_ip} 的自检请求，已生成诊断结论（{first}）") return note def _open_selfcheck_window(self, from_ip: str) -> None: """有窗口的模式下才用得上（静默模式不会走到这里）。""" self._log(f"[自检] 收到来自 {from_ip} 的请求，已生成诊断结论") def launch() -> None: target = getattr(self, "ui", None) if target is not None and hasattr(target, "open_diagnostics"): try: target.ui_call(lambda: target.open_diagnostics(from_ip), timeout=3.0) return except Exception as exc: # noqa: BLE001 self._log(f"[自检] 打开界面失败：{exc}") return self._log("[自检] 没有可用的界面，跳过") # 弹窗口这件事要放到主线程做，避免和界面的消息循环打架 target = getattr(self, "ui", None) if target is not None and hasattr(target, "ui_call"): try: target.ui_call(launch, timeout=3.0) return except Exception: pass try: import threading threading.Thread(target=launch, name="selfcheck", daemon=True).start() except Exception: pass def stop(self) -> None: self._running = False self._stop_relay() self._stop_streamer() if self._discovery is not None: self._discovery.on_selfcheck = None self._discovery.stop() self._discovery = None with self._lock: sessions = list(self.sessions) self.sessions.clear() for session in sessions: session.close() if self._sock: try: self._sock.close() except OSError: pass self._sock = None self._log("服务已停止") def restart(self, new_port: int | None = None) -> None: self.stop() if new_port: self.store.set("port", int(new_port)) time.sleep(0.3) self.start() @property def running(self) -> bool: return self._running # ---------------- 中转（跨网络）---------------- def adopt_relay_socket(self, relay_socket) -> None: """把一条中转连接当成普通 TCP 连接接纳进来。 中转那边说"有手表接入了"，这里就造一个 Session 跑同样的 _client_loop —— 上面的协议、配对、鉴权一行都不用改。 """ address = ("中转服务器", 0) session = Session(relay_socket, address, self) with self._lock: self.sessions.append(session) self._log("设备接入（经中转，跨网络）") self._notify_ui() threading.Thread(target=self._client_loop, args=(session,), name="client-relay", daemon=True).start() def enable_relay(self, url: str = "") -> bool: """启用中转。地址优先用参数，其次读设置。""" from .relay import RelayLink target = (url or "").strip() or str(self.store.get("relay_url") or "") if not target.strip(): # 没配过就用内置默认值 —— 现在是空字符串，表示"只走局域网"。 # 这时候**什么都不做**是正确的：局域网监听已经起来了， # 不报错、不重试、不占用任何后台线程。 target = DEFAULT_RELAY_URL if not target.strip(): self._log("没配中转地址，只走局域网" "（需要跨网络时在设置里填中转地址）") # ⚠ 这里以前是 `if self._relay is not None: return True` —— # 只要对象建过就报成功。而"没配中转地址"那条路也会把对象建起来 # （start("") 返回 False），于是第二次调用会谎报中转在跑。 # 改成看**真实运行状态**。 if self._relay is not None: return bool(getattr(self._relay, "running", False)) self._relay = RelayLink(self.store, log=self._log, on_session=self.adopt_relay_socket, on_alert=self._alert) started = self._relay.start(target) if started: self._log(f"已启用中转：{self._relay.url} 配对码 {self.store.pairing_code()}") return started def _alert(self, title: str, text: str) -> None: """给用户看的通知（配对码被占这类必须让人知道的事）。 走界面的通知，用户在电脑上能直接看到。 界面还没起来时退回写日志，至少不丢信息。 """ self._log(f"[需要处理] {title}：{text}") try: ui = getattr(self.actions, "ui", None) if ui is not None and hasattr(ui, "notify"): ui.notify(title, text) except Exception: # noqa: BLE001 pass def relay_status(self) -> dict: if self._relay is None: return {"enabled": False} return self._relay.status() def _stop_relay(self) -> None: if self._relay is None: return try: self._relay.stop() except Exception: # noqa: BLE001 pass self._relay = None # ---------------- 接受连接 ---------------- def _accept_loop(self) -> None: while self._running: try: conn, addr = self._sock.accept() self._total_connections += 1 except socket.timeout: continue except OSError: break conn.settimeout(None) try: conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) except OSError: pass session = Session(conn, addr, self) with self._lock: self.sessions.append(session) self._log(f"设备接入 {addr[0]}:{addr[1]}") self._notify_ui() threading.Thread(target=self._client_loop, args=(session,), name=f"client-{addr[0]}", daemon=True).start() def _client_loop(self, session: Session) -> None: try: for line in self._read_lines(session.conn): session.last_seen = time.time() try: msg = P.loads(line) except ValueError: session.send({"ok": False, "error": "消息格式错误"}) continue if not isinstance(msg, dict): continue try: reply = self._dispatch(session, msg) except ActionError as exc: reply = {"ok": False, "error": str(exc)} except Exception as exc: # noqa: BLE001 reply = {"ok": False, "error": f"内部错误: {exc}"} self._log(f"[错误] {exc}") if reply is not None: # ⚠ 这里必须兜住**所有**异常。 # 以前只在外层捕获 OSError，而 session.send() 在 # 中转链路上会抛 RelayError（实测："请求失败: Idle"）—— # 那个异常一路冒出去，把 client-relay 线程整个打死。 # 线程一死，电脑就再也不往手表回任何东西， # 而 relay-link 还在长轮询、中转看着"设备在线"， # 用户看到的就是"连上了，按一下毫无反应"。 # 发失败最多踢掉这个会话让手表重连，绝不能丢线程。 try: session.send(reply) except Exception as exc: # noqa: BLE001 self._log(f"[错误] 回包失败，断开该会话：{exc}") break except OSError: pass except Exception as exc: # noqa: BLE001 self._log(f"[错误] 会话循环异常退出：{exc}") finally: with self._lock: if session in self.sessions: self.sessions.remove(session) session.close() session.wants_screen = False # 没有观看者了就立刻停掉抓屏，不留后台耗电/耗资源 if not any(s.wants_screen for s in self._authed_sessions()): self._stop_streamer() self._log(f"设备断开 {session.addr[0]}" f"{' / ' + session.device_name if session.device_name else ''}") self._notify_ui() @staticmethod def _read_lines(conn: socket.socket): buf = b"" while True: try: data = conn.recv(65536) except OSError: return if not data: return buf += data if len(buf) > P.MAX_LINE_BYTES and b"\n" not in buf: return while b"\n" in buf: line, buf = buf.split(b"\n", 1) if line.strip(): yield line # ---------------- 命令分发 ---------------- def _dispatch(self, session: Session, msg: dict) -> dict | None: cmd = str(msg.get("cmd") or "") # --- 握手 --- if cmd == P.CMD_PAIR: return self._do_pair(session, msg) if cmd == P.CMD_AUTH: return self._do_auth(session, msg) if not session.authed: return {"ok": False, "error": "未鉴权：请先配对或使用已保存的设备令牌"} if cmd == P.CMD_PING: return {"ok": True, "pong": True, "time": time.time()} if cmd == P.CMD_IMAGE: return self._recv_media(session, msg, P.CMD_IMAGE, "图片") if cmd == P.CMD_AUDIO: return self._recv_media(session, msg, P.CMD_AUDIO, "音频") if cmd == P.CMD_FILE_PUSH: return self._recv_media(session, msg, P.CMD_FILE_PUSH, "文件") if cmd == P.CMD_MEDIA: note = self.actions.media_control(str(msg.get("action") or "")) self._log(f"[媒体] {note}") return {"ok": True, "note": note} if cmd == P.CMD_CLIP_SET: note = self.actions.clip_set(str(msg.get("text") or "")) self._last_clip = str(msg.get("text") or "") self._log(f"[剪贴板] {note}") return {"ok": True, "note": note} if cmd == P.CMD_CLIP_GET: text = self.actions.clip_get() return {"ok": True, "text": text} if cmd == P.CMD_POWER: note = self.actions.power(str(msg.get("action") or "")) self._log(f"[电源] {note}") return {"ok": True, "note": note} if cmd == P.CMD_RUN: result = self.actions.run_command(str(msg.get("command") or "")) return {"ok": True, **result} if cmd == P.CMD_SAY: try: rate = int(msg.get("rate") or 0) except (TypeError, ValueError): rate = 0 note = self.actions.say(str(msg.get("text") or ""), str(msg.get("voice") or ""), rate) self._log(f"[播报] {note}") return {"ok": True, "note": note} if cmd == P.CMD_VOICE_LIST: voices = self.actions.list_voices() return {"ok": True, "voices": voices, "current": str(self.store.get("tts_voice") or "")} if cmd == P.CMD_SFX_LIST: items = self.actions.list_sfx() # 同时给一个扁平的 names —— 手表端的 stringList(reply, key) # 只认字符串数组，和 audio_list 的约定保持一致。 return {"ok": True, "sfx": items, "count": len(items), "names": [it.get("name", "") for it in items], "dir": self.actions.sfx_dir_hint()} if cmd == P.CMD_SFX_PLAY: name = str(msg.get("name") or "").strip() if not name: return {"ok": False, "error": "没说要放哪个音效"} try: msg = self.actions.play_sfx(name) except Exception as exc: return {"ok": False, "error": str(exc)} return {"ok": True, "message": msg, "name": name} if cmd == P.CMD_AUDIO_LIST: files = self.actions.list_local_audio() # 顺手给出扁平的 names/paths，手表端不用再解析嵌套结构 return {"ok": True, "files": files, "names": [f.get("name", "") for f in files], "paths": [f.get("path", "") for f in files], "roots": [str(p) for p in self.actions.audio_roots()]} if cmd == P.CMD_SHUTDOWN_TIMER: cancel = _truthy(msg.get("cancel")) try: minutes = int(msg.get("minutes") or 0) except (TypeError, ValueError): minutes = 0 note = self.actions.shutdown_timer(minutes, cancel) self._log(f"[电源] {note}") return {"ok": True, "note": note} if cmd == P.CMD_MONITOR: note = self.actions.monitor_power(_truthy(msg.get("on"))) self._log(f"[显示] {note}") return {"ok": True, "note": note} if cmd == P.CMD_OPEN_URL: note = self.actions.open_url(str(msg.get("url") or "")) self._log(f"[网址] {note}") return {"ok": True, "note": note} if cmd == P.CMD_QUICK: note = self.actions.quick_launch(str(msg.get("app") or "")) self._log(f"[快捷] {note}") return {"ok": True, "note": note} if cmd == P.CMD_OPEN_FOLDER: note = self.actions.open_folder(str(msg.get("folder") or "")) self._log(f"[文件夹] {note}") return {"ok": True, "note": note} if cmd == P.CMD_PROC_LIST: procs = self.actions.list_processes() return {"ok": True, "names": [f"{p['name']} ({p['mem_kb'] // 1024} MB)" for p in procs], "pids": [p["pid"] for p in procs]} if cmd == P.CMD_PROC_KILL: try: pid = int(msg.get("pid") or 0) except (TypeError, ValueError): pid = 0 note = self.actions.kill_process(pid, str(msg.get("name") or "")) self._log(f"[进程] {note}") return {"ok": True, "note": note} if cmd == P.CMD_BATTERY: info = self.actions.battery() if not info.get("present"): return {"ok": True, "note": info.get("note") or "没有电池（台式机）"} detail = f"{info['percent']}%" if info.get("charging"): detail += "（充电中）" return {"ok": True, "note": detail, **info} if cmd == P.CMD_DISK: drives = self.actions.disk() lines = [f"{d['drive']} {d['free_gb']}/{d['total_gb']} GB 可用" f"（已用 {d['used_percent']}%）" for d in drives] return {"ok": True, "note": " | ".join(lines) or "读不到磁盘", "drives": drives} if cmd == P.CMD_PERF: info = self.actions.perf() note = (f"CPU {info.get('cpu_percent')}% " f"内存 {info.get('mem_percent')}%" f"（{info.get('mem_used_gb')}/{info.get('mem_total_gb')} GB）") return {"ok": True, "note": note, **info} if cmd == P.CMD_CLIP_HISTORY: items = self.actions.clip_history() return {"ok": True, "note": f"共 {len(items)} 条", "texts": [i.get("text", "") for i in items], "times": [i.get("at", "") for i in items]} if cmd == P.CMD_SET_WALLPAPER: note = self.actions.set_wallpaper(str(msg.get("path") or "")) self._log(f"[壁纸] {note}") return {"ok": True, "note": note} if cmd == P.CMD_SAY_CLIP: note = self.actions.say_clipboard() self._log(f"[朗读] 剪贴板内容") return {"ok": True, "note": note} if cmd == P.CMD_AUDIO_PAUSE: note = self.actions.audio_pause() self._log(f"[音频] {note}") return {"ok": True, "note": note} if cmd == P.CMD_AUDIO_RESUME: note = self.actions.audio_resume() self._log(f"[音频] {note}") return {"ok": True, "note": note} if cmd == P.CMD_AUDIO_STATUS: return {"ok": True, "note": self.actions.audio_status()} 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: voice = str(msg.get("voice") or "").strip() self.store.set("tts_voice", voice) self._log(f"[音色] 已设为 {voice or '系统默认'}") return {"ok": True, "note": voice or "系统默认"} if cmd == P.CMD_PLAY_LOCAL: note = self.actions.play_local(str(msg.get("path") or "")) self._log(f"[本机播放] {note}") return {"ok": True, "note": note} if cmd == P.CMD_NOTIFY: note = self.actions.notify(str(msg.get("title") or ""), str(msg.get("text") or "")) return {"ok": True, "note": note} if cmd == P.CMD_FILE_LIST: return self._file_list() if cmd == "media_list": return self._media_list() if cmd == "media_pull": return self._media_pull(session, str(msg.get("path") or "")) if cmd == P.CMD_FILE_PULL: return self._file_pull(session, str(msg.get("path") or "")) if cmd == P.CMD_SCREEN_START: return self._screen_start(session, msg) if cmd == P.CMD_SCREEN_STOP: return self._screen_stop(session) if cmd == P.CMD_SCREENSHOT: return self._screenshot(session, msg) if cmd == P.CMD_INFO: return {"ok": True, "info": self.actions.info()} if cmd == P.CMD_UPDATE: return self._do_update(msg) if cmd == P.CMD_MOUSE: return self._do_mouse(msg) if cmd == P.CMD_SHOW_DESKTOP: try: IN.show_desktop() except IN.InputError as exc: return {"ok": False, "error": str(exc)} return {"ok": True, "done": "show_desktop"} return {"ok": False, "error": f"未知命令: {cmd}"} # ---------------- 远程鼠标 / 键盘注入 ---------------- def _do_mouse(self, msg: dict) -> dict: """处理 {"cmd":"mouse","action":...}。 坐标系是**屏幕像素**（相对虚拟屏幕左上角），由手表按比例映射好 再发过来 —— 电脑端不做缩放猜测，免得两边算法不一致导致漂移。 """ action = str(msg.get("action") or "").lower() try: if action == "move": # 分辨率可能被改过（投屏切换等），每次移动前刷一下缓存。 # 不刷的话，插上投影仪之后鼠标会跑到屏幕外。 IN.refresh_screen() IN.move_to(int(msg.get("x", 0)), int(msg.get("y", 0))) return {"ok": True} if action == "moveby": IN.move_by(int(msg.get("dx", 0)), int(msg.get("dy", 0))) return {"ok": True} if action == "click": IN.click(str(msg.get("button") or "left"), double=bool(msg.get("double"))) return {"ok": True} if action == "down": IN.down(str(msg.get("button") or "left")) return {"ok": True} if action == "up": IN.up(str(msg.get("button") or "left")) return {"ok": True} if action == "scroll": IN.scroll(int(msg.get("amount", 0)), horizontal=bool(msg.get("horizontal"))) return {"ok": True} if action == "desktop": IN.show_desktop() return {"ok": True} return {"ok": False, "error": "不认识的鼠标动作：%s" % action} except IN.InputError as exc: return {"ok": False, "error": str(exc)} except (TypeError, ValueError) as exc: return {"ok": False, "error": "参数不对：%s" % exc} # ---------------- 静默自动更新 ---------------- def _do_update(self, msg: dict) -> dict: """处理 {"cmd":"update","action":...}。 取包的地方是**用户自己配的中转站**（store 里的 relay_url）， 不是任何写死的公网地址 —— 这一点是刻意的，见 updater.py 的说明。 """ from . import __version__ as my_version action = str(msg.get("action") or "check").lower() relay = (self.store.relay_url() or "").strip() if action == "check": try: info = U.check(relay, my_version) except U.UpdateError as exc: return {"ok": False, "error": str(exc)} if not info.get("supported"): return {"ok": False, "error": info.get("error") or "不支持自动更新"} if "error" in info: return {"ok": False, "error": info["error"]} return { "ok": True, "current": my_version, "latest": info["latest"], "needUpdate": bool(info["needUpdate"]), "size": info.get("size", 0), "name": info.get("name", ""), } if action != "start": return {"ok": False, "error": "action 只能是 check 或 start"} # ---- 真的更新 ---- # 先查一遍（在调用方线程里查，失败能立刻报回去） try: info = U.check(relay, my_version) except U.UpdateError as exc: return {"ok": False, "error": str(exc)} if not info.get("supported") or "error" in info: return {"ok": False, "error": info.get("error") or "不支持自动更新"} if not info["needUpdate"]: return {"ok": True, "alreadyLatest": True, "current": my_version, "message": "电脑端已经是最新版本 %s" % my_version} # 下载和应用放到后台线程，先回一句让手表知道开始了 threading.Thread( target=self._run_update, args=(info, my_version), name="ykxmz-update", daemon=True).start() return { "ok": True, "started": True, "from": my_version, "to": info["latest"], "size": info.get("size", 0), "message": "正在下载 %s（%.1f MB），下完会自动装好并重启，" "设置和配对码都不变" % (info["latest"], (info.get("size") or 0) / 1048576.0), } def _run_update(self, info: dict, my_version: str) -> None: """后台线程：下载 -> 校验 -> 起帮手 -> 自己退出。""" def log(text: str) -> None: try: self._log(text) except Exception: # noqa: BLE001 pass try: log("[更新] 开始下载 %s（%.1f MB）" % (info["latest"], (info.get("size") or 0) / 1048576.0)) work = Path(tempfile.gettempdir()) / "ykxmz-update" last = [0] def progress(got: int, total: int) -> None: # 每 25% 报一次，别把日志刷爆 if total: pct = int(got * 100 / total) if pct - last[0] >= 25: last[0] = pct log("[更新] 已下载 %d%%（%.1f/%.1f MB）" % (pct, got / 1048576.0, total / 1048576.0)) pkg = U.download(info["url"], work, info.get("sha256", ""), info.get("size", 0), progress=progress) log("[更新] 下载完成并校验通过：%s" % pkg.name) # 应用：起分离的 .cmd 帮手，然后**本进程必须退出** app_exe = Path(sys.executable if getattr(sys, "frozen", False) else sys.argv[0]).resolve() logfile = U.apply_update(pkg, app_exe) log("[更新] 已交给更新帮手（%s），被控端即将退出并重启" % logfile) U.cleanup_old_downloads() # 给日志和网络一点时间落完，再退出 time.sleep(2) os._exit(0) except U.UpdateError as exc: log("[更新] 失败：%s（保持当前版本 %s）" % (exc, my_version)) except Exception as exc: # noqa: BLE001 log("[更新] 意外失败：%s: %s" % (type(exc).__name__, exc)) # ---------------- 握手实现 ---------------- def _do_pair(self, session: Session, msg: dict) -> dict: code = str(msg.get("code") or "").strip() expected = self.store.pairing_code() if not expected: return {"ok": False, "error": "被控端尚未生成配对码"} # 配对窗口关着、但对方报的码是对的 —— 照样放行。 # # 为什么：配对码本身就是密钥。老逻辑是"首台设备配对成功后 # 自动关闭配对窗口"（close_pair_after_first），结果手表重装、 # 换新表、或清掉 App 数据之后令牌没了，重新配对会被拒， # 用户看到"被控端已关闭新设备配对"就彻底连不上了 —— # 而这时候他能做的只有跑去管理页手动开窗口，完全不合理。 # # 码不对时才用"已关闭配对"这个说法挡回去， # 防暴力枚举的按 IP 限速（下面那段）仍然生效。 if not self.pairing_enabled and code != expected: return {"ok": False, "error": "被控端已关闭新设备配对" + "（配对码不对；要在管理页临时开启的话，" "打开中转站管理页找「允许新设备配对」）"} # 防暴力枚举：6 位数字只有 100 万种，不限速的话局域网里慢慢试就行。 # 按**来源 IP**分别计数 —— 全局计数的话，别人发 10 个错码 # 就能把你这个合法用户也锁在门外，那本身就是个拒绝服务的口子。 now = time.time() ip = session.addr[0] buckets = getattr(self, "_pair_failures", None) if not isinstance(buckets, dict): buckets = {} mine = [t for t in buckets.get(ip, []) if now - t < 300] buckets = {k: [t for t in v if now - t < 300] for k, v in buckets.items() if any(now - t < 300 for t in v)} buckets[ip] = mine self._pair_failures = buckets if len(mine) >= 10: wait = max(1, int(300 - (now - mine[0]))) self._log(f"[配对] {ip} 失败次数过多，锁定中（{wait} 秒）") return {"ok": False, "error": f"配对码错误次数过多，请等 {wait} 秒后再试"} # 用 compare_digest，避免按字符比较带来的时间侧信道 if not secrets.compare_digest(code, expected): mine.append(now) self._log(f"[配对失败] 来自 {ip}，配对码不匹配（{len(mine)}/10）") return {"ok": False, "error": "配对码错误"} self._pair_failures[ip] = [] device_id = str(msg.get("deviceId") or "").strip() or f"dev-{session.addr[0]}" device_name = str(msg.get("deviceName") or "手表").strip() token = self.store.remember_device(device_id, device_name) session.device_id = device_id session.device_name = device_name session.authed = True # 首次配对成功后自动把配对窗口关上 —— 用户设完码只需配一次， # 之后不用再操心"要不要关配对"。 if self.store.get("close_pair_after_first"): self.pairing_enabled = False self.store.set("allow_new_pairing", False) self.store.set("close_pair_after_first", False) self._log("[配对] 首台设备已配对，配对窗口已自动关闭") self._log(f"[配对成功] {device_name} ({session.addr[0]})") self._notify_ui() return { "ok": True, "token": token, "serverName": self.store.get("server_name", ""), "protocol": P.PROTOCOL_VERSION, } def _do_auth(self, session: Session, msg: dict) -> dict: device_id = str(msg.get("deviceId") or "").strip() token = str(msg.get("token") or "").strip() if not device_id or not token: return {"ok": False, "error": "缺少设备标识或令牌"} if not self.store.verify_token(device_id, token): self._log(f"[鉴权失败] {device_id} 令牌无效") return {"ok": False, "error": "令牌无效，请重新配对", "needPair": True} session.device_id = device_id session.device_name = str(msg.get("deviceName") or "").strip() or \ self.store.devices.get(device_id, {}).get("name", "设备") session.authed = True self._log(f"[已连接] {session.device_name} ({session.addr[0]})") self._notify_ui() return { "ok": True, "serverName": self.store.get("server_name", ""), "protocol": P.PROTOCOL_VERSION, } # ---------------- 媒体/文件接收 ---------------- def _recv_media(self, session: Session, msg: dict, kind: str, label: str) -> dict: seq = int(msg.get("seq") or 0) last = _truthy(msg.get("last")) data = P.b64decode(str(msg.get("data") or "")) name = os.path.basename(str(msg.get("name") or "")) or f"{kind}_{int(time.time())}" assembler = session.buffers[kind] try: blob = assembler.feed(seq, last, data) except P.ChunkTooLarge as exc: # 顺序不对或超限：丢掉这个 kind 的缓冲，回一条明确的错误 session.buffers.pop(kind, None) raise ActionError(str(exc)) from exc if blob is None: # 中间分块不回包：避免回复洪泛与后续命令的消息错位 return None target = self.store.received_dir() / name target = _unique_path(target) try: target.write_bytes(blob) except OSError as exc: raise ActionError(f"保存失败: {exc}") from exc size = len(blob) import hashlib digest = hashlib.sha256(blob).hexdigest()[:16] self._log(f"[{label}] 收到 {name} ({_human(size)}) sha256={digest}") if kind == P.CMD_IMAGE: note = self.actions.show_image(str(target), str(msg.get("title") or "")) elif kind == P.CMD_AUDIO: note = self.actions.play_audio(str(target)) else: note = f"已保存到 {target}" return {"ok": True, "note": note, "path": str(target), "size": size} # 一次最多返回多少个文件。 # 为什么不返回全部：手表屏幕就那么大，几千条也翻不动， # 而且一条 JSON 太长会把单条消息撑爆。 FILE_LIST_LIMIT = 100 def _file_list(self) -> dict: directory = self.store.received_dir() try: entries = [p for p in directory.iterdir() if p.is_file()] except OSError as exc: raise ActionError(f"读取目录失败: {exc}") from exc total = len(entries) entries.sort(key=lambda p: p.stat().st_mtime, reverse=True) shown = entries[:self.FILE_LIST_LIMIT] files = [] for entry in shown: try: files.append({ "name": entry.name, "size": entry.stat().st_size, "mtime": entry.stat().st_mtime, }) except OSError: # 正好被别的进程删掉了，跳过就好，别让整条列表失败 continue return { "ok": True, "dir": str(directory), "files": files, # 说清楚有没有被截断 —— 不能默默藏起来， # 用户会以为文件丢了（这个坑在测试里踩过一次）。 "total": total, "shown": len(files), "truncated": total > len(files), } def _file_pull(self, session: Session, path: str) -> dict: target = self._resolve_received(path) if target is None: raise ActionError(f"文件不存在: {path}") blob = target.read_bytes() if not session.send_chunks(P.EVT_FILE, target.name, blob): raise ActionError("发送文件失败：连接已断开") return {"ok": True, "name": target.name, "size": len(blob)} # ---------------- 播放器数据源（2.3.0）---------------- # # 手表上的播放器要"能看进度、能左右划 seek" —— 那就必须手表自己解码播放。 # 所以电脑端负责两件事：**告诉它有哪些媒体** + **把文件发过去**。 # 允许扫描的媒体目录（相对用户主目录）。 MEDIA_SUBDIRS = ("Music", "Videos", "Downloads", "Desktop", "音乐", "视频", "下载", "桌面") AUDIO_EXT = {".mp3", ".wav", ".m4a", ".aac", ".flac", ".ogg", ".wma", ".amr", ".opus", ".ape"} VIDEO_EXT = {".mp4", ".mkv", ".avi", ".mov", ".wmv", ".flv", ".webm", ".3gp", ".m4v", ".ts", ".rmvb"} MEDIA_LIST_LIMIT = 300 def _media_roots(self) -> list: """允许访问的根目录（都 resolve 成绝对路径）。""" roots = [] home = Path.home() for sub in self.MEDIA_SUBDIRS: p = home / sub try: if p.is_dir(): roots.append(p.resolve()) except OSError: continue try: roots.append(Path(self.store.received_dir()).resolve()) except Exception: # noqa: BLE001 pass return roots def _media_list(self) -> dict: roots = self._media_roots() files = [] for root in roots: try: entries = list(root.iterdir()) except OSError: continue for p in entries: try: if not p.is_file(): continue ext = p.suffix.lower() if ext in self.AUDIO_EXT: kind = "audio" elif ext in self.VIDEO_EXT: kind = "video" else: continue files.append({ "name": p.name, "path": str(p), "size": p.stat().st_size, "kind": kind, "mtime": p.stat().st_mtime, }) except OSError: continue # 新的排前面 files.sort(key=lambda f: f.get("mtime", 0), reverse=True) total = len(files) shown = files[:self.MEDIA_LIST_LIMIT] return { "ok": True, "files": shown, "total": total, "shown": len(shown), "truncated": total > len(shown), "roots": [str(r) for r in roots], } def _media_pull(self, session: "Session", path: str) -> dict: """把一个媒体文件发给手表。 ⚠ **安全边界**：只允许媒体根目录里的文件。 不能用 `str(p).startswith(root)` 这种判断 —— 那样 「音乐-evil」这种**同前缀的兄弟目录**会被误判成子目录。 必须用 `relative_to`，它在路径不真的位于该目录下时会抛 ValueError。 """ wanted = Path(str(path or "")).expanduser() try: wanted = wanted.resolve() except (OSError, ValueError) as exc: raise ActionError(f"路径不合法: {exc}") from exc allowed = False for root in self._media_roots(): try: wanted.relative_to(root) allowed = True break except ValueError: continue if not allowed: raise ActionError("这个文件不在允许的媒体目录里") if not wanted.is_file(): raise ActionError(f"文件不存在: {wanted.name}") size = wanted.stat().st_size if size <= 0: raise ActionError("文件是空的") if size > 512 * 1024 * 1024: raise ActionError("文件太大（超过 512MB），播放器放不下") blob = wanted.read_bytes() if not session.send_chunks(P.EVT_FILE, wanted.name, blob): raise ActionError("发送失败：连接已断开") return {"ok": True, "name": wanted.name, "size": len(blob)} def _is_sensitive(self, path: Path) -> bool: """程序自己的数据目录：配对码、设备令牌都在这儿。 注意：接收目录必须豁免 —— 它在程序目录里面， 不豁免的话拉文件会全被挡住（"文件不存在"）。 """ try: received = Path(self.store.received_dir()).resolve() path.resolve().relative_to(received) return False except (ValueError, OSError): pass if self.features is not None: try: return self.features.is_sensitive(path) except Exception: # noqa: BLE001 pass try: folder = Path(self.store.dir).resolve() resolved = path.resolve() return resolved == folder or folder in resolved.parents except OSError: return False def _resolve_received(self, path: str) -> Path | None: """按名字在接收目录里定位文件。 同名文件会被 _unique_path 存成 name_1.bin / name_2.bin， 所以这里必须取**最新的一份**，否则「同名文件发两次」时 客户端拉回的会是上一轮的旧文件。 """ 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()) if self._is_sensitive(resolved): self._log(f"[文件] 拒绝拉取数据目录: {raw}") return None for root in allowed: try: resolved.relative_to(root.resolve()) return resolved except (ValueError, OSError): continue self._log(f"[文件] 拒绝拉取允许范围外的路径: {raw}") return None # 相对路径也要过一遍检查。Path("data/settings.json").is_absolute() 是 False， # 上面那一段挡不住它 —— 而 EXE 的 CWD 就是程序目录， # 于是已配对设备能把 data/settings.json（配对码 + 全部设备令牌）拉走。 # # 注意：这里**不能**因为"不在允许目录内"就 return None —— # 后面的逻辑还要按文件名在接收目录里找。只有命中敏感目录时才直接拒绝。 try: relative = (Path.cwd() / raw).resolve() except OSError: relative = None if relative is not None and relative.is_file(): # 只有真的命中一个**存在的文件**时才做敏感判断。 # 否则 (cwd/文件名) 会拼出一个程序目录内的假路径， # 被误判成敏感目录 -> 直接返回 None -> 走不到下面的按名字查找, # 表现就是"接收文件永远拿不到东西"。 if self._is_sensitive(relative): self._log(f"[文件] 拒绝拉取数据目录: {raw}") return None allowed_roots = [self.store.received_dir()] if self.features is not None: allowed_roots.extend(self.features.allowed_roots()) for root in allowed_roots: try: relative.relative_to(root.resolve()) return relative except (ValueError, OSError): continue self._log(f"[文件] 拒绝拉取允许范围外的路径: {relative}") return None # 落到下面：按文件名在「接收目录」里找最新的那一份 directory = self.store.received_dir() basename = os.path.basename(path) if not basename: return None stem, suffix = os.path.splitext(basename) pattern = re.compile( rf"^{re.escape(stem)}(?:_\d+)?{re.escape(suffix)}$", re.IGNORECASE) matches: list[Path] = [] try: for entry in directory.iterdir(): if entry.is_file() and pattern.match(entry.name): matches.append(entry) except OSError: return None if matches: return max(matches, key=lambda p: p.stat().st_mtime) return None # ---------------- 屏幕镜像 / 截屏 ---------------- def _ensure_capturer(self) -> ScreenCapturer | None: """拿到屏幕采集器。**设置变了要重建。** 以前只判断 self._capturer is None，于是改了分辨率/画质之后 正在推流的会话永远用旧设置 —— 用户会以为"改了没用"。 现在记住建它时用的参数，参数一变就重建。 """ width = int(self.store.get("screen_max_width", 480) or 480) quality = int(self.store.get("screen_quality", 55) or 55) signature = (width, quality) if self._capturer is not None and \ getattr(self, "_capturer_signature", None) == signature: return self._capturer self._capturer = ScreenCapturer(max_width=width, quality=quality) self._capturer_signature = signature return self._capturer def _require_screen(self) -> ScreenCapturer: if not self.store.get("allow_screen", True): raise ActionError("被控端已关闭屏幕共享（可在被控端窗口重新开启）") capturer = self._ensure_capturer() if capturer is None or not capturer.available(): raise ActionError("当前系统不支持屏幕捕获") return capturer def _screen_start(self, session: Session, msg: dict) -> dict: capturer = self._require_screen() fps = _clamp(int(msg.get("fps") or self.store.get("screen_fps", 3) or 3), 1, 15) quality = _clamp(int(msg.get("quality") or self.store.get("screen_quality", 55) or 55), 20, 90) max_width = _clamp(int(msg.get("maxWidth") or self.store.get("screen_max_width", 480) or 480), 160, 1920) capturer.quality = quality capturer.max_width = max_width session.wants_screen = True self._ensure_streamer().start(fps) self._set_indicator(True) who = session.device_name or session.addr[0] self._log(f"[屏幕] {who} 开始观看屏幕镜像（{fps} fps, {max_width}px, 质量 {quality}）") self._notify_ui() return {"ok": True, "fps": fps, "maxWidth": max_width, "quality": quality, "note": f"屏幕共享已开始（{fps} fps）"} def _screen_stop(self, session: Session) -> dict: session.wants_screen = False if not any(s.wants_screen for s in self._authed_sessions()): self._stop_streamer() self._log(f"[屏幕] {session.device_name or session.addr[0]} 停止观看") self._notify_ui() return {"ok": True, "note": "屏幕共享已停止"} def _screenshot(self, session: Session, msg: dict) -> dict: capturer = self._require_screen() quality = _clamp(int(msg.get("quality") or self.store.get("screen_quality", 55) or 55), 30, 95) max_width = _clamp(int(msg.get("maxWidth") or 1280), 160, 3840) capturer.quality = quality capturer.max_width = max_width # 截屏也是一次屏幕读取，同样给出可见提示（短暂闪烁） self._flash_indicator() try: jpeg, width, height = capturer.grab_jpeg() except ScreenError as exc: raise ActionError(str(exc)) from exc name = f"screenshot_{time.strftime('%Y%m%d_%H%M%S')}.jpg" who = session.device_name or session.addr[0] self._log(f"[屏幕] {who} 截取屏幕 {width}x{height}（{_human(len(jpeg))}）") return {"ok": True, "name": name, "w": width, "h": height, "size": len(jpeg), "data": P.b64encode(jpeg), "note": "截屏已获取"} # ---- 帧流 ---- def _ensure_streamer(self) -> FrameStreamer: if self._streamer is None: self._streamer = FrameStreamer( self._ensure_capturer(), self._on_frame, on_stop=self._on_stream_stopped) return self._streamer def _on_frame(self, jpeg: bytes, width: int, height: int) -> bool: """把一帧推给所有正在观看的设备。返回 False 表示应停止推流。""" viewers = [s for s in self._authed_sessions() if s.wants_screen] if not viewers: return False payload = { "event": P.EVT_FRAME, "w": width, "h": height, "size": len(jpeg), "ts": time.time(), "data": P.b64encode(jpeg), } delivered = 0 for viewer in viewers: if viewer.send(payload): delivered += 1 else: viewer.wants_screen = False return delivered > 0 def _on_stream_stopped(self) -> None: self._set_indicator(False) self._notify_ui() def _stop_streamer(self) -> None: if self._streamer is not None and self._streamer.running: self._log("[屏幕] 已停止屏幕镜像") if self._streamer is not None: self._streamer.stop() self._set_indicator(False) def _set_indicator(self, active: bool) -> None: """开启/关闭被控端上的「屏幕正在被共享」可见提示。 这个提示是**强制**的：只要在推流就一定显示，界面里没有关闭开关。 远控工具必须让使用者知道屏幕正在被共享。 """ if self._screen_indicator_on == active: return self._screen_indicator_on = active if self.ui and hasattr(self.ui, "set_screen_sharing"): try: self.ui.set_screen_sharing(active) except Exception: pass def _flash_indicator(self) -> None: """截屏时短暂显示提示。""" if self.ui and hasattr(self.ui, "flash_screen_indicator"): try: self.ui.flash_screen_indicator(2.5) except Exception: pass # ---------------- 剪贴板同步 ---------------- def _clip_loop(self) -> None: time.sleep(2.0) while self._running: time.sleep(1.5) if not self._authed_sessions(): continue try: text = self.actions.clip_get() except Exception: continue if text and text != self._last_clip: self._last_clip = text self.broadcast({"event": P.EVT_CLIP, "text": text}) def _authed_sessions(self) -> list[Session]: with self._lock: return [s for s in self.sessions if s.authed] # ---------------- 广播 / UI 通知 ---------------- def broadcast(self, obj: dict) -> None: for session in self._authed_sessions(): session.send(obj) def stop_screen_sharing(self) -> int: """本机使用者主动停止屏幕共享（点击红色提示条上的按钮）。 这是被控端的最终否决权：任何时候都能一键掐断。 """ with self._lock: sessions = list(self.sessions) affected = 0 for session in sessions: if session.wants_screen: session.wants_screen = False affected += 1 self._stop_streamer() self._set_indicator(False) self.broadcast({"event": "screen_stopped", "reason": "被控端本机已停止共享"}) self._log(f"[屏幕] 本机使用者停止了屏幕共享（影响 {affected} 台设备）") self._notify_ui() return affected def disconnect_all(self) -> int: with self._lock: sessions = list(self.sessions) for session in sessions: session.close() self._log(f"已断开 {len(sessions)} 个连接") self._notify_ui() return len(sessions) def status(self) -> dict: """给界面用的实时状态。""" with self._lock: sessions = list(self.sessions) devices = [] for s in sessions: devices.append({ "addr": s.addr[0], "name": s.device_name or "未鉴权", "authed": s.authed, "since": s.connected_at, "last_seen": s.last_seen, }) return { "running": self._running, "port": self.port, "pairing_code": self.store.pairing_code(), "pairing_enabled": self.pairing_enabled, "connected": devices, "paired_saved": self.store.device_list(), "screen_sharing": bool(self._streamer and self._streamer.running), "screen_viewers": sum(1 for s in sessions if s.wants_screen), "screen_fps": round(self._streamer.actual_fps, 1) if self._streamer and self._streamer.running else 0.0, "discovery": self._discovery.status() if self._discovery else {"running": False}, "total_connections": self._total_connections, "local_ip": local_ip(), } def _notify_ui(self) -> None: if self.ui and hasattr(self.ui, "on_state_changed"): try: self.ui.on_state_changed() except Exception: pass def _log(self, text: str) -> None: if self.ui and hasattr(self.ui, "log"): try: self.ui.log(text) except Exception: pass def _clamp(value: int, low: int, high: int) -> int: return max(low, min(high, int(value))) def _unique_path(path: Path) -> Path: if not path.exists(): return path stem, suffix = path.stem, path.suffix for index in range(1, 1000): candidate = path.with_name(f"{stem}_{index}{suffix}") if not candidate.exists(): return candidate return path.with_name(f"{stem}_{int(time.time())}{suffix}") def _human(num: float) -> str: for unit in ("B", "KB", "MB", "GB"): if abs(num) < 1024.0: return f"{num:.1f}{unit}" num /= 1024.0 return f"{num:.1f}TB" def lan_ips() -> list[str]: """列出所有可用的局域网 IPv4 地址，优先真实网卡。 这台机器上装了 Tailscale，`gethostbyname(hostname)` 会返回 Tailscale 的 169.254 链路地址——那是没用的。所以这里枚举网卡并做优先级排序。 """ candidates: list[tuple[int, str]] = [] try: for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET): addr = info[4][0] if addr and not addr.startswith("127."): candidates.append((_ip_rank(addr), addr)) except OSError: pass # 主路由地址（UDP 连 8.8.8.8 不会真的发包，只用来问内核选哪张网卡） try: probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) probe.connect(("8.8.8.8", 80)) addr = probe.getsockname()[0] probe.close() if addr and not addr.startswith("127."): candidates.append((0, addr)) # 最优 except OSError: pass ranked: dict[str, int] = {} for rank, addr in candidates: if addr not in ranked or rank < ranked[addr]: ranked[addr] = rank return [addr for addr, _ in sorted(ranked.items(), key=lambda kv: kv[1])] def _ip_rank(addr: str) -> int: """数字越小越可能是真实局域网地址。""" if addr.startswith("169.254."): return 90 # APIPA / Tailscale 链路地址 if addr.startswith("192.168."): return 10 if addr.startswith("10."): return 20 if addr.startswith("172."): try: second = int(addr.split(".")[1]) if 16 <= second <= 31: return 25 except (ValueError, IndexError): pass return 60 def local_ip() -> str: """本机在局域网中最可能的地址。""" ips = lan_ips() return ips[0] if ips else "127.0.0.1"