# -*- coding: utf-8 -*- import sys, os, subprocess, time, ctypes, threading, functools, base64, bisect from pathlib import Path from PyQt5.QtWidgets import ( QApplication, QWidget, QLabel, QPushButton, QVBoxLayout, QHBoxLayout, QFrame, QMessageBox, QTextEdit, QCheckBox, QMenu, QDialog, QLineEdit, QListWidget, QListWidgetItem, ) from PyQt5.QtCore import Qt, QTimer, QRectF, pyqtSignal from PyQt5.QtGui import QFont, QIcon, QPixmap, QImage, QColor, QPainter # GUI 窗口在 96 DPI / 100% 缩放下的逻辑尺寸；启用 Qt 高分屏缩放后， # Qt 会自动按 devicePixelRatio 放大/缩小，因此代码里保持逻辑尺寸不变即可。 _BASE_WINDOW_WIDTH = 340 _BASE_WINDOW_HEIGHT = 470 # 内容区 + 加大日志区 + 底部两行提示小字 # ── 配色（Qt StyleSheet 使用）──────────────────────────── C_TITLE = "#8899aa" # 标题 / 副标题 C_GREEN = "#51cf66" # 开启按钮、已连接状态 C_YELLOW = "#ffd43b" # 提示色（保留） C_RED = "#ff6b6b" # 关闭按钮、未连接 / 错误状态 C_BG = "#ffffff" # 主背景 C_CARD = "#f0f2f5" # 卡片背景 C_BORDER = "#d0d5dd" # 边框 C_TEXT = "#333333" # 正文 C_SUBTEXT = "#889099" # 次要文字 # ── 路径 ────────────────────────────────────────────── # 已被打包为 EXE（PyInstaller 的 sys.frozen / Nuitka 的 __compiled__）： # BASE_DIR -> 内置数据目录：PyInstaller 用 _MEIPASS，Nuitka onefile 用 # __file__ 所在解压临时目录（无 _MEIPASS 时回退） # PERSIST_DIR -> EXE 所在目录，服务安装等需要持久化的操作都放这里 if getattr(sys, 'frozen', False) or "__compiled__" in globals(): BASE_DIR = Path(getattr(sys, "_MEIPASS", None) or Path(__file__).resolve().parent) PERSIST_DIR = Path(sys.executable).resolve().parent else: BASE_DIR = Path(__file__).resolve().parent PERSIST_DIR = BASE_DIR # ── 普通设置（config.json）──────────────────────────── # 版本号、更新链接等设置，缺失时用默认值兜底 _CONFIG = None def _load_config(): """加载 config.json；失败或缺失返回空字典（各设置用默认值兜底）。""" global _CONFIG if _CONFIG is not None: return _CONFIG _CONFIG = {} try: import json path = _data_path("config.json") if path is None: log("config.json 缺失，使用默认设置") return _CONFIG data = json.loads(Path(path).read_text("utf-8")) if not isinstance(data, dict): raise ValueError("config.json 顶层应为对象") _CONFIG = data log(f"已加载普通设置: {len(_CONFIG)} 项 (config.json)") except Exception as e: log_red(f"config.json 加载失败: {e}") return _CONFIG def _config_val(key, default=""): return _load_config().get(key, default) # ── 节点配置（nodes.json）──────────────────────────── # 节点列表，当前固定取第一个（后续可扩展选择） _NODES = None def _load_nodes(): """加载 nodes.json 节点列表（用户数据目录优先）；失败或缺失时返回空列表。""" global _NODES if _NODES is not None: return _NODES _NODES = [] try: import json path = _user_path("nodes.json") if path is None: log_red("nodes.json 缺失，无法加载节点配置") return _NODES data = json.loads(_nodes_decrypt(Path(path).read_bytes())) if not isinstance(data, list): raise ValueError("nodes.json 顶层应为节点数组") _NODES = data if not _SELECTED_RESTORED: _restore_selected_node(_NODES) # 恢复上次使用的节点 log(f"已加载节点配置: {len(_NODES)} 个 ({path})") if data and not str(data[0].get("domain") or "").strip(): log_red("节点未填写域名：请配置 config.json 的 nodes_url 在线更新，" "或使用 others/nodes_tool 导入节点配置") except Exception as e: log_red(f"nodes.json 加载失败: {e}") return _NODES _SELECTED_NODE = 0 # 当前选中节点下标（默认第一条） _SELECTED_RESTORED = False # 启动时是否已按设置恢复过上次选中的节点 def _current_node(): """当前使用的节点：默认取列表第一个；无节点时返回 None。""" nodes = _load_nodes() if not nodes: return None i = _SELECTED_NODE return nodes[i] if 0 <= i < len(nodes) else nodes[0] def _selected_index(): """当前选中节点下标。""" return _SELECTED_NODE def _switch_node(index): """切换当前节点下标；相同或越界返回 False。""" global _SELECTED_NODE nodes = _load_nodes() if not nodes or not (0 <= index < len(nodes)): log_red("节点切换失败: 列表为空或下标越界") return False if index == _SELECTED_NODE: log("已在当前节点") return False _SELECTED_NODE = index _save_current_node(nodes[index].get("domain") or "") # 记录到设置，下次启动恢复 return True def _restore_selected_node(nodes): """首次加载时按设置里记录的 current_node（域名）恢复上次选中的节点。""" global _SELECTED_NODE, _SELECTED_RESTORED _SELECTED_RESTORED = True saved = str(_settings_load().get("current_node") or "") for i, n in enumerate(nodes): if saved and str(n.get("domain") or "") == saved: _SELECTED_NODE = i return _SELECTED_NODE = 0 def _save_current_node(domain): """把当前选中节点域名写入设置文件（下次启动自动继续使用）。""" data = _settings_load() data["current_node"] = str(domain or "") _settings_save(data) def _reload_nodes(): """强制重新读取 nodes.json（菜单打开时调用，外部改动即时生效）。""" global _NODES _NODES = None return _load_nodes() def _node_val(key, default=""): node = _current_node() if not node: return default v = node.get(key, default) return v if v not in (None, "") else default # 空字符串视为未填写 def _endpoint_domain(): return _node_val("domain", "???") def _endpoint_port(): return int(_node_val("port", 443)) def _endpoint_country(): return _node_val("country", "") def _proxy_auth_basic(): """与 nginx 正向代理使用同一套 Basic 认证凭证（取自 nodes.json）。""" node = _current_node() if not node: return "" return base64.b64encode(f"{node.get('username', '')}:{node.get('password', '')}".encode()).decode() _tls_forwarder = None _pac_server = None PROXY_HOST = "127.0.0.1" PROXY_REG = r"HKCU\Software\Microsoft\Windows\CurrentVersion\Internet Settings" # _TLS_LOCAL_PORT / PROXY_PORT 定义在 _data_path 之后（取自 config.json local_port） # PAC 模式使用全量转发 PAC：所有流量送进本地转发器按 IP 智能分流 # （中国 IP 直连、其余走加密隧道）；域名表在 pac_domains.json。 _BUILT_IN_PAC = r'''function FindProxyForURL(url, host) { var proxy = "PROXY 127.0.0.1:1080"; return proxy; } ''' def _build_pac(proxy_spec: str) -> str: """生成全量转发 PAC，替换为当前模式对应的代理地址。""" return _BUILT_IN_PAC.replace( 'var proxy = "PROXY 127.0.0.1:1080";', f'var proxy = "PROXY {proxy_spec}";' ) _LAN_BYPASS_PATTERNS = ["10.*", "192.168.*", "169.254.*", "127.*"] _LAN_BYPASS_PATTERNS += [f"172.{i}.*" for i in range(16, 32)] _LAN_BYPASS_JS = ( " // LAN bypass: RFC1918 / link-local / loopback (resolves domain first)\n" " var _lanIp = dnsResolve(host);\n" " if (_lanIp && (" + " ||\n".join(f' shExpMatch(_lanIp, "{p}")' for p in _LAN_BYPASS_PATTERNS) + ")) {\n" " return \"DIRECT\";\n" " }\n" ) def _apply_lan_bypass(pac_content: str) -> str: """Inject LAN-direct rules into PAC so private IPs skip the local forwarder.""" marker = "function FindProxyForURL(url, host) {" if marker not in pac_content: return pac_content return pac_content.replace(marker, marker + "\n" + _LAN_BYPASS_JS, 1) def _proxy_override(bypass_lan: bool = False) -> str: """ProxyOverride for fixed-proxy mode; append common LAN ranges when enabled.""" if not bypass_lan: return "" parts = ["", "10.*"] parts += [f"172.{i}.*" for i in range(16, 32)] parts += ["192.168.*", "169.254.*", "127.*"] return ";".join(parts) def _start_pac_server(pac_content: str) -> int: """在 127.0.0.1 启动内存 PAC HTTP 服务，返回实际监听的端口。""" global _pac_server _stop_pac_server() import http.server import socketserver class _PACHandler(http.server.BaseHTTPRequestHandler): def do_GET(self): if self.path == "/proxy.pac": body = pac_content.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/x-ns-proxy-autoconfig") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) else: self.send_response(404) self.end_headers() def log_message(self, format, *args): pass # 避免在日志区输出大量 HTTP 访问日志 class _ReuseAddrTCPServer(socketserver.TCPServer): allow_reuse_address = True srv = _ReuseAddrTCPServer(("127.0.0.1", 0), _PACHandler) port = srv.server_address[1] threading.Thread(target=srv.serve_forever, daemon=True).start() _pac_server = {"server": srv, "port": port} return port def _stop_pac_server(): """停止本地 PAC HTTP 服务；shutdown() 唤醒 serve_forever 后优雅退出。""" global _pac_server if _pac_server: srv = _pac_server["server"] _pac_server = None # 不要直接 close socket：serve_forever 的 select 仍注册着它，Windows 上 # 对已关闭 socket 做 select 会抛 WinError 10038（非套接字操作）。 # shutdown() 让 select 在 poll_interval（0.5s）内返回并正常退出。 srv.daemon_threads = True # 请求处理线程不阻塞 serve_forever 退出 def _shutdown(): try: srv.shutdown() except Exception: pass try: srv.server_close() except Exception: pass t = threading.Thread(target=_shutdown, daemon=True) t.start() t.join(timeout=1.0) # ── 日志输出 ────────────────────────────────────────── # GUI 初始化后会把 _log_func 设为 MainWindow.log_signal.emit _log_func = None def log(msg: str): """输出一行日志到 GUI 日志区。""" if _log_func: _log_func(msg) def log_red(msg: str): """输出一行红色警告日志到 GUI 日志区。""" if _log_func: _log_func(f'{msg}') # ── 通用工具函数 ────────────────────────────────────── def _run(cmd, warn=True): """无窗口执行子进程，失败时记录警告日志。""" r = subprocess.run(cmd, capture_output=True, text=True, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)) if warn and r.returncode != 0: err = r.stderr.strip() if r.stderr else "" log(f"[警告] {' '.join(str(c) for c in cmd)} -> {r.returncode} {err}") return r def _notify_proxy_change(): """通知 WinINET 和顶层窗口代理设置已变更，让已运行程序立即重读。""" try: import ctypes ctypes.windll.Wininet.InternetSetOptionW(None, 39, None, 0) ctypes.windll.Wininet.InternetSetOptionW(None, 37, None, 0) result = ctypes.c_ulong() ctypes.windll.user32.SendMessageTimeoutW( 0xFFFF, 0x001A, 0, ctypes.c_wchar_p("InternetSettings"), 0x0002, 5000, ctypes.byref(result)) except Exception: pass def proxy_on(pac_url=None, bypass_lan=False): """设置 Windows 系统代理。pac_url 不为空时启用 PAC 模式，否则使用固定代理。""" if pac_url: # PAC 模式：仅设置 AutoConfigURL，ProxyEnable 置 0， # 避免程序读到"固定代理已启用但服务器为空"的混乱状态 _run(["reg", "add", PROXY_REG, "/v", "ProxyEnable", "/t", "REG_DWORD", "/d", "0", "/f"]) # PAC 模式：清理固定代理遗留项，避免两种设置冲突 _run(["reg", "delete", PROXY_REG, "/v", "ProxyServer", "/f"], warn=False) _run(["reg", "delete", PROXY_REG, "/v", "ProxyOverride", "/f"], warn=False) _run(["reg", "add", PROXY_REG, "/v", "AutoConfigURL", "/t", "REG_SZ", "/d", pac_url, "/f"]) log("系统代理已设置 (PAC)") else: hp = f"{PROXY_HOST}:{PROXY_PORT}" _run(["reg", "add", PROXY_REG, "/v", "ProxyEnable", "/t", "REG_DWORD", "/d", "1", "/f"]) _run(["reg", "add", PROXY_REG, "/v", "ProxyServer", "/t", "REG_SZ", "/d", hp, "/f"]) _run(["reg", "add", PROXY_REG, "/v", "ProxyOverride", "/t", "REG_SZ", "/d", _proxy_override(bypass_lan), "/f"]) log(f"系统代理已设置: {hp}") _notify_proxy_change() def proxy_off(): """关闭系统代理并停止 TLS 转发器 / PAC 本地服务。""" log("正在关闭系统代理...") _run(["reg", "add", PROXY_REG, "/v", "ProxyEnable", "/t", "REG_DWORD", "/d", "0", "/f"]) # 清除 ProxyServer / ProxyOverride / AutoConfigURL，确保两种模式都清理干净 _run(["reg", "delete", PROXY_REG, "/v", "ProxyServer", "/f"], warn=False) _run(["reg", "delete", PROXY_REG, "/v", "ProxyOverride", "/f"], warn=False) _run(["reg", "delete", PROXY_REG, "/v", "AutoConfigURL", "/f"], warn=False) log("系统代理已关闭") _notify_proxy_change() _tls_stop() _stop_pac_server() # 打印本次转发流量统计 up, down = _traffic_snapshot() total = up + down log(f"本次转发流量共 {_fmt_bytes(total)}") # ── 本地认证转发器 ───────────────────────────────────── # ── 流量统计 ────────────────────────────────────────── # 所有代理流量都经过本地转发器（直连 / PAC 均如此），转发器里计数即可覆盖全部模式； # 每次"开启代理"时清零，即"本次连接"流量。 _traffic_lock = threading.Lock() _traffic = {"up": 0, "down": 0} _traffic_prev = {"up": 0, "down": 0, "ts": 0.0} def _traffic_add(up=0, down=0): with _traffic_lock: _traffic["up"] += up _traffic["down"] += down def _traffic_reset(): with _traffic_lock: _traffic["up"] = _traffic["down"] = 0 _traffic_prev["up"] = _traffic_prev["down"] = 0 _traffic_prev["ts"] = time.time() def _fmt_bytes(n): f = float(n) for unit in ("B", "KB", "MB", "GB", "TB"): if f < 1024 or unit == "TB": return f"{int(f)} {unit}" if unit == "B" else f"{f:.1f} {unit}" f /= 1024 def _traffic_rate(): with _traffic_lock: now = time.time() up_total = _traffic["up"] down_total = _traffic["down"] prev_up = _traffic_prev["up"] prev_down = _traffic_prev["down"] prev_ts = _traffic_prev["ts"] if prev_ts > 0 and now > prev_ts: dt = now - prev_ts up_rate = int((up_total - prev_up) / dt) down_rate = int((down_total - prev_down) / dt) else: up_rate = down_rate = 0 _traffic_prev["up"] = up_total _traffic_prev["down"] = down_total _traffic_prev["ts"] = now return up_rate, down_rate, up_total, down_total def _traffic_snapshot(): with _traffic_lock: return _traffic["up"], _traffic["down"] def _tls_start(remote_host, remote_port=443, use_tls=True, bypass_lan=False, smart=False, quiet=False): """启动本地认证转发器：127.0.0.1: -> remote_host:remote_port（可选 TLS 包裹）。 smart=True 时启用 IP 智能分流（域名表/非中国 IP 走隧道，中国 IP 直连）。 quiet=True 时不输出启动日志（临时转发器用）。 """ global _tls_forwarder _tls_stop() import socket, ssl # 服务器证书为 Let's Encrypt 正式证书，默认上下文完整校验 ctx = ssl.create_default_context() # Windows 上 Hyper-V/WSL 的端口保留区间会动态变动，bind 可能瞬时被拒 # (WinError 10013/10048)；重试 3 次，间隔 1 秒，每次用全新 socket last_err = None svr = None for attempt in range(3): svr = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: svr.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) svr.bind(("127.0.0.1", _TLS_LOCAL_PORT)) break except OSError as e: last_err = e svr.close() svr = None if attempt < 2: log(f"端口 {_TLS_LOCAL_PORT} 绑定失败: {e}，1 秒后重试 ({attempt + 1}/3)") time.sleep(1) if svr is None: raise last_err svr.listen(32) svr.settimeout(1) _tls_forwarder = {"sock": svr, "running": True} proto = "TLS" if use_tls else "TCP" def _run(): while _tls_forwarder and _tls_forwarder.get("running"): try: client, addr = svr.accept() threading.Thread(target=_tls_relay, args=(client, remote_host, remote_port, ctx, use_tls, bypass_lan, smart), daemon=True).start() except socket.timeout: continue except Exception: break t = threading.Thread(target=_run, daemon=True) t.start() if not quiet: log(f"本地转发器已启动 ({proto} 127.0.0.1:{_TLS_LOCAL_PORT} -> {remote_host}:{remote_port})") _HTTP_METHODS = (b"CONNECT ", b"GET ", b"POST ", b"PUT ", b"DELETE ", b"HEAD ", b"OPTIONS ", b"PATCH ") def _starts_with_method(data: bytes) -> bool: return any(data.startswith(m) for m in _HTTP_METHODS) def _inject_proxy_auth(data: bytes) -> bytes: """在 HTTP 代理请求头中注入 Authorization（如果尚未存在）。 nginx 的 auth_basic 校验 Authorization 而非 Proxy-Authorization， 因此转发器向上游发送 Basic 认证时使用 Authorization 头。 """ if not _starts_with_method(data): return data end = data.find(b"\r\n\r\n") if end == -1: return data headers_lower = data[:end].lower() if b"authorization:" in headers_lower: return data auth_line = f"Authorization: Basic {_proxy_auth_basic()}\r\n".encode() return data[:end + 2] + auth_line + data[end + 2:] def _make_request_injector(): """为明文 HTTP 代理连接生成逐条注入认证的流式处理器。 keep-alive 连接上的后续请求同样需要 Authorization，不能只在首包注入。 请求头跨包时先缓冲，凑齐 \r\n\r\n（或超过 64KB）后统一处理； 请求体数据不以方法开头，原样透传。 注意：仅用于非 CONNECT 连接；CONNECT 隧道的后续流量是加密的，不能触碰。 """ pending = b"" def feed(data: bytes) -> bytes: nonlocal pending if pending: data, pending = pending + data, b"" if not _starts_with_method(data): return data if b"\r\n\r\n" in data or len(data) >= 65536: return _inject_proxy_auth(data) pending = data return b"" return feed @functools.lru_cache(maxsize=256) def _is_private_ip(host: str) -> bool: """判断目标是否命中常用内网/回环 IPv4；域名会解析后判断。""" try: import ipaddress ip = ipaddress.ip_address(host) addresses = [ip] except ValueError: try: import socket addresses = [] for addr in socket.getaddrinfo(host, None, socket.AF_INET, socket.SOCK_STREAM): try: addresses.append(ipaddress.ip_address(addr[4][0])) except ValueError: pass except Exception: return False for ip in addresses: if ip.version != 4: continue n = int(ip) if ( (n & 0xFF000000) == 0x0A000000 or (n & 0xFFF00000) == 0xAC100000 or (n & 0xFFFF0000) == 0xC0A80000 or (n & 0xFF000000) == 0x7F000000 or (n & 0xFFFF0000) == 0xA9FE0000 ): return True return False # ── IP 智能分流（域名表优先，非中国 IP 走隧道，中国 IP 直连） ───── _CN_RANGES = None # 合并排序后的 (start, end) 整数段 _CN_RANGES_LOADED = False _FORCE_PROXY_SUFFIXES = None _FORCE_PROXY_LOADED = False def _data_path(name: str): """数据文件路径：只取打包内置目录（config.json 随 exe 内置）。""" p = Path(BASE_DIR) / name return p if p.is_file() else None # 用户数据目录：Windows 惯例 %LOCALAPPDATA%（C:\Users\<用户>\AppData\Local）， # 非 Windows 环境回退用户主目录；测试可覆盖 _USER_DIR = (Path(os.environ["LOCALAPPDATA"]) if os.name == "nt" and os.environ.get("LOCALAPPDATA") else Path.home()) / "RRPP" def _user_path(name: str): """用户数据文件路径：dev 模式（未打包）源码目录优先，方便本地测试； 打包后用户数据目录优先（首启模板/在线更新写在这里）。""" if not getattr(sys, 'frozen', False): p = _data_path(name) if p is not None: return p p = _USER_DIR / name if p.is_file(): return p return _data_path(name) # ── 本地设置（settings.json，用户数据目录，明文） ───────────── def _settings_path(): return _USER_DIR / "settings.json" def _settings_load(): """读取本地设置；缺失或损坏返回空字典。""" try: import json f = _settings_path() if not f.is_file(): return {} data = json.loads(f.read_text("utf-8")) return data if isinstance(data, dict) else {} except Exception: return {} def _settings_save(data): """本地设置原子写回。""" import json _USER_DIR.mkdir(parents=True, exist_ok=True) tmp = _settings_path().with_suffix(".json.tmp") tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", "utf-8") os.replace(tmp, _settings_path()) def _parse_port(text): """解析端口号；非法（非数字或超出 1-65535）返回 None。""" try: port = int(str(text).strip()) except (TypeError, ValueError): return None return port if 1 <= port <= 65535 else None def _port_val(): """代理转发端口：本地设置 > 内置 config.json > 默认 1080。""" port = _parse_port(_settings_load().get("local_port")) if port is not None: return port port = _parse_port(_config_val("local_port", 1080)) return port if port is not None else 1080 def _set_local_port(port): """更新代理转发端口并写回设置文件（下次开启代理生效）。""" global _TLS_LOCAL_PORT, PROXY_PORT _TLS_LOCAL_PORT = port PROXY_PORT = port data = _settings_load() data["local_port"] = port _settings_save(data) log(f"代理转发端口已保存: {port}（下次开启代理生效）") def _update_via_proxy(): """更新检查/下载是否走节点隧道（CONNECT）；设置项，默认关=直连仓库。""" return bool(_settings_load().get("update_via_proxy", False)) def _set_update_via_proxy(enabled): """写回「使用代理更新」设置（立即生效）。""" data = _settings_load() data["update_via_proxy"] = bool(enabled) _settings_save(data) log(f"更新方式已设为: {'走代理隧道' if enabled else '直连仓库'}") def _skip_update_check(): """是否跳过 PAC 数据与程序版本检查（设置项，默认关）。 不影响节点配置更新与客户端认证。""" return bool(_settings_load().get("skip_update_check", False)) def _set_skip_update_check(enabled): """写回「跳过检查更新」设置（立即生效）。""" data = _settings_load() data["skip_update_check"] = bool(enabled) _settings_save(data) log(f"检查更新已{'跳过' if enabled else '恢复'}（PAC 与程序版本；节点更新不受影响）") # ── 代理白名单（用户可配置，强制直连，settings.json 的 whitelist 键） ───────── # 每条规则 {"value": 域名/IP/CIDR, "enabled": 是否启用}，命中即直连（不经过代理）。 # 规则保存在内存缓存 _WHITELIST，连接建立时直接读取，不落盘回读； # _WHITELIST_VERSION 在保存时 +1，使按 (host, version) 键控的匹配缓存失效。 _WHITELIST = None # 内存缓存：规范化后的规则列表 _WHITELIST_VERSION = 0 # 规则版本号（保存时 +1，用于匹配缓存失效） def _whitelist_rules(): """白名单规则列表（内存缓存，首次读取 settings.json；保存时同步更新）。""" global _WHITELIST if _WHITELIST is None: raw = _settings_load().get("whitelist") if not isinstance(raw, list): raw = [] _WHITELIST = [{"value": str(r.get("value") or "").strip().lower(), "enabled": bool(r.get("enabled", True))} for r in raw if isinstance(r, dict) and str(r.get("value") or "").strip()] return _WHITELIST def _whitelist_save(rules): """保存白名单规则（规范化 + 去空白条目），更新内存缓存并记录日志。""" global _WHITELIST, _WHITELIST_VERSION clean = [{"value": _whitelist_normalize(r.get("value") or ""), "enabled": bool(r.get("enabled", True))} for r in rules if _whitelist_normalize(r.get("value") or "")] data = _settings_load() data["whitelist"] = clean _settings_save(data) _WHITELIST = clean _WHITELIST_VERSION += 1 log(f"代理白名单已保存: {len(clean)} 条（立即生效）") def _whitelist_normalize(entry: str) -> str: """条目规范化：去协议头/通配符/末尾点，统一小写。""" s = str(entry).strip().lower() if "://" in s: s = s.split("://", 1)[1] return s.rstrip("/").rstrip(".").lstrip("*.") def _whitelist_validate_entry(entry: str): """校验并规范化白名单条目；返回 (规范值, 错误)，错误非空表示非法。""" s = _whitelist_normalize(entry) if not s: return "", "内容不能为空" if any(ch.isspace() for ch in s): return "", "不能包含空格" import ipaddress if "/" in s: # CIDR（IPv4/IPv6 网段或单地址） try: ipaddress.ip_network(s, strict=False) except ValueError: return "", "IP/CIDR 格式不正确（如 10.0.0.0/8）" elif s.count(".") == 3 and all(p.isdigit() for p in s.split(".")): # 形如 x.x.x.x：必须是合法 IPv4，防止 300.1.2.3 这类误当域名 try: ipaddress.ip_address(s) except ValueError: return "", "IP 格式不正确" elif ":" in s: # IPv6 字面量（域名不含冒号） try: ipaddress.ip_address(s) except ValueError: return "", "IP 格式不正确" return s, "" def _ver_tuple(v): """版本号转数字元组：'v0.1.0' -> (0, 1, 0)，无法解析的段按 0 处理。""" parts = [] for p in str(v).strip().lstrip("vV").split("."): try: parts.append(int(p)) except ValueError: parts.append(0) return tuple(parts) def _version_newer(latest, current): """latest 是否比 current 新（逐段数字比较，位数不同补 0）。""" lt, ct = _ver_tuple(latest), _ver_tuple(current) n = max(len(lt), len(ct)) lt += (0,) * (n - len(lt)) ct += (0,) * (n - len(ct)) return lt > ct # 以下模块级取值需在 _data_path 定义之后执行 VERSION = _config_val("version", "999908162319") _TLS_LOCAL_PORT = _port_val() # 本地转发器监听端口（设置 > 内置 config > 1080） PROXY_PORT = _TLS_LOCAL_PORT # ── PAC 数据在线更新（CDN 直连，无需认证） ───────────── # 更新源地址取 config.json 的 pac_base_url，目录下直接放两个数据文件: # {pac_base_url}/pac_domains.json 强制隧道域名表（version/date 字段随文件下发） # {pac_base_url}/pac_ip_cn.json 中国 IP 段（同上） _PAC_FILES = ("pac_domains", "pac_ip_cn") def _pac_read_version(path): """读取 PAC 数据文件内的 version/date 字段。""" try: import json data = json.loads(Path(path).read_text("utf-8")) return str(data.get("version") or ""), str(data.get("date") or "") except Exception: return "", "" def _pac_no_proxy_opener(): """直连 opener：显式空代理，忽略系统/环境代理设置（CDN 无需认证）。""" import urllib.request return urllib.request.build_opener(urllib.request.ProxyHandler({})) def _pac_fetch_json(url, timeout=15, tries=2): opener = _pac_no_proxy_opener() import json for i in range(tries): try: with opener.open(url, timeout=timeout) as r: return json.loads(r.read().decode("utf-8")) except Exception: if i == tries - 1: raise def _pac_validate(name, data): if name == "pac_domains": return isinstance(data.get("domains"), list) if name == "pac_ip_cn": return isinstance(data.get("cidrs"), list) return False def _pac_write(name, data): """PAC 数据原子写入用户数据目录（tmp+replace，写不坏旧文件）。""" import json _USER_DIR.mkdir(parents=True, exist_ok=True) tmp = _USER_DIR / f"{name}.json.tmp" tmp.write_text(json.dumps(data, ensure_ascii=False), "utf-8") os.replace(tmp, _USER_DIR / f"{name}.json") # ── nodes.json 加密存储（Windows DPAPI，本机本用户可解） ───────────── # 非 Windows（dev/测试）不加密，读写明文；读取对密文/明文都兼容。 class _DPAPI_BLOB(ctypes.Structure): _fields_ = [("cbData", ctypes.c_uint32), ("pbData", ctypes.POINTER(ctypes.c_char))] def _dpapi_protect(data: bytes) -> bytes: """DPAPI 加密 CryptProtectData（无附加熵，用户+机器绑定）。""" crypt32 = ctypes.WinDLL("crypt32") kernel32 = ctypes.WinDLL("kernel32") crypt32.CryptProtectData.argtypes = [ctypes.POINTER(_DPAPI_BLOB), ctypes.c_wchar_p, ctypes.POINTER(_DPAPI_BLOB), ctypes.c_void_p, ctypes.c_void_p, ctypes.c_uint32, ctypes.POINTER(_DPAPI_BLOB)] crypt32.CryptProtectData.restype = ctypes.c_bool kernel32.LocalFree.argtypes = [ctypes.c_void_p] buf = ctypes.create_string_buffer(data, len(data)) blob_in = _DPAPI_BLOB(len(data), ctypes.cast(buf, ctypes.POINTER(ctypes.c_char))) blob_out = _DPAPI_BLOB() if not crypt32.CryptProtectData(ctypes.byref(blob_in), "RRPP nodes.json", None, None, None, 0, ctypes.byref(blob_out)): raise OSError("CryptProtectData 调用失败") try: return ctypes.string_at(blob_out.pbData, blob_out.cbData) finally: kernel32.LocalFree(blob_out.pbData) def _dpapi_unprotect(data: bytes) -> bytes: """DPAPI 解密 CryptUnprotectData。""" crypt32 = ctypes.WinDLL("crypt32") kernel32 = ctypes.WinDLL("kernel32") crypt32.CryptUnprotectData.argtypes = [ctypes.POINTER(_DPAPI_BLOB), ctypes.POINTER(ctypes.c_wchar_p), ctypes.POINTER(_DPAPI_BLOB), ctypes.c_void_p, ctypes.c_void_p, ctypes.c_uint32, ctypes.POINTER(_DPAPI_BLOB)] crypt32.CryptUnprotectData.restype = ctypes.c_bool kernel32.LocalFree.argtypes = [ctypes.c_void_p] buf = ctypes.create_string_buffer(data, len(data)) blob_in = _DPAPI_BLOB(len(data), ctypes.cast(buf, ctypes.POINTER(ctypes.c_char))) descr = ctypes.c_wchar_p() blob_out = _DPAPI_BLOB() if not crypt32.CryptUnprotectData(ctypes.byref(blob_in), ctypes.byref(descr), None, None, None, 0, ctypes.byref(blob_out)): raise OSError("CryptUnprotectData 调用失败") try: return ctypes.string_at(blob_out.pbData, blob_out.cbData) finally: kernel32.LocalFree(blob_out.pbData) if descr.value: kernel32.LocalFree(descr) def _nodes_encrypt(text: str) -> bytes: """加密 nodes.json 内容；非 Windows 环境明文（dev/测试）。""" if os.name != "nt": return text.encode("utf-8") return _dpapi_protect(text.encode("utf-8")) def _nodes_decrypt(blob: bytes) -> str: """解密 nodes.json；解密失败回退按明文解析（兼容旧文件与在线拉取写入的明文）。""" if os.name != "nt": return blob.decode("utf-8") try: return _dpapi_unprotect(blob).decode("utf-8") except Exception: return blob.decode("utf-8") def _auth_save(username, password): """登录凭据加密保存到用户数据目录（DPAPI，本机本用户可解）。""" import json text = json.dumps({"username": username, "password": password}, ensure_ascii=False) _USER_DIR.mkdir(parents=True, exist_ok=True) (_USER_DIR / "auth.json").write_bytes(_nodes_encrypt(text)) def _auth_load(): """读取已保存的登录凭据；不存在或损坏返回 None。""" try: import json f = _USER_DIR / "auth.json" if not f.is_file(): return None data = json.loads(_nodes_decrypt(f.read_bytes())) if not (isinstance(data, dict) and data.get("username") and data.get("password")): return None return data except Exception: return None def _nodes_sync_remote(): """启动时用已保存凭据拉取最新节点配置（内容比对，变化才覆盖）。""" cred = _auth_load() if not cred: return # 从未登录过，等登录框处理 url = str(_config_val("nodes_url", "") or "").strip() if not url: return try: import json data = _fetch_nodes_auth(url, cred["username"], cred["password"]) if not (isinstance(data, list) and data and all(isinstance(n, dict) for n in data) and str(data[0].get("domain") or "").strip()): raise ValueError("返回的节点数据无效") text = json.dumps(data, ensure_ascii=False, indent=2) + "\n" local = _USER_DIR / "nodes.json" if local.is_file() and _nodes_decrypt(local.read_bytes()) == text: log("节点配置已是最新") return _USER_DIR.mkdir(parents=True, exist_ok=True) local.write_bytes(_nodes_encrypt(text)) global _NODES _NODES = None # 使后续读取（节点菜单/开启代理）拿到新配置 log(f"节点配置已更新: {len(data)} 个节点 (nodes.json)") except Exception as e: log(f"节点配置在线更新失败，使用本地配置: {e}") def _fetch_nodes_auth(url, user, password, timeout=10): """带 Basic 认证拉取 nodes.json（参与 nginx 认证；直连，不经过系统代理）。""" import json import urllib.request opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) req = urllib.request.Request(url) req.add_header("Authorization", "Basic " + base64.b64encode( f"{user}:{password}".encode()).decode()) with opener.open(req, timeout=timeout) as r: return json.loads(r.read().decode("utf-8")) def _user_init_local(): """初始化用户数据目录（同步）：建目录，nodes.json 缺失时写入加密空模板。""" _USER_DIR.mkdir(parents=True, exist_ok=True) if _user_path("nodes.json") is None: import json template = [{"domain": "", "port": 443, "country": "", "username": "", "password": "", "remark": ""}] (_USER_DIR / "nodes.json").write_bytes(_nodes_encrypt( json.dumps(template, ensure_ascii=False, indent=2) + "\n")) log_red(f"已创建 {_USER_DIR / 'nodes.json'}（加密存储），请配置 config.json 的 " f"nodes_url 在线更新，或使用 others/nodes_tool 导入节点配置") def _user_init_remote(base_url=None): """检查并更新 PAC 数据（后台线程）： 直接拉取两个数据文件（内含 version/date），按版本比对后更新；失败用本地副本。 节点配置的拉取走登录框（nodes_url 需 nginx 认证，见 LoginDialog）。 """ if _skip_update_check(): return # 设置跳过：不拉取也不打日志 base = (base_url or str(_config_val("pac_base_url", "") or "")).strip().rstrip("/") if not base: log("未配置 PAC 更新源（config.json 的 pac_base_url），使用本地数据") else: import json updated = failed = 0 for name in _PAC_FILES: try: data = _pac_fetch_json(f"{base}/{name}.json") if not _pac_validate(name, data): raise ValueError(f"{name}.json 数据格式不正确") rver = str(data.get("version") or "") rdate = str(data.get("date") or "") # 版本只与用户数据目录本地文件比对（回退文件仅用于读取，不参与更新判定） local = _USER_DIR / f"{name}.json" lver = _pac_read_version(local)[0] if local.is_file() else "" if rver: if rver > lver: _pac_write(name, data) log(f"PAC 数据已更新: {name}.json {rver} ({rdate})") updated += 1 # 否则已是最新，跳过 else: # 远端缺版本号：本地没有 -> 兜底下载；本地有 -> 按内容比对 new_bytes = json.dumps(data, ensure_ascii=False).encode("utf-8") if local.is_file() and local.read_bytes() == new_bytes: continue _pac_write(name, data) log(f"PAC 数据已下载: {name}.json 无版本号 ({rdate})") updated += 1 except Exception as e: log(f"PAC 数据 {name}.json 更新失败，使用本地副本: {e}") failed += 1 if updated == 0 and failed == 0 and (_USER_DIR / "pac_domains.json").is_file(): log("PAC 数据已是最新") def _load_cn_ranges(): """加载 pac_ip_cn.json 中的中国 IPv4 段，合并为整数区间供二分查找。""" global _CN_RANGES, _CN_RANGES_LOADED if _CN_RANGES_LOADED: return _CN_RANGES _CN_RANGES_LOADED = True try: import json, ipaddress path = _user_path("pac_ip_cn.json") if path is None: log_red("未找到 pac_ip_cn.json，智能分流不可用（全部走隧道）") return _CN_RANGES data = json.loads(Path(path).read_text("utf-8")) cidrs = data.get("cidrs") or [] ranges = [] for c in cidrs: net = ipaddress.ip_network(str(c).strip(), strict=False) if net.version != 4: continue ranges.append((int(net.network_address), int(net.broadcast_address))) ranges.sort() merged = [] for s, e in ranges: if merged and s <= merged[-1][1] + 1: if e > merged[-1][1]: merged[-1] = (merged[-1][0], e) else: merged.append((s, e)) _CN_RANGES = merged log(f"已加载中国 IP 段: {len(cidrs)} 条 CIDR，合并为 {len(merged)} 段") except Exception as e: log_red(f"pac_ip_cn.json 加载失败: {e}") return _CN_RANGES def _load_force_proxy_suffixes(): """加载 pac_domains.json：命中这些域名后缀的连接强制走隧道（优先于 IP 判定）。""" global _FORCE_PROXY_SUFFIXES, _FORCE_PROXY_LOADED if _FORCE_PROXY_LOADED: return _FORCE_PROXY_SUFFIXES _FORCE_PROXY_LOADED = True try: import json path = _user_path("pac_domains.json") if path is None: return _FORCE_PROXY_SUFFIXES # 可选文件，缺失不影响 IP 分流 data = json.loads(Path(path).read_text("utf-8")) suffixes = sorted({str(d).strip().lower().lstrip("*.") for d in (data.get("domains") or []) if str(d).strip()}) _FORCE_PROXY_SUFFIXES = suffixes log(f"已加载强制代理域名: {len(suffixes)} 条 (pac_domains.json)") except Exception as e: log_red(f"pac_domains.json 加载失败: {e}") return _FORCE_PROXY_SUFFIXES def _is_private_int(n: int) -> bool: """整数形式的 IPv4 是否属于常用内网/回环段。""" return ((n & 0xFF000000) == 0x0A000000 or (n & 0xFFF00000) == 0xAC100000 or (n & 0xFFFF0000) == 0xC0A80000 or (n & 0xFF000000) == 0x7F000000 or (n & 0xFFFF0000) == 0xA9FE0000) @functools.lru_cache(maxsize=16384) def _is_china_ip(host: str): """判断目标是否命中中国 IP 段；域名会先解析。 任一解析结果命中中国段 -> True（直连）；全部不命中 -> False（走隧道）； 解析失败或数据未加载 -> None（无法判定，安全起见走隧道）。 """ import ipaddress try: ip = ipaddress.ip_address(host) addresses = [ip] except ValueError: try: import socket addresses = [] for addr in socket.getaddrinfo(host, None, socket.AF_INET, socket.SOCK_STREAM): try: addresses.append(ipaddress.ip_address(addr[4][0])) except ValueError: pass except Exception: return None if not addresses: return None ranges = _load_cn_ranges() if not ranges: return None for ip in addresses: if ip.version != 4: continue n = int(ip) # 0.0.0.0 / 内网 / 回环解析结果疑似污染或局域网，不参与判定 if n == 0 or _is_private_int(n): continue i = bisect.bisect_right(ranges, (n, 0xFFFFFFFF)) if i > 0 and ranges[i - 1][0] <= n <= ranges[i - 1][1]: return True return False @functools.lru_cache(maxsize=8192) def _is_force_proxy(host: str) -> bool: """pac_domains.json 命中的域名强制走隧道（如 claude.ai：国内可连但显示拒绝页）。""" suffixes = _load_force_proxy_suffixes() if not suffixes: return False h = host.rstrip(".").lower() return any(h == s or h.endswith("." + s) for s in suffixes) def _resolve_host_ips(host: str): """解析域名的 IPv4/IPv6 地址集合（字面量 IP 返回自身）；解析失败返回空集合。""" import ipaddress try: return {ipaddress.ip_address(host)} except ValueError: pass try: import socket # 双栈解析（AF_UNSPEC）：IPv6 白名单条目也能命中域名解析出的 IPv6 地址 return {ipaddress.ip_address(a[4][0]) for a in socket.getaddrinfo(host, None, socket.AF_UNSPEC, socket.SOCK_STREAM)} except Exception: return set() def _whitelist_entry_match(host: str, entry: str) -> bool: """单条白名单规则匹配：IP/CIDR 条目按解析地址比对，域名条目按后缀匹配。""" h = host.rstrip(".").lower() entry = entry.rstrip(".").lower() import ipaddress net = None # 只有含点/冒号才按 IP 解析：纯数字串（如 8080）会被 ip_network 误解析为 # 整型地址，应视为域名条目 if "." in entry or ":" in entry: try: net = ipaddress.ip_network(entry, strict=False) except ValueError: net = None if net is not None: return any(addr in net for addr in _resolve_host_ips(h)) # 域名条目：精确匹配或子域后缀匹配（与 pac_domains 同一规则） return h == entry or h.endswith("." + entry) @functools.lru_cache(maxsize=8192) def _is_whitelisted(host: str, version: int) -> bool: """host 是否命中已启用的白名单规则（按版本键控缓存，规则变更即失效）。""" entries = [r["value"] for r in _whitelist_rules() if r.get("enabled")] if not entries: return False return any(_whitelist_entry_match(host, e) for e in entries) def _decide_direct(host: str, bypass_lan: bool, smart: bool) -> bool: """判断连接是否直连。判定顺序（白名单最优先）： 白名单命中 -> 直连；域名表命中 -> 隧道；内网 + bypass_lan -> 直连； 中国 IP -> 直连，其余 -> 隧道。 白名单在 PAC 与固定代理模式下均生效（固定模式除白名单外全部走隧道）。 """ if _is_whitelisted(host, _WHITELIST_VERSION): return True if smart and _is_force_proxy(host): return False if bypass_lan and _is_private_ip(host): return True if smart: return _is_china_ip(host) is True return False def _parse_connect_authority(first_line: str): """从 CONNECT 首行提取 (host, port)，失败返回 None。""" parts = first_line.split(" ") if len(parts) < 2: return None authority = parts[1] if ":" not in authority: return None host, _, port_s = authority.rpartition(":") if not host or not port_s.isdigit(): return None return host, int(port_s) def _parse_http_absolute(first_line: str): """从明文 HTTP 代理首行提取 (host, port)，失败返回 None。""" parts = first_line.split(" ") if len(parts) < 2 or not parts[1].lower().startswith(("http://", "https://")): return None from urllib.parse import urlsplit parsed = urlsplit(parts[1]) if not parsed.hostname: return None port = parsed.port or (443 if parsed.scheme == "https" else 80) return parsed.hostname, port def _force_http_close(buf: bytes) -> bytes: """直连明文 HTTP 时强制 Connection: close，避免 keep-alive 复用错连其他主机。""" end = buf.find(b"\r\n\r\n") if end == -1: return buf head, tail = buf[:end], buf[end:] lines = head.split(b"\r\n") out = [lines[0]] conn_seen = False for line in lines[1:]: low = line.lower() if low.startswith(b"connection:"): out.append(b"Connection: close") conn_seen = True elif low.startswith(b"proxy-connection:"): continue else: out.append(line) if not conn_seen: out.append(b"Connection: close") return b"\r\n".join(out) + tail def _tls_relay(client, remote_host, remote_port, ctx, use_tls, bypass_lan=False, smart=False): """处理一条客户端连接：内网(bypass_lan)/中国IP(智能分流)目标直接连接；其余走远端隧道。""" import socket, select client.settimeout(2.0) buf = b"" is_http = None is_connect = False injector = None direct_target = None max_headers = 65536 try: while len(buf) < max_headers: try: chunk = client.recv(4096) except socket.timeout: break if not chunk: break buf += chunk if is_http is None: first_crlf = buf.find(b"\r\n") if first_crlf != -1: is_http = _starts_with_method(buf[:first_crlf] + b" ") if not is_http: break if b"\r\n\r\n" in buf: break if is_http and b"\r\n\r\n" in buf: is_connect = buf.startswith(b"CONNECT ") first_line = buf.split(b"\r\n", 1)[0].decode("latin-1", "replace") if is_connect: target = _parse_connect_authority(first_line) else: target = _parse_http_absolute(first_line) if target and _decide_direct(target[0], bypass_lan, smart): direct_target = target except Exception as e: log(f"[转发器] 读取首包失败: {e}") client.close() return count_traffic = direct_target is None remote = None try: if direct_target: remote = socket.create_connection(direct_target, 10) if is_connect: client.sendall(b"HTTP/1.1 200 Connection Established\r\n\r\n") buf = b"" else: buf = _force_http_close(buf) else: remote = socket.create_connection((remote_host, remote_port), 10) if use_tls: remote = ctx.wrap_socket(remote, server_hostname=remote_host) if is_http and b"\r\n\r\n" in buf: if not is_connect: injector = _make_request_injector() buf = _inject_proxy_auth(buf) if buf: remote.sendall(buf) if count_traffic: _traffic_add(up=len(buf)) except Exception as e: log(f"[转发器] 连接目标失败: {e}") try: client.close() except Exception: pass if remote is not None: try: remote.close() except Exception: pass return client.settimeout(None) remote.settimeout(None) # CONNECT 隧道承载 WebSocket / 推送等长连接，空闲属正常，不设超时； # 明文 HTTP 连接空闲 120 秒回收 idle_timeout = None if is_connect else 120 try: while True: r, _, _ = select.select([client, remote], [], [], idle_timeout) if not r: break if client in r: data = client.recv(65536) if not data: break if injector is not None: data = injector(data) if data: remote.sendall(data) if count_traffic: _traffic_add(up=len(data)) if remote in r: data = remote.recv(65536) if not data: break client.sendall(data) if count_traffic: _traffic_add(down=len(data)) except Exception: pass finally: try: client.close() except Exception: pass try: remote.close() except Exception: pass def _tls_stop(): """停止 TLS 转发器监听线程。""" global _tls_forwarder if _tls_forwarder: _tls_forwarder["running"] = False try: _tls_forwarder["sock"].close() except Exception: pass _tls_forwarder = None def _ensure_forwarder(): """确保本地转发器在运行；未运行时启动临时全隧道转发器（静默），返回是否由本次启动。""" if _tls_forwarder: return False _tls_start(_endpoint_domain(), _endpoint_port(), use_tls=True, bypass_lan=False, smart=False, quiet=True) return True def _release_forwarder(started_here): """停止由 _ensure_forwarder 启动的临时转发器；非本次启动则保持不动。""" if started_here: _tls_stop() def is_proxy_on(): """系统代理是否生效：固定代理已开启，或 AutoConfigURL 指向本地 PAC 服务。""" r = _run(["reg", "query", PROXY_REG, "/v", "ProxyEnable"], warn=False) if "0x1" in (r.stdout or ""): return True r = _run(["reg", "query", PROXY_REG, "/v", "AutoConfigURL"], warn=False) return "127.0.0.1" in (r.stdout or "") def _cleanup_stale_proxy(): """清理上次异常退出残留的系统代理设置。 TLS 转发器 / PAC 服务只活在进程内，程序启动时若注册表仍指向 本地转发端口或本地 PAC 地址，说明是残留——不清理会导致全网不通。 """ try: if not is_proxy_on(): return r = _run(["reg", "query", PROXY_REG, "/v", "ProxyServer"], warn=False) r2 = _run(["reg", "query", PROXY_REG, "/v", "AutoConfigURL"], warn=False) stale = (f"{PROXY_HOST}:{PROXY_PORT}" in (r.stdout or "") or "127.0.0.1" in (r2.stdout or "")) if stale: proxy_off() except Exception: pass # ── 图标 ────────────────────────────────────────────── # 程序图标 base64 编码（原 ico.ico），避免打包后依赖外部资源文件 _ICO_B64 = "iVBORw0KGgoAAAANSUhEUgAAAEAAAABACAYAAACqaXHeAAAfqElEQVR4nMV7CZClV3Xed++/v33p193Ts7UGabQOICSDAWGBCImR4zgmjFhjIsAhKYeUK46xUpgaERuIVRRVxCmnAGODkyrioYxEMJjFILEYKO0aZkaj2VozPTO9v37rv94lde7/3kz3aEYMRCRX9dTT/d77/3vOPct3vnN+hhdoaa0ZAIsxJi76+7VS4hbLwi0AbmyvdncsL3dbUZiVm62yu3VbE4yxjHM+ALAKYB7A0wAeB/AogMOMMbnhejYAxRhTL8S+2QskOD+/Sa2ZBl4F4J8BuAPADQAKaaqxcHYVGhqNRgmVauFKLp8CeAbAQwC+AuB7jLFkdF9rpAiN/19L55vI/93vT2qt/8Nw5eyTy8ePaiW0WVJIfeCpE/LggTnR6w2l1lqNXlop82O8xn+nz4jRz4vXM1rrfVrrnRv3MDqEn2uxn1NwTj9I+72zZyccW7zfn5h5H7g9BTFE++hBnQgtF9bBU6/GZq+aYtNT9fF3zW3Zld1Zj15k7nz0otUF8DkAn2CMnR7v6YVyiys/dS3+tRbZvFo4rpPuitY6zbKoJ+PTB/WpR36kH3v4kO4PY32J0/55F1lFdv4XKdpaiHu0PuhuiA+/uKUffNDcQMfxtVqKb403Mlw6nZ07/LgylqtifeyRR/TBQyd1OhL6BRJ+41KbFCGyR3WavuKFcIlLLrrgyOwhkvAdUmYdc+cszGQ6VFrG+uzBJ7RMh/rU8Tl9+Oi8Xu2FZA/6F7mklEqmibmJzNJUp+nvbtjvFSmB/bQPjC9E/q61/hiAewABZKmEZRl30GDoL5zDmcV1OM1pVKtlZKkA40DRd1EIPDj2ec95QZdWCkwJCa04HJ8hy/4SBw68j916a3YlcYE/78W1Zvfeey8pgA07i38FiHvC9bNShEMNy7GSfhdLzxxGe/4UMm1D2AGq9SqyjIQ3X0NvmKA3CE3wi5IUUr2wcYpxDs25RaKoOBRwnLux56avaq1LJPzYci/7fTyP8OP3hxn2Fx38i/X+MMNwzSkGPtzqBFQSIpMSdP8TJ5dQn56GlNmmy5Lgk80SskxhtTuEbXG4tgXXseA5NhzbxqDfQ6lcBufPu9fLL62hlQQj5aosg1d0kCY/WO3175xotQaETS5nCfxywj+UozrVT/B5Er4dIYNfdKRTJI0j6nWguQWvUMXyYhtuqQQhBZQ2+zFLKY2C78KxbPTCCDYJqIEklegNYrQHMY4dncN6LzQWE8Up4lRASGm+e8WLMTBuQYJBgDtIogy2e5vL+P0YZYbLxQT7Mpe0XseY6Cb6Y2UX7yThGYND2Zi5JXSjHrjFsHr6GbJBhE4VW7ZVjeSZkJBSQSoNMsxqyccwjpGmZCn8vG1Yto2o1zObrzXqRuD1fgg6RIod3LIgohCO66BaKSHw3JELaVicgTMO0qdtWRSfzHU6gwhSCEzWSo6IB1ml2Xh9f3nxLytTW945ymDipypAa23wfCfUbyu7uKeTIGOAk78JWK4H22sBFNPiIRaOHEHzxTtRKfgbr2I2SgohoS3GUS0FUGQaOrcMpQSYTFCfnETg2lBKEZ4Gt5lBPqQqEiwc9FGrVcxVkySDIAVYDDa3YFvMxOGxUsvFAForAuawLMdBGmfleu0dJ48cOcCuu+6+Bx980H7d6163SQnsUggvjrFL2XhCKhQzaZS7yVVIEN8GuEhx4uizuP7Fu8lfcGUrB3XD3jpsx4UXFMf3Hut4YwTBoL2KcnPyCq89+hYpVyhACyo95MqZU1xDvmbqmj0/1Pv3W+yuu+TlLIAxzlQn0p8t2Sh3Qkg6wItvYEGj6HCcW+2jNdU0wtP+SQckiDHJSyyRpYgGPZTrE1AigVuubbhx/p3N32TG1aQUsKwrB3kUDwxelRZjFrgm4ypU/0JrfTMZEsWDcRHFLzJ9mWT6PVUPt3ciiEsJT4KWXCBWQJoK1GulkQBAe3UVUp5X7uYvAYh6awhKFSiRwrZHvmvO/FJLn09zSlBm+VkWAwUICtbgFld+SbS2bb8WwAdH2eC83HwkPO1EdTq6zhk+Qv/mGvzirZFbFxxtjHiYaFhMw/U9816/2zW+bdv2eXOmlYQ9KK0IQcKh+OG4kCKG7XpXhMUoyOVu8zOqgHGjPDqZQuBZiys9usjvaR29KPfiHB/wsSLIJKwAv2tzTEWZUhVfGygzFoVk8iwNhwOR4JCZMEFoLEA0DNGYaG5yAcrNWRIanJAlQzj+yN8pZ1t5XN3ot5ddGxT6s2khV4DtuiyJyGbhD2J738j8zSb5yB+oympo4Hf6GfQwg5VKoOIp4wN08jbTKNgag3RUn4oMnpsLIQT5qGUEC4e98/eXxtRz3yXL4yM/5lBGKeOVhF1kabhB2AsCp1n2c9M2+UFwMNuBkMrqSSgw/lat9e6RzJwswOxkkOFdZRfNTEBanLNIMKQZQ8lTcLhG4AKJZJDauBcUBaYRvhdZBse1kcQxtLhwkpTQbDs3dU1aHAlCKY8EVVIiSxMkw3W4jg+dJXQxaPMSUFIhSRODCf5vFqViMkwloUoep1P7d+O3bDooCoCdGO+NKbyP3IKUFxmBGSpefjADkQulx5hgDF21hOPYEEoZkLIxXWbRGoRMEPXXYNsc2ioijAWsbADGPcikC5/cISMoS8ojUJNH8oSUSdZltnlxirzyRZiEYgIdttmT1G/TWv8hY6xnfD/K8PLAxQ1hSizPhsxgNKGNIPSfa9G/R2+alDfWcG7+BP3G78s0RGd10QAXZVfMywmIFXIh20cRd1cgBm2ouAeXFBkPgCQ2L5amgEiR9tZgiT7EYAXZYBlRZwEiIe70ymLCeH9SKFiuS9iFUQXt2nwikvhVs3f6n5D4Tc9AoAvhlr5Lfh84Gv0E6CcMgQMENhUedEIsN2X6nOsZBRAa7y2cMH/LpEKmHfDlo8aKdLgEnQ7BZIzC4ml44Sp00ocrpQEtOkkAcgHyeRFDRTHSKISlhrDdGmy7CM5dRL0VyCzaJKi46PcNKsj3kpEVMZCMUcYMHhUCbzIKoECggH9kqNYN+ZHiZMnViFMGoRkou/cSBs8mJZCL0Knn/m7grm1BZyHSNMKg0wYLSrDiLoLlNuRgGSzsQdOJDtfgNq+D4wMs7sAiy0kTsDSBphcJk2aI4xR22oc1fwQy6oDBMi4jlAVrFFfGK4uHY8e85CJs4tvaxK9UwYoUmNR4jda6zHsJXqSB6+M82BoXIxUVHQ2hGBKy7NEbaqQEm4gOR4FZY31puEEZSW8Bbm0n+oMuev0IttTgW29CMSjAp9jfWYaibDB7HRyvCjsLoYU0J08BkBTAsgwqSUxWCNrnEPQciBNPATIB18zUAzLubzphY5LPkyoNGaM1YpKFM5Yk0LaNmVjgVl708EsFH77IKEXA2IdnUxEDhBmVmRfCzhgXDIkQylJ4ruEizb0pxRVqU2BZF9pvQKyfgSpVoRtNOMxBYeZG6GIDen3R3MRyJ+CkFPhICTE5J5jMINOBMVm7vwQvToCX3gZn8mpkZw4bJcAuIuqvG6FNZtmoiOesfO9knVG2wb8BRbWM1HgV4eRbSAzKFCbf87zQGVK+v0TIpewgFRBHCdwgGP0t/6Bf3QqvVIHn+ghkH97y4dw0lQAvBHAmd8KubTOpDoUSrPoWqPVjkEnH7FVkPWRJZD5vRz30pnejx10khRrsqV1Iu2fA4i7iMDbutnFPl7KAcUlCCiClblykDKlxqz3IcGPZA8qOZFHGTdCLszz9PbemoRqfIez2DdFhov+m4odKVRuuz+BOzEDMnwSYg0xrnHrsAPrnFrBtzw2Y2NmgQgLW9CxcJaDWz0JTppCS7gDZO4vjT57C6dOPYBD14DZbePVdb8b07hfDXjiO9fkDiIsegomrck7wpyRH3/cg0zXDNYwWz/LK/FoK9DsjUp5grOwzEAJMBJn+pS9GKUUMeqju2DKqAC8IT8dIeMCBBHdKYJUmrEYN3/6zz+JrH7kPTZFi++w2vPlzn0Rp9sXQSQR7x40Qrg/ZOWe+I9I16FIBh77yNRz6wZM4S1km7uK7n/wkbn//+/HP7/kDTEQhonOnoSeuMmeZH/6l3YAOiICU51jodwaoT5QQR4qJjNActtlgmJACcCiv08044FgaqSLWZfOFHJdh+dQCtrSqeaGx6Z6jclZLWKRpZsNp1PA399yLB/7k45j8lTfia9e/G3GqcPWKh9uu1lAJYCVD2K0dEEEAuXQMKg3hzezCbXe/By/992V84PA0njy4gNcc/St8ed+9WD4xh/d96r/BOv440kEPXrFsrGAjREqiIbjtwHHIuXPBpmdaOHhoDra9DdVGAXFk3L5sK6XKNuPwHMZ6sTZCF12AZdpAX/qdhPcDhjMnFlD0LRTKJWOuF0NUYnmIDnfsCJZnY+HUGr73iT/FLb/1b/CKj92HuR9IfPtYigeXl3CbTEfFiknksEstUFtVL83BSTKsBi3MT12LO8pFPBTuRPjKm/D6ygA/+ev78eirX42Xv/VOdDpr8IiLFGITBxGGAwTFChznQhzwPAfXXLMND//4KIqVEm665WqILJfXJb+PUgoKDJliGKTMlL0mdyoNz2M4duAU5p6ZQ7FUxJmTc8gIrV3sHsnQBBsRd8G0DW17+JU33I6XvenN+Mz313F9KcK79lB5akOG2QUobSJrBqfQgjf7CrDGdqSugwcOS8ytCvzWDQo7mw6ueeWrMVMpY/mJpwCvYEjYOBpR7UTKjA+CgqzxYY1Op5tnrjBCd0AVaxmNiYpxG00WT+QGBUgy+bG2hCbkx1F2lQFEmWDwi77x3x6lB9s1NcEG/zAnSaansxhuZYsJcpM7ZrDj9tvQuOV6vKVr4bqdDbRKwNOHu0ilQuCOdjF2IaKyKBX5LuqzW/E7u8vYMU1sMzC/0kfQvA3HPvd5vOQ33mjAk6Uz9Lp9+GNDHBULBHyoloiTFEvnltFZHyBMCQ062H3NDNxSgH5oMAFsoZAlkrkbIz790yC/lKHsagObtsxOmb/T5warqxdsa3RnQoXxYA2l8gQ4d6DJt0tFVGbqqAzamGQe+ssp2qcitHSEwPZM1UcbveC/I9whJLbWXBw8t4hVVkeBZWhawHBpEa967zux/dYboZeX4UIgSkPwYEPHWUsQHUju2e8PUKiUUW5NoUZY0hoVcUohcBhiwWCnig+4hYYQphA6LxWBFbpsLx1ZgtQGG1g2AyfcP6rz8zTIkfY7sLSCX6oDyQCMORBRB7OvuBEOMtTSDjoLIQqBj2KtAt0fAEEB2iW8wUH/XQgmGqXAxcu2ltFtrwFhHzWPQ+9swt3zBqgFyhgeuI7BpYJb3TKyQgaZpaOsQOW8MIowdYvQyEwaZBCSG5hvGdqLYW0z35ibs7YZtUbMr1QIEeoteQw+1zBvjRQwBiDh8kkUSk2zeUKmhhoXAkmokIYpVJSi5QB+FkN0egTgjZuQyWtBm97MJWoSzAZaHtCyCcAPDC8hu30oNzBNGTbswUracPyysSYTh6SAtiyzX7omKWA0kmACJVkJbS8UDJ7Fhhwa8yTQOJESEcZEBre9AmvYz+kSxjDIODybwdbCNEMNM0Cnz3mOzS0XXrmV+3EeiUyQ0vDAOguwVGYqxLw3mMcWc0ut8uKGbc4oZIxaaohMIEkERCKRDWPTLFFL58DoJynFoozrE0NjvpclMSzLHZOd4/7tZrJdQZNyAZwj2Z+2aJaHWu4Wh9PvoPjsATAqTamAmJxFtG03HJlAa8+0tKhbkwfv3OSyNEbQ3G4+L0UEizmQcggpNLjvwCrZsL0qtIxzytrxoMns6MXoDEiledS+2BKtwDdluYqoSjQFPTExgO0YziCob4WkapAp2F5gWCnbzRspeVv7uWwSeYTPYWVSHSNe8zE6E804s6RE8dQh8LAPbTvQlgN/ZR5OdwW60cJw+24MhhGmp2rm9BfnnjZ+VK5PGuori/tQdAKBhzQZQGkXzHGQqHlYrg1mN3LGmDbveQAVUwZQEdGSgY+Yn5F2wUhBfiGPEY4DnQVk46AEr4dDA7udUgvxYBV+bQZaJIZFKpbzIo0O/+IymZJMYJAaR5Txx4ilfjiMVcZd7tjtdbA0QjS7x+TlZGIHyscfhSUSDKZmkSQSNiRc10O/s4ZCpYFKcxK95bNwygHiqI+g0IAWQySphuYENBVkdRfCbA2F2i4wyzO8CwlnKEnK1+Qq5BOkADXuLjPj59qS4EEALT1j5hRw6XuDhZ/A23G1MXlYDrjnI1pfgGLe+SrVcxzEm/pMgMsB1wKPyJBs/JC7wDMMOGpyKXU1tUZarCLc8SI4YQc8GiAr1qECF8O1NsrVMsJhH47nozoxbfCA0swQIUoyWK6PeDiAAPn12JQ5UiEQdlYgdAxG9LjlglG5SXyd7UJ1l0zWMEogK7FsqP4qspWTIPaEmF0WlKEcjkF7Hqpah12ZMX2HQnUSKuqjs76SWwXheQI5tr2pp0eZzbeVlpyzOMNK0cbDNjVCe4l+MABu6AZlxTjj5aOPQBaqsCmdyQy6UEUaKhOtHaeIcDhEY6JlGpHnsUGvj8AvEBmImCoqq2hOk6o7aAHba0ChgGRlFaI4gFtpgTu5SRu3CAcGe1heMa/utDINFBZUciuxLKThGuL+EEo5KLd2Ix6so1ibhkxjtNeXoIMWAkuYJqvxdcoASV4GE69ZdgjZQtkWuM3xI8ZYO2eAFR4gwlf4BR5P7wInWNpbBUtjiEIN6dQOrM6fxfR0A912B9V640JkZZZpdcXDnhFIiQiS2fnkxqhtTSavuAvqxWi7ADEUiJbmkbRPQVFgdFzwyR3gQRVZ3DXkqhl68AMwt4AsWsdg9QzCzhBCe/ACB1Ik4F7ecW6vnEPGKyhVavB9CsA5THccB0qlxu+JDyTkN0w1NURJpQ8YK6HpiRLwg26CucDRVw2ndinmBdwerCKFD7V9FguLHbTqJSRxAtcP8iYIpcCRGqguiNcXobftRhp1QO0UU0mO+IK8V5cjPuP/FGDhQ547CSs6DNmchioUDDCyggaSwQIsr2SKJKklkv4Q2i6BgAGXISzbQao5ZNjL71vZZcy9VPBM5kmGPRQqE7BsF5yoep33NQaJ1rbLrV6KXsXF3xorIW+j8VMOfD4gbAyl7OktGM7uQXLVNehFGnLQx8RkDWtrXdSb9QskyMjUXLoZWYoEev0BFA1C0MkzDknkKQUu83n66UASXqdURkxPqmAvLMI5dhT82EHq6+cg6vFvw4picNeH5oGJDYb6IFdIIyTDEOH6PHgwAS1TBAXfnLDjFgyIEpTGGYNL/IQIkVGhpyErDiPvup8xRoONhtU0uK3s4c9TqYZFF1aUSC1SBS41Fk/M45prZnDuzBJq9SoutUrVBpxiBdnScfDuoml6KIsiMUV2EpwMLTOpjvrtpAQCSqnoQhmgQhZCkbyOrL8Kr7ELbHK7yUTUQ6ROrbQsKG5DZn3EUc+0rtnEjZCWDxpTLhRyes5E/1IVYW/d/LtYLqHXWUckDHvNIwEV+PjkBUzAzCQVWcFZz+J/YVOEFEy6AceZuWXMtMjkGZ565MT5SY1N/X8iSmyOYq0Jp30GXmcZeu0MhIigyNTp1G3yRQHJ6HdKbS4y0UVqcbAshSpWoK+/GdZLXgPtOMj6C7B3vQyssdXMIujes1AUa5CiGEYodTqwRQT4ZShuwXH4qCOVd2tsxwPjtqlObdeH7zpI+h1Z98HTTH3VY+yJ0QidPM9rU5M0BP6LVKpX8TjvrEVa9jqY2rEFj3z/EFaWaC6Im37dpjVSRrHWQtutoD11DezZl8CWoTFZwuWSTlGGRhHSsiF1BuHXkG15ObpX7Qa78WZYrRmI/hJ0dyUfjNAMllcB4y6KxTI4YiirgAgCSToETwY5JLf8EXurN++nUkevvWo6ILVaRVtRj2KVLAT8Q+eBBs73AfOhgSJj5+JEf5RiTfv0vNxF5KUS+PEPj2H7bAvPtyQVI0EdhUoTjaktqE9tBQ1OS3AI5kMkkYG/dPqKZrkKTXDLRb+6HSIZmsIoW30WVmUqb6Fpkb9oEm36OjjEFskYcVBCdPUrMSxM0NiqwQux1CYTbbRKSoGOV0BnZZkAgWxt22p126uf8Rh7anz6uGhMTu3fv98qBtYnjs8tH9jWqtrFRlGunV3DzLYmduwkBRDz8lzy0UR6i6PWrJu0ZCZAgiKKgWto9kajjEJzC0Q6gFZDVJoNtFoTmKz5qBaLyLrr5hAt5sMu1qA3zhoazsSGpuKH8H6phdbUdlgEzihVkmtoDpGOelsGeOWFWrXZxOkTcwpCWJBqoTU988HRYMR5M+YXLJnpvXv30rBAVuDi7karIAikdLtDfeOereBmXmQjg3NBeFNmihTf+dFxdFYWwGzbbKpQbeLMs8/i0/f/BH/z3TPwHGBiagp+ZdJ8hyixSrkM8m596iSSpVOIw3Xjv9R3NB11j+r+AfzwHCDJpz3D8we+b1gsY30SiKIoV8i4WTTiKfygoA4/epDBdt5HwCcX9cIp8s3unAfErbNbH4cQ/xFe2UqTTARFCirs0ifPGNZW2/i3H/suPvSlFF94cBGIu2bzllvA/GqCD3z+LP7ukEZ96zXmb+fvPiJdFAlz+inwm15lWnEijcEDD9xTWFlcwDcenMN/+qaPL3zrDGrFvIdf8oB6ANRKNtysjaKdj9nl03lEdppmXnbtLTfb7fUePVfwFRqT2/j4Da3njF6ZyQn6YLnxybX5szdPTjbfNUijTArtbJ7gIFDDwGSKP/7cQRxot7Dnhib+fn4Bv/LDZ3DHHb9sPrtzdiumG12UnRRZew7exNXn2RtzP0JszRnEN70M5YltpmbI1o7j2eOr+NJjAg8dA3rZNoSJQHD8EP7VrwtMFIB/eGIeX3pwDpVmCyXdxfVXx3jJ9Ro7d0zn52oGETnBwr+/7dde//uj5xyeM8HFcan12tdKmqf7r39+9L1O0f7WRLPhKEVFwahtTDUeTYIy4MmnF/H4UhmzM3WIJEalPolPfTvEuTPzRsgtEwVMlQ3IR38QorNyLk+jG5RAJGxCSI/OLO3j89+LcO9fL+HLBz2UfA+1ACjaAmsDhU4vAnSGx59exP94ooBvHq3gy3O7cN83HPz2n57ARz/zMDrttqDaWil1AJa1d2Tyl3y+yL6U/KPReH3vXhMs3rS6uPp35WL5NmiVMc2cnHLLU893nuqD03iLysysrmspnA1n8On7j+AD76kZNnnbhGOmypzGVWZaLOy3USg30Cfc4FfQXls3Pr++dBof/+Icvn58C7bXA+z05lGgAopNouYO4JYTPHbgFFpBil+7fTe+cfgZJH4BBY/IVBcWL+HLh4cisZbtD749eEao4M5ikXWeb2yeX9ICLqRGChiD1Q7uBNPfJlaCGDytlBkhowrwqbkUgcPPT4YIqWkqC8+ei/DsgScMHp8ocThMwLcUmq0pM2HW77ZRqE5jfW0Njl9CtVrFA986iP/1iI1f3hkisDJwu4w47KNiD8xzB2n5Btz7vwW+/sgCdu+exa//UhWdHsUbbpojWSayrVsm7a4oHHj29Pzri0V2dv9+A/IuO2dnX+6NsRJG2utrre9UQnyW2847hZSUS9RPjnX5woCjWMojMW2EGikElnhxAm65gpW5o2hYA7gFz3STyfLLjUl01pYw98whNKe2wg98ZGmESr0J21ZYiSsosGVq7CO1prA8LJprFwuAt7OCT/94DTdcfxJCSGRZRtFIxRnT1XrTubqVfPO+u3e8lTG2TsLfddfmoPczKWCshH379pESKDn/Sy3EQW5ZHyXr+YdD60Jxh0o+RrOWQuc9dmL5HBnCq78I6elDOLUwgD9Zz9PbaFis1pyCTlPUmxNmooRy/T957Y14cu4JHD97HPOxZ3I+NWodLmEzSuc0daaReg3cd38fbDhAubRNCNh2s1bEnu3iE/vetvX3Nxzc8wpP64qeUPjwhz9MF8zHaW37TzjwumEvPHh8vWhzy2JCKrIJw2pp7qLEeuiHQyTDGLMvvRl+rWlmDOl2G9v49S3bzc98ilyjVCzhQ++4ClvsJSwlVcN9OjwnNMy3R7OrJSfFUlqS0qnp2WpmT9Xt0//0pfZv3vv2Lb+XY6DLPyBx8eK4wkWBkTQ6yqXfO/Ldp18xOxN8ZFvTHzK3bsWCM0/0hKNjLbmPpiF7Ulh+GRkLUPSey86Oh6xozqC73sbK2VNw69vwj197PYosonHKDRsgYEN2piU9hVYsBlatXlUv3cH++5+9p3Lr3b/aemDvXnpiLN/rlcpl42dcNG8/qh5pROMPHz3S+5/7fzz4g7kV/+1xL3a5TNFjJVl0HdQDOkPO1taHuGHHeOT9wrTfuKpUSsINiiiViia9vurlV+MNR4/hoTlFg9n0sCDJTCSp5RdKVtUPsW0C999x3dTH3vDy1iP3vBvYu19bX7yLySue2v95FUBrNGbK7voi+K3XsSMA7v7+seTj33yo/9srfestZ9LydLExg8mduzDsD7DQSWWzaia7Rj6weZtkCK4XIChXyCp0sVjSb7+jpZ/4XJcPVY15nmcVXAslL17f2eL3375n4lN33lx7mB5hw979lt6/V12Jv/9C1r59mtMmxr9rret/9IUzb/nOj098QWt9+sjxRb3rbV/V3/rh3HOffhw9ULm+uqoXl1af8/4Xvn5M/8YfnVx8/6cW7v/4l1bfPRhognnjO/F9P+WJsP+na5/W/PZ9+ZOl46W1rh05vnLHu/74Rx88eHThS1rrw1LK7sUK6LbX9eKiUUBfSnlUa/23Wuv/HMfxG7XutTZJuVdbL6Tg/wcpnXblom8x+wAAAABJRU5ErkJggg==" def load_icon(): """加载程序图标，base64 解码失败时返回纯色兜底图标。""" try: import base64 data = base64.b64decode(_ICO_B64) pixmap = QPixmap() pixmap.loadFromData(data) if not pixmap.isNull(): return QIcon(pixmap) except Exception: pass # 兜底：生成一个纯色方形图标 img = QImage(64, 64, QImage.Format_ARGB32) img.fill(QColor(C_TITLE)) return QIcon(QPixmap.fromImage(img)) # ── 可点击节点标签 ───────────────────────────────────── _NL_BASE = f"color: {C_TITLE}; font-size: 12px;" _NL_HOVER = _NL_BASE + " text-decoration: underline;" class NodeLabel(QLabel): """可点击标签：hover 显示下划线+手型光标，点击发出 clicked 信号。""" clicked = pyqtSignal() def __init__(self, text="", tooltip="点击切换节点", base=None, hover=None): super().__init__(text) self._base = base or _NL_BASE self._hover = hover or self._base + " text-decoration: underline;" self.setStyleSheet(self._base) if tooltip: self.setToolTip(tooltip) def enterEvent(self, e): self.setStyleSheet(self._hover) self.setCursor(Qt.PointingHandCursor) super().enterEvent(e) def leaveEvent(self, e): self.setStyleSheet(self._base) self.unsetCursor() super().leaveEvent(e) def mouseReleaseEvent(self, e): if e.button() == Qt.LeftButton: self.clicked.emit() super().mouseReleaseEvent(e) # ── 节点配置登录框 ───────────────────────────────────── class LoginDialog(QDialog): """节点配置登录框：凭据参与 nodes_url 的 nginx 认证，成功后拉取并加密保存 nodes.json。""" _done = pyqtSignal(bool, str) def __init__(self, parent=None): super().__init__(parent) self.setWindowTitle("节点配置登录") self.setWindowIcon(load_icon()) self.setFixedSize(420, 170) self.setStyleSheet(f""" QDialog {{ background-color: {C_BG}; }} QLabel {{ color: {C_TEXT}; font-family: "Microsoft YaHei", "Segoe UI", sans-serif; font-size: 12px; }} QLineEdit {{ background-color: {C_CARD}; border: 1px solid {C_BORDER}; border-radius: 6px; padding: 6px 8px; color: {C_TEXT}; font-size: 12px; }} QPushButton#btnOk {{ background-color: {C_GREEN}; color: #fff; border: none; border-radius: 6px; padding: 4px 12px; font-size: 12px; font-weight: bold; }} QPushButton#btnOk:hover {{ background-color: #5ed978; }} QPushButton#btnOk:disabled {{ background-color: #2a3a2e; color: #666; }} QPushButton#btnCancel {{ background-color: {C_CARD}; color: {C_SUBTEXT}; border: 1px solid {C_BORDER}; border-radius: 6px; padding: 4px 12px; font-size: 12px; }} """) root = QHBoxLayout(self) root.setSpacing(14) root.setContentsMargins(20, 18, 20, 16) # 左侧 logo + 品牌小字 left = QVBoxLayout() left.setSpacing(6) left.addStretch() logo = QLabel() logo.setPixmap(load_icon().pixmap(80, 80)) logo.setFixedSize(80, 80) logo.setAlignment(Qt.AlignCenter) left.addWidget(logo, 0, Qt.AlignHCenter) brand = QLabel("R.R.P.P.") brand.setStyleSheet(f"color: {C_TITLE}; font-size: 10px;") brand.setAlignment(Qt.AlignCenter) left.addWidget(brand, 0, Qt.AlignHCenter) left.addStretch() root.addLayout(left) # 右侧表单（外层容器按内容高度垂直居中，避免被横向布局拉伸出大段留白） form = QVBoxLayout() form.setSpacing(6) self.ed_user = QLineEdit() self.ed_user.setPlaceholderText("登录名") form.addWidget(self.ed_user) self.ed_pass = QLineEdit() self.ed_pass.setPlaceholderText("密码") self.ed_pass.setEchoMode(QLineEdit.Password) self.ed_pass.returnPressed.connect(self._submit) form.addWidget(self.ed_pass) # 保存过凭据时预填，重新登录只需点一下 cred = _auth_load() if cred: self.ed_user.setText(cred["username"]) self.ed_pass.setText(cred["password"]) self.lb_err = QLabel("") self.lb_err.setStyleSheet(f"color: {C_RED}; font-size: 11px;") self.lb_err.setFixedHeight(14) self.lb_err.hide() # 无错误时不占空间，避免密码框下方留大片空白 form.addWidget(self.lb_err) bottom = QHBoxLayout() bottom.setSpacing(10) hint = QLabel("登录以获取节点配置") hint.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 10px;") bottom.addWidget(hint) bottom.addStretch() self.btn_cancel = QPushButton("取消") self.btn_cancel.setObjectName("btnCancel") self.btn_cancel.clicked.connect(self.reject) bottom.addWidget(self.btn_cancel) self.btn_ok = QPushButton("登录") self.btn_ok.setObjectName("btnOk") self.btn_ok.clicked.connect(self._submit) bottom.addWidget(self.btn_ok) form.addLayout(bottom) box = QWidget() box.setLayout(form) root.addWidget(box, 1, Qt.AlignVCenter) self._done.connect(self._on_done) def _submit(self): user = self.ed_user.text().strip() password = self.ed_pass.text() if not user or not password: self.lb_err.setText("请输入用户名和密码") self.lb_err.show() return self.btn_ok.setEnabled(False) self.btn_cancel.setEnabled(False) self.btn_ok.setText("登录中") # 登录按钮变灰并显示"登录中" threading.Thread(target=self._worker, args=(user, password), daemon=True).start() def _worker(self, user, password): """后台登录：Basic 认证拉取 nodes.json，校验后加密写入用户数据目录。""" try: import json url = str(_config_val("nodes_url", "") or "").strip() if not url: raise RuntimeError("config.json 未配置 nodes_url") data = _fetch_nodes_auth(url, user, password) if not (isinstance(data, list) and data and all(isinstance(n, dict) for n in data) and str(data[0].get("domain") or "").strip()): raise ValueError("返回的节点数据无效") text = json.dumps(data, ensure_ascii=False, indent=2) + "\n" _USER_DIR.mkdir(parents=True, exist_ok=True) (_USER_DIR / "nodes.json").write_bytes(_nodes_encrypt(text)) _auth_save(user, password) # 记住凭据，之后每次启动自动同步节点 self._done.emit(True, str(len(data))) except Exception as e: msg = ("认证失败，请检查用户名/密码" if getattr(e, "code", None) == 401 else f"登录失败: {e}") self._done.emit(False, msg) def _on_done(self, ok, info): self.btn_ok.setEnabled(True) self.btn_cancel.setEnabled(True) self.btn_ok.setText("登录") if ok: log(f"登录成功，已获取 {info} 个节点（nodes.json 已加密保存）") self.accept() else: self.lb_err.setText(info) self.lb_err.show() # ── 代理白名单弹窗 ───────────────────────────────────── class WhitelistDialog(QDialog): """代理白名单编辑弹窗：可编辑列表（Windows 环境变量风格）， 每条规则带复选框启用/禁用，支持添加/删除，保存后立即生效。""" def __init__(self, parent=None): super().__init__(parent) self.setWindowTitle("代理白名单") self.setWindowIcon(load_icon()) self.setFixedSize(380, 400) self.setStyleSheet(f""" QDialog {{ background-color: {C_BG}; }} QLabel {{ color: {C_TEXT}; font-family: "Microsoft YaHei", "Segoe UI", sans-serif; font-size: 12px; }} QLineEdit {{ background-color: {C_CARD}; border: 1px solid {C_BORDER}; border-radius: 6px; padding: 6px 8px; color: {C_TEXT}; font-size: 12px; }} QListWidget {{ background-color: {C_CARD}; border: 1px solid {C_BORDER}; border-radius: 6px; padding: 4px; color: {C_TEXT}; font-size: 12px; }} QListWidget::item {{ padding: 3px 6px; border-radius: 4px; }} QListWidget::item:selected {{ background-color: #e3e8ee; color: {C_TEXT}; }} QPushButton#btnOk {{ background-color: {C_GREEN}; color: #fff; border: none; border-radius: 6px; padding: 4px 12px; font-size: 12px; font-weight: bold; }} QPushButton#btnOk:hover {{ background-color: #5ed978; }} QPushButton#btnCancel {{ background-color: {C_CARD}; color: {C_SUBTEXT}; border: 1px solid {C_BORDER}; border-radius: 6px; padding: 4px 12px; font-size: 12px; }} QPushButton#btnSmall {{ background-color: {C_CARD}; color: {C_SUBTEXT}; border: 1px solid {C_BORDER}; border-radius: 6px; padding: 3px 10px; font-size: 11px; }} QPushButton#btnSmall:hover {{ background-color: #e3e8ee; color: {C_TEXT}; }} """) layout = QVBoxLayout(self) layout.setSpacing(10) layout.setContentsMargins(20, 16, 20, 16) title = QLabel("代理白名单") title.setStyleSheet(f"color: {C_TITLE}; font-size: 15px; font-weight: bold;") title.setAlignment(Qt.AlignCenter) layout.addWidget(title) hint = QLabel("命中以下规则的连接将强制直连（不经过代理）") hint.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 11px;") hint.setAlignment(Qt.AlignCenter) layout.addWidget(hint) self.lb_err = QLabel("") self.lb_err.setStyleSheet(f"color: {C_RED}; font-size: 11px;") self.lb_err.setFixedHeight(14) self.lb_err.hide() # 无错误时不占空间 layout.addWidget(self.lb_err) # 规则列表：每条前面带复选框（勾选启用 / 取消禁用） self.list_widget = QListWidget() self.list_widget.setToolTip("勾选启用规则；可多选后删除") for r in _whitelist_rules(): self._add_item(r["value"], r.get("enabled", True)) layout.addWidget(self.list_widget, 1) # 添加行：输入框 + 添加按钮 add_row = QHBoxLayout() add_row.setSpacing(8) self.ed_add = QLineEdit() self.ed_add.setPlaceholderText("输入域名 / IP / CIDR，如 example.com") self.ed_add.returnPressed.connect(self._add_rule) add_row.addWidget(self.ed_add, 1) self.btn_add = QPushButton("添加") self.btn_add.setObjectName("btnOk") self.btn_add.setStyleSheet(f""" QPushButton {{ background-color: {C_GREEN}; color: #fff; border: none; border-radius: 6px; padding: 6px 14px; font-size: 12px; font-weight: bold; }} QPushButton:hover {{ background-color: #5ed978; }} """) self.btn_add.clicked.connect(self._add_rule) add_row.addWidget(self.btn_add) layout.addLayout(add_row) # 删除选中 + 底部按钮 mid_row = QHBoxLayout() self.btn_delete = QPushButton("删除选中") self.btn_delete.setObjectName("btnSmall") self.btn_delete.clicked.connect(self._delete_selected) mid_row.addWidget(self.btn_delete) mid_row.addStretch() layout.addLayout(mid_row) btns = QHBoxLayout() btns.setSpacing(10) btns.addStretch() self.btn_cancel = QPushButton("取消") self.btn_cancel.setObjectName("btnCancel") self.btn_cancel.clicked.connect(self.reject) btns.addWidget(self.btn_cancel) self.btn_ok = QPushButton("保存") self.btn_ok.setObjectName("btnOk") self.btn_ok.clicked.connect(self._save) btns.addWidget(self.btn_ok) layout.addLayout(btns) def _add_item(self, value, enabled): item = QListWidgetItem(value) item.setFlags(item.flags() | Qt.ItemIsUserCheckable) item.setCheckState(Qt.Checked if enabled else Qt.Unchecked) self.list_widget.addItem(item) def _add_rule(self): value, err = _whitelist_validate_entry(self.ed_add.text()) if err: self.lb_err.setText(err) self.lb_err.show() return # 去重（不区分大小写） for i in range(self.list_widget.count()): if self.list_widget.item(i).text().lower() == value.lower(): self.lb_err.setText("该规则已存在") self.lb_err.show() return self.lb_err.hide() self._add_item(value, True) self.ed_add.clear() def _delete_selected(self): for item in self.list_widget.selectedItems(): self.list_widget.takeItem(self.list_widget.row(item)) def _save(self): """收集列表内容保存（未点保存前所有改动均可取消）。""" rules = [] for i in range(self.list_widget.count()): item = self.list_widget.item(i) value = str(item.text()).strip() if value: rules.append({"value": value, "enabled": item.checkState() == Qt.Checked}) _whitelist_save(rules) self.accept() # ── 设置弹窗 ───────────────────────────────────────── class SettingsDialog(QDialog): """设置弹窗：代理转发端口 + 退出登录。""" def __init__(self, parent=None, port_locked=False): super().__init__(parent) self.setWindowTitle("设置") self.setWindowIcon(load_icon()) self.setFixedSize(300, 354) self.setStyleSheet(f""" QDialog {{ background-color: {C_BG}; }} QLabel {{ color: {C_TEXT}; font-family: "Microsoft YaHei", "Segoe UI", sans-serif; font-size: 12px; }} QLineEdit {{ background-color: {C_CARD}; border: 1px solid {C_BORDER}; border-radius: 6px; padding: 2px 8px; color: {C_TEXT}; font-size: 12px; }} """) layout = QVBoxLayout(self) layout.setSpacing(12) layout.setContentsMargins(20, 16, 20, 16) title = QLabel("设置") title.setStyleSheet(f"color: {C_TITLE}; font-size: 15px; font-weight: bold;") title.setAlignment(Qt.AlignCenter) layout.addWidget(title) # 版本号 + 手动「检查更新」入口（整体居中，中间隔小空） ver_row = QHBoxLayout() ver_row.addStretch() ver_lab = QLabel(f"版本号: {VERSION}") ver_lab.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 11px;") ver_row.addWidget(ver_lab) ver_row.addSpacing(4) self.link_check_update = NodeLabel( "检查更新", tooltip="手动检查更新", base=f"color: {C_SUBTEXT}; font-size: 11px;", hover=f"color: {C_TEXT}; font-size: 11px; text-decoration: underline;") self.link_check_update.clicked.connect(self._manual_check_update) ver_row.addWidget(self.link_check_update) ver_row.addStretch() layout.addLayout(ver_row) # 提示语：标题正下方（保存结果/错误也显示在这行） self.lb_hint = QLabel("代理运行中，关闭代理后可修改" if port_locked else "保存更改后下次开启代理生效") self.lb_hint.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 11px;") self.lb_hint.setFixedHeight(14) self.lb_hint.setAlignment(Qt.AlignCenter) layout.addWidget(self.lb_hint) layout.addStretch() # 设置项在空白部分纵向居中（上下均分） # 代理白名单：第一个设置项，点「配置」打开编辑弹窗 wl_row = QHBoxLayout() wl_lab = QLabel("代理白名单") wl_lab.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;") wl_lab.setToolTip("命中列表的域名 / IP 将强制直连（不经过代理），每条规则可单独启用") wl_row.addWidget(wl_lab) wl_row.addStretch() self.btn_whitelist = QPushButton("配置") self.btn_whitelist.setCursor(Qt.PointingHandCursor) self.btn_whitelist.setStyleSheet(f""" QPushButton {{ background-color: {C_CARD}; color: {C_SUBTEXT}; border: 1px solid {C_BORDER}; border-radius: 6px; padding: 2px 10px; font-size: 11px; }} QPushButton:hover {{ background-color: #e3e8ee; color: {C_TEXT}; }} """) self.btn_whitelist.clicked.connect(self._open_whitelist) wl_row.addWidget(self.btn_whitelist) layout.addLayout(wl_row) # 代理转发端口 port_row = QHBoxLayout() port_lab = QLabel("代理转发端口") port_lab.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;") port_row.addWidget(port_lab) port_row.addStretch() self.ed_port = QLineEdit(str(_TLS_LOCAL_PORT)) self.ed_port.setFixedSize(54, 22) # 与右侧开关同高，和其他设置项同一水平 self.ed_port.setEnabled(not port_locked) port_row.addWidget(self.ed_port) layout.addLayout(port_row) # 更新方式：左文字 + 右苹果式开关（默认关 = 直连仓库） upd_row = QHBoxLayout() upd_lab = QLabel("使用代理更新") upd_lab.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;") upd_row.addWidget(upd_lab) upd_row.addStretch() self.cb_update_proxy = ToggleSwitch(checked=_update_via_proxy()) self.cb_update_proxy.setToolTip("开启后检查更新与下载走节点隧道（CONNECT）；默认直连仓库") upd_row.addWidget(self.cb_update_proxy) layout.addLayout(upd_row) # 跳过检查更新：左文字 + 右开关（默认关；不影响节点更新与认证） skip_row = QHBoxLayout() skip_lab = QLabel("跳过检查更新") skip_lab.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;") skip_row.addWidget(skip_lab) skip_row.addStretch() self.cb_skip_update = ToggleSwitch(checked=_skip_update_check()) self.cb_skip_update.setToolTip("跳过 PAC 数据与程序版本检查（节点更新与登录不受影响），被跳过的检查不输出日志") skip_row.addWidget(self.cb_skip_update) layout.addLayout(skip_row) layout.addStretch() # 底部：保存更改 + 退出登录 并排 btns = QHBoxLayout() btns.setSpacing(10) btns.addStretch() self.btn_save = QPushButton("保存更改") self.btn_save.setStyleSheet(f""" QPushButton {{ background-color: {C_GREEN}; color: #fff; border: none; border-radius: 6px; padding: 5px 14px; font-size: 12px; font-weight: bold; }} QPushButton:disabled {{ background-color: #2a3a2e; color: #666; }} """) self.btn_save.setEnabled(not port_locked) self.btn_save.clicked.connect(self._save_settings) btns.addWidget(self.btn_save) self.btn_logout = QPushButton("退出登录") self.btn_logout.setStyleSheet(f""" QPushButton {{ background-color: {C_RED}; color: #fff; border: none; border-radius: 6px; padding: 5px 14px; font-size: 12px; font-weight: bold; }} QPushButton:hover {{ background-color: #ff8787; }} """) self.btn_logout.setToolTip("清除登录凭据与本地节点配置，下次启动需重新登录") self.btn_logout.clicked.connect(self._logout) btns.addWidget(self.btn_logout) btns.addStretch() layout.addLayout(btns) def _manual_check_update(self): """手动触发一次检查更新（后台线程，复用主窗口逻辑）。""" parent = self.parent() if not isinstance(parent, MainWindow): return self.lb_hint.setText("正在检查更新...") self.lb_hint.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 11px;") try: parent._update_done.connect(self._on_check_done) except TypeError: pass # 已连接过，忽略 threading.Thread(target=parent._check_update, kwargs={"manual": True}, daemon=True).start() def _on_check_done(self): self.lb_hint.setText("检查完成，结果见主界面日志") self.lb_hint.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 11px;") try: self.parent()._update_done.disconnect(self._on_check_done) except Exception: pass def _open_whitelist(self): """打开代理白名单编辑弹窗（保存后立即生效，无需重启代理）。""" WhitelistDialog(self).exec_() def _save_settings(self): """保存更改：统一校验并应用全部设置项（后续新增设置项在此扩展）。""" port = _parse_port(self.ed_port.text()) if port is None: self.lb_hint.setText("端口应为 1-65535 的数字") self.lb_hint.setStyleSheet(f"color: {C_RED}; font-size: 11px;") return _set_local_port(port) _set_update_via_proxy(bool(self.cb_update_proxy.isChecked())) _set_skip_update_check(bool(self.cb_skip_update.isChecked())) self.ed_port.setText(str(port)) self.lb_hint.setText("已保存，下次开启代理生效") self.lb_hint.setStyleSheet(f"color: {C_GREEN}; font-size: 11px;") def _logout(self): """退出登录：删除加密凭据与本地节点配置，清空内存缓存。""" global _NODES for name in ("auth.json", "nodes.json"): try: (_USER_DIR / name).unlink(missing_ok=True) except Exception: pass _NODES = None log("已退出登录：凭据与节点配置已清除，下次启动或开启代理时需重新登录") self.accept() # ── 苹果风格开关 ───────────────────────────────────── class ToggleSwitch(QCheckBox): """苹果风格开关：左关右开，绿/灰胶囊 + 滑动圆钮（自绘，无图片资源）。""" def __init__(self, checked=False, parent=None): super().__init__(parent) self.setChecked(checked) self.setFixedSize(42, 22) self.setCursor(Qt.PointingHandCursor) def paintEvent(self, e): p = QPainter(self) p.setRenderHint(QPainter.Antialiasing) p.setPen(Qt.NoPen) # 胶囊背景：开=主题绿，关=浅灰 p.setBrush(QColor(C_GREEN) if self.isChecked() else QColor("#c8ccd4")) p.drawRoundedRect(QRectF(0, 0, 42, 22), 11, 11) # 白色圆钮：开在右，关在左 knob_x = 20.0 if self.isChecked() else 2.0 p.setBrush(QColor("#ffffff")) p.drawEllipse(QRectF(knob_x, 2.0, 18.0, 18.0)) p.end() def mousePressEvent(self, e): # 显式处理点击切换：自绘控件在部分样式引擎下默认点击不触发 if e.button() == Qt.LeftButton: self.setChecked(not self.isChecked()) self.repaint() e.accept() return super().mousePressEvent(e) # ── 状态指示灯（小圆点） ─────────────────────────────── class StatusDot(QLabel): def __init__(self): super().__init__() self.setFixedSize(14, 14) self.set_off() def set_on(self): self.setStyleSheet(f""" QLabel {{ background-color: {C_GREEN}; border-radius: 7px; border: 1px solid rgba(81, 207, 102, 50); }} """) def set_off(self): self.setStyleSheet(f""" QLabel {{ background-color: {C_RED}; border-radius: 7px; border: 1px solid rgba(255, 107, 107, 50); }} """) def set_disabled(self): self.setStyleSheet(f""" QLabel {{ background-color: {C_BORDER}; border-radius: 7px; border: 1px solid rgba(136, 153, 170, 50); }} """) # ── 主窗口 ──────────────────────────────────────────── class MainWindow(QWidget): log_signal = pyqtSignal(str) _op_done = pyqtSignal(str, bool) _update_done = pyqtSignal() _update_download_done = pyqtSignal(bool, str) _status_ready = pyqtSignal(object) def __init__(self): super().__init__() self._proxy_set_by_us = False # 系统代理是否由本程序开启（退出时只清理自己设置的） self._status_busy = False # 防止状态轮询线程堆叠 # 把全局 log() 函数接到 GUI 日志区 global _log_func _log_func = self.log_signal.emit self.log_signal.connect(self._append_log) self._op_done.connect(self._after_op) self._update_done.connect(self._on_update_done) self._update_download_done.connect(self._on_update_download_done) self._status_ready.connect(self._apply_status) _user_init_local() # 首启初始化用户数据目录（节点模板），须在 init_ui 读节点之前 self.init_ui() # 每 2 秒刷新一次状态指示灯 self.timer = QTimer() self.timer.timeout.connect(self.refresh_status) self.timer.start(2000) self.refresh_status() # 启动时后台检查更新，完成前禁用开启按钮 threading.Thread(target=self._check_update, daemon=True).start() # 后台检查 PAC 数据更新（首次启动自动拉取到用户数据目录） threading.Thread(target=_user_init_remote, daemon=True).start() # 后台用已保存凭据同步最新节点配置（服务端改动下次启动生效） threading.Thread(target=_nodes_sync_remote, daemon=True).start() # 节点未配置时延迟弹出登录框（等窗口显示后） QTimer.singleShot(600, self._maybe_show_login) def init_ui(self): """初始化主界面布局与样式。""" self.setWindowTitle("R.R.P.P.") self.setWindowFlags(self.windowFlags() & ~Qt.WindowContextHelpButtonHint) self.setWindowIcon(load_icon()) self.setFixedSize(_BASE_WINDOW_WIDTH, _BASE_WINDOW_HEIGHT) self.setStyleSheet(f""" QWidget {{ color: {C_TEXT}; font-family: "Microsoft YaHei", "Segoe UI", sans-serif; font-size: 13px; }} """) layout = QVBoxLayout(self) layout.setSpacing(14) layout.setContentsMargins(24, 20, 24, 20) # 标题 title = QLabel("Royirin Rosmontis Private Proxy") title.setFont(QFont("Microsoft YaHei", 16, QFont.Bold)) title.setStyleSheet(f"color: {C_TITLE};") title.setAlignment(Qt.AlignCenter) layout.addWidget(title) # 节点副标题（域名可点击切换：hover 下划线+手型提示，点击弹节点列表） node_row = QHBoxLayout() node_row.addStretch() prefix = QLabel("节点: ") prefix.setStyleSheet(f"color: {C_TITLE}; font-size: 12px;") node_row.addWidget(prefix) self.subtitle = NodeLabel(f"{_endpoint_domain()} ({_endpoint_country()})") self.subtitle.clicked.connect(self._show_node_menu) node_row.addWidget(self.subtitle) node_row.addStretch() layout.addLayout(node_row) # 连接方式选择区 pac_layout = QHBoxLayout() pac_layout.setSpacing(8) pac_layout.addStretch() self.cb_bypass_lan = QCheckBox("不代理内网 IP") self.cb_bypass_lan.setChecked(True) # 默认勾选：内网 IP 直连，不经过本地转发器 self.cb_bypass_lan.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;") pac_layout.addWidget(self.cb_bypass_lan) self.cb_pac = QCheckBox("PAC 规则模式") self.cb_pac.setChecked(True) # 默认开启智能分流（域名表/中国IP判定已足够完善） self.cb_pac.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;") self.cb_pac.setToolTip("只代理海外服务，使用域名和解析 IP 匹配。") pac_layout.addWidget(self.cb_pac) pac_layout.addStretch() layout.addLayout(pac_layout) # 分隔线 line = QFrame() line.setFrameShape(QFrame.HLine) line.setStyleSheet("background-color: rgba(136, 153, 170, 80); max-height: 1px;") layout.addWidget(line) # 系统代理状态卡片 px_frame = QFrame() px_frame.setStyleSheet(f""" QFrame {{ background-color: {C_CARD}; border: 1px solid {C_BORDER}; border-radius: 8px; }} """) px_layout = QHBoxLayout(px_frame) px_layout.setContentsMargins(16, 12, 16, 12) px_label = QLabel("IE 代理") px_label.setStyleSheet(f"color: {C_TEXT}; font-weight: bold; background: transparent;") self.px_dot = StatusDot() self.px_status = QLabel("检测中...") self.px_status.setStyleSheet(f"color: {C_SUBTEXT}; background: transparent;") px_layout.addWidget(px_label) px_layout.addStretch() px_layout.addWidget(self.px_status) px_layout.addWidget(self.px_dot) layout.addWidget(px_frame) traffic_cards = QHBoxLayout() traffic_cards.setSpacing(8) self.traffic_up = QLabel("↑0 B/s") self.traffic_down = QLabel("↓0 B/s") self.traffic_total = QLabel("0 B") for card in (self.traffic_up, self.traffic_down, self.traffic_total): card.setStyleSheet(f""" QLabel {{ background-color: {C_CARD}; border: 1px solid {C_BORDER}; border-radius: 8px; color: {C_SUBTEXT}; font-size: 12px; padding: 10px 4px; }} """) card.setAlignment(Qt.AlignCenter) traffic_cards.addWidget(card, 1) layout.addLayout(traffic_cards) layout.addStretch() # 开启 / 关闭 代理按钮 self.btn_toggle = QPushButton("开启代理") self.btn_toggle.setMinimumHeight(40) self._set_toggle_button_style(is_start=True) self.btn_toggle.clicked.connect(self.on_toggle) self.btn_toggle.setEnabled(False) # 更新检查完成前禁用，避免旧版本被使用 layout.addWidget(self.btn_toggle) # 日志输出区 self.log_text = QTextEdit() self.log_text.setReadOnly(True) self.log_text.setMaximumHeight(140) self.log_text.setStyleSheet(f""" QTextEdit {{ background-color: {C_CARD}; border: 1px solid {C_BORDER}; border-radius: 6px; color: {C_SUBTEXT}; font-size: 11px; padding: 4px 6px; }} QScrollBar:vertical {{ width: 6px; background: transparent; }} QScrollBar::handle:vertical {{ background: {C_BORDER}; border-radius: 3px; min-height: 20px; }} """) layout.addWidget(self.log_text) hints = QWidget() hints_layout = QHBoxLayout(hints) hints_layout.setContentsMargins(0, 0, 0, 0) hints_layout.setSpacing(8) # 左侧两行小字（左对齐），右侧完全留给设置按钮 left_col = QVBoxLayout() left_col.setContentsMargins(0, 0, 0, 0) left_col.setSpacing(1) warn_label = QLabel("请勿二次分发此软件！") warn_label.setStyleSheet(f"color: {C_RED}; font-size: 11px;") left_col.addWidget(warn_label) privacy_label = QLabel("使用此软件将会收集并上报您的 IP 地址及部分设备信息") privacy_label.setStyleSheet("color: #c8ccd4; font-size: 9px;") left_col.addWidget(privacy_label) hints_layout.addLayout(left_col, 1) self.btn_settings = QPushButton("设置") self.btn_settings.setCursor(Qt.PointingHandCursor) self.btn_settings.setStyleSheet(f""" QPushButton {{ background-color: {C_CARD}; color: {C_SUBTEXT}; border: 1px solid {C_BORDER}; border-radius: 6px; padding: 3px 10px; font-size: 11px; }} """) self.btn_settings.clicked.connect(self._show_settings_dialog) hints_layout.addWidget(self.btn_settings, 0, Qt.AlignVCenter) layout.addWidget(hints) def closeEvent(self, event): """关闭窗口时自动清理代理和隧道。""" try: self.quit_app() finally: event.accept() # 即使清理异常窗口也必须关闭 # ── 开关操作 ── def _set_toggle_button_style(self, is_start=True): if is_start: self.btn_toggle.setText("开启代理") self.btn_toggle.setStyleSheet(f""" QPushButton {{ background-color: {C_GREEN}; color: #fff; border: none; border-radius: 8px; padding: 10px 24px; font-size: 14px; font-weight: bold; }} QPushButton:hover {{ background-color: #5ed978; }} QPushButton:pressed {{ background-color: #45b85a; }} QPushButton:disabled {{ background-color: #2a3a2e; color: #666; }} """) else: self.btn_toggle.setText("关闭代理") self.btn_toggle.setStyleSheet(f""" QPushButton {{ background-color: {C_RED}; color: #fff; border: none; border-radius: 8px; padding: 10px 24px; font-size: 14px; font-weight: bold; }} QPushButton:hover {{ background-color: #ff8787; }} QPushButton:pressed {{ background-color: #e05555; }} QPushButton:disabled {{ background-color: #3a1e1e; color: #666; }} """) def on_toggle(self): """根据按钮当前文字判断是开启还是关闭。""" if self.btn_toggle.text() == "开启代理": self.on_start() else: self.on_stop() def on_start(self): # 节点未配置：先弹出登录框，成功后继续；取消/失败则中止开启 if _endpoint_domain() == "???": self._show_login_dialog() if _endpoint_domain() == "???": log_red("未登录，无法开启代理") return self.btn_toggle.setEnabled(False) self.cb_pac.setEnabled(False) self.cb_bypass_lan.setEnabled(False) # 在 GUI 线程读取控件状态传参，避免子线程直接访问 Qt 控件（非线程安全） use_pac = self.cb_pac.isChecked() use_bypass_lan = self.cb_bypass_lan.isChecked() threading.Thread(target=self._start_thread, args=(use_pac, use_bypass_lan), daemon=True).start() def _start_thread(self, use_pac, use_bypass_lan): try: self._start_proxy(use_pac, use_bypass_lan) self._op_done.emit("start", True) except Exception as e: log(f"[错误] 开启失败: {e}") self._op_done.emit("start", False) def _start_proxy(self, use_pac, use_bypass_lan): """启动本地转发器并设置系统代理（PAC/固定两种模式），可在后台线程调用。""" if _endpoint_domain() == "???": # 未配置节点时禁止开启：否则系统代理会指向连不通任何目标的转发器，导致断网 raise RuntimeError("未配置节点：请先通过 nodes_url 在线更新或 nodes_tool 导入节点配置") _traffic_reset() log("TLS 直连模式") ep = _endpoint_domain() log(f" 目标域名: {ep}") # PAC 模式：全部流量进转发器，按域名表/解析 IP 智能分流 if use_pac: _load_cn_ranges() _load_force_proxy_suffixes() _tls_start(ep, _endpoint_port(), use_tls=True, bypass_lan=use_bypass_lan, smart=use_pac) if use_pac: # 本地生成全量转发 PAC（不再从服务器拉取），全部流量交给转发器做智能分流 pac_content = _build_pac(f"{PROXY_HOST}:{PROXY_PORT}") if use_bypass_lan: pac_content = _apply_lan_bypass(pac_content) # 在内存中启动本地 PAC HTTP 服务，不往 EXE 目录写任何文件 pac_port = _start_pac_server(pac_content) pac_url = f"http://127.0.0.1:{pac_port}/proxy.pac" log(f"使用 PAC: {pac_url}") proxy_on(pac_url=pac_url) else: proxy_on(bypass_lan=use_bypass_lan) if use_bypass_lan: log("不代理内网 IP：内网地址将直连") self._proxy_set_by_us = True # 标记系统代理由本程序设置，退出时据此清理 log("验证代理连通性...") import urllib.request # 显式 ProxyHandler 强制经本地转发器；默认 opener 会套用系统/环境代理 # 设置（PAC 模式下注册表 ProxyEnable=0，urllib 认为无代理而绕开），造成假"可达" opener = urllib.request.build_opener( urllib.request.ProxyHandler({"http": f"{PROXY_HOST}:{PROXY_PORT}", "https": f"{PROXY_HOST}:{PROXY_PORT}"})) # baidu 验证国内直连路径，google 验证境外隧道路径 test_urls = [("baidu.com", "http://www.baidu.com"), ("google.com", "http://www.google.com")] for name, url in test_urls: try: opener.open(url, timeout=10) log(f" {name} 可达") except Exception as e: log(f" {name} 不可达: {e}") log("开启成功") def on_stop(self): """禁用按钮并在后台线程执行关闭流程。""" self.btn_toggle.setEnabled(False) threading.Thread(target=self._stop_thread, daemon=True).start() def _stop_thread(self): """后台关闭代理：清理系统代理、停止 TLS/PAC 服务。""" try: log("正在关闭代理...") proxy_off() self._proxy_set_by_us = False log("已关闭") self._op_done.emit("stop", True) except Exception as e: log(f"[错误] 关闭失败: {e}") self._op_done.emit("stop", False) # ── 设置 ── def _show_settings_dialog(self): """打开设置弹窗（代理运行中锁定端口修改）；退出登录后立即触发登录流程。""" SettingsDialog(self, port_locked=self._proxy_set_by_us).exec_() self._refresh_subtitle() if _endpoint_domain() == "???": self._maybe_show_login() # 退出登录后重新走登录（隐藏主窗口/取消退出） # ── 节点登录 ── def _maybe_show_login(self): """启动时节点未配置：隐藏主窗口只显示登录框；登录成功恢复，取消则退出。""" if _endpoint_domain() == "???": self.hide() self._show_login_dialog() if _endpoint_domain() == "???": self.quit_app() # 未登录且无可用节点，直接退出 return self.show() def _show_login_dialog(self): """弹出登录框；成功后重载节点并刷新标题。""" dlg = LoginDialog(self) if dlg.exec_() == QDialog.Accepted: _reload_nodes() self._refresh_subtitle() # ── 节点切换 ── def _show_node_menu(self): """弹出节点切换菜单（每次打开重新读取 nodes.json，外部改动即时生效）。""" if self._proxy_set_by_us: log("代理运行中，请先关闭代理再切换节点") return nodes = _reload_nodes() if not nodes: QMessageBox.information(self, "提示", "nodes.json 中没有可用节点。\n" "可在右下角「设置」中退出登录后重新登录获取节点配置。") return menu = QMenu(self) menu.setStyleSheet(f""" QMenu {{ background-color: {C_CARD}; border: 1px solid {C_BORDER}; border-radius: 8px; padding: 4px; }} QMenu::item {{ padding: 6px 18px; border-radius: 6px; color: {C_TEXT}; font-size: 12px; }} QMenu::item:selected {{ background-color: #e3e8ee; }} """) cur = _selected_index() for i, n in enumerate(nodes): domain = n.get("domain") or "???" country = n.get("country") or "" remark = n.get("remark") or "" text = f"{domain} ({country})" if remark and remark != domain: text += f"　{remark}" if i == cur: text = "✓ " + text menu.addAction(text).triggered.connect( lambda _=False, idx=i: self._on_node_selected(idx)) menu.exec_(self.subtitle.mapToGlobal(self.subtitle.rect().bottomLeft())) def _on_node_selected(self, idx): """切换节点：更新标题。""" if not _switch_node(idx): return node = _current_node() self._refresh_subtitle() log(f"已切换节点: {node.get('domain')} ({node.get('country')})") def _refresh_subtitle(self): """刷新节点副标题文字。""" self.subtitle.setText(f"{_endpoint_domain()} ({_endpoint_country()})") def _update_node_label(self): """代理运行中时提示禁止切换节点。""" if self._proxy_set_by_us: self.subtitle.setToolTip("代理运行中，请先关闭代理再切换节点") else: self.subtitle.setToolTip("点击切换节点") def _after_op(self, op, ok): """开启/关闭操作完成后刷新界面状态并恢复按钮/控件可用性。""" self.refresh_status() self.btn_toggle.setEnabled(True) # 恢复模式/PAC 控件的时机：关闭完成后、或开启失败时。 # 不能靠按钮文字判断——refresh_status 是异步的，此刻按钮样式还没更新 is_off = (op == "stop") or (op == "start" and not ok) self.cb_pac.setEnabled(is_off) self.cb_bypass_lan.setEnabled(is_off) self._update_node_label() # ── 日志 ── _early_logs = [] # 窗口初始化期间（log_text 创建前）到达的日志缓存 def _append_log(self, msg: str): """追加日志并自动滚动，最多保留 200 行；log_text 未创建时先缓存。""" if not hasattr(self, "log_text"): self._early_logs.append(msg) return for m in self._early_logs + [msg]: self._append_log_line(m) self._early_logs.clear() def _append_log_line(self, msg: str): self.log_text.append(msg) scrollbar = self.log_text.verticalScrollBar() scrollbar.setValue(scrollbar.maximum()) if self.log_text.document().blockCount() > 200: cursor = self.log_text.textCursor() cursor.movePosition(cursor.Start) cursor.select(cursor.BlockUnderCursor) cursor.removeSelectedText() cursor.deleteChar() # ── 状态刷新 ── def refresh_status(self): """触发一次后台状态轮询（不在 GUI 线程起子进程，避免周期性卡顿）。""" if self._status_busy: return self._status_busy = True # GUI 线程读取模式选择，传给后台线程 threading.Thread(target=self._poll_status, daemon=True).start() def _poll_status(self): try: px = is_proxy_on() except Exception: px = False self._status_ready.emit(px) def _apply_status(self, px_on_): """GUI 线程：根据轮询结果刷新状态指示灯、IE代理、流量卡片。""" self._status_busy = False # IE 代理状态 if px_on_: self.px_dot.set_on() self.px_status.setText("已开启") self.px_status.setStyleSheet(f"color: {C_GREEN}; background: transparent;") self._set_toggle_button_style(is_start=False) else: self.px_dot.set_off() self.px_status.setText("已关闭") self.px_status.setStyleSheet(f"color: {C_RED}; background: transparent;") self._set_toggle_button_style(is_start=True) # 流量卡片（始终更新，未连接时显示 0） px_on = is_proxy_on() if px_on: up_rate, down_rate, up_total, down_total = _traffic_rate() else: up_rate, down_rate, up_total, down_total = 0, 0, 0, 0 self.traffic_up.setText(f"↑{_fmt_bytes(up_rate)}/s") self.traffic_down.setText(f"↓{_fmt_bytes(down_rate)}/s") self.traffic_total.setText(_fmt_bytes(up_total + down_total)) def _check_update(self, manual=False): """后台检查服务器是否有新版本；manual=True 时无视「跳过检查更新」设置。""" # 先初始化结果，任何路径（含跳过）完成后 _on_update_done 都能安全读取 self._latest_version = "" self._download_url = "" try: if _skip_update_check() and not manual: return # 设置跳过（仅自动检查）：不检查也不打日志 import json, urllib.request # 基于 Gitea 仓库 releases API 检查更新；config.json 无 update_url 时用默认地址 url = str(_config_val("update_url", "") or "").strip() if not url: url = ("https://git.bu4.ren/api/v1/repos/RoyirinRosmontis/" "Private-Proxy/releases/latest") log("正在检查更新...") if _update_via_proxy(): # 走节点隧道（CONNECT，设置「使用代理更新」开启时）；显式代理避免残留干扰 started_here = _ensure_forwarder() try: opener = urllib.request.build_opener(urllib.request.ProxyHandler( {"http": f"{PROXY_HOST}:{PROXY_PORT}", "https": f"{PROXY_HOST}:{PROXY_PORT}"})) req = urllib.request.Request(url) data = opener.open(req, timeout=20).read().decode("utf-8-sig") finally: _release_forwarder(started_here) else: # 默认直连仓库；显式空代理避免系统/环境残留代理干扰 opener = _pac_no_proxy_opener() req = urllib.request.Request(url) data = opener.open(req, timeout=20).read().decode("utf-8-sig") info = json.loads(data) # 基于 Gitea 仓库 releases API：版本取 tag_name，下载取 exe 附件链接 self._latest_version = str(info.get("tag_name") or "") assets = info.get("assets") or [] self._download_url = (assets[0].get("browser_download_url") or "" if assets else "") log(f"当前版本: {VERSION}，最新版本: {self._latest_version or '未知'}") except Exception as e: log(f"更新检查失败: {e}") self._latest_version = "" self._download_url = "" finally: self._update_done.emit() def _on_update_done(self): """更新检查完成后恢复按钮；发现新版本时弹窗询问是否下载。""" self.btn_toggle.setEnabled(True) if self._latest_version and _version_newer(self._latest_version, VERSION): reply = QMessageBox.question( self, "发现新版本", f"当前版本: {VERSION}\n最新版本: {self._latest_version}\n\n是否下载更新？", QMessageBox.Yes | QMessageBox.No, QMessageBox.Yes, ) if reply == QMessageBox.Yes: threading.Thread(target=self._download_update, daemon=True).start() def _download_update(self): """后台下载新版本 EXE 到 EXE 所在目录。""" try: import urllib.request, urllib.parse, base64 url = self._download_url if not url: self._update_download_done.emit(False, "下载链接为空") return # 对 URL 进行安全编码，兼容空格或已编码空格 url = urllib.parse.quote(url, safe=':/?#[]@!$&\'()*+,;=%') save_path = PERSIST_DIR / f"Royirin Rosmontis Private Proxy {self._latest_version}.exe" log(f"正在下载新版本: {url}") log(f"保存到: {save_path}") # 同更新检查：按「使用代理更新」设置选择隧道或直连 if _update_via_proxy(): started_here = _ensure_forwarder() try: opener = urllib.request.build_opener(urllib.request.ProxyHandler( {"http": f"{PROXY_HOST}:{PROXY_PORT}", "https": f"{PROXY_HOST}:{PROXY_PORT}"})) req = urllib.request.Request(url) with opener.open(req, timeout=120) as resp: data = resp.read() finally: _release_forwarder(started_here) else: opener = _pac_no_proxy_opener() req = urllib.request.Request(url) with opener.open(req, timeout=120) as resp: data = resp.read() if len(data) < 1024: self._update_download_done.emit(False, "下载文件过小，可能不是有效的 EXE") return save_path.write_bytes(data) log(f"新版本已下载: {save_path}") self._update_download_done.emit(True, str(save_path)) except Exception as e: log(f"下载更新失败: {e}") self._update_download_done.emit(False, str(e)) def _on_update_download_done(self, success, message): """下载完成后提示用户手动替换 EXE。""" if success: QMessageBox.information( self, "下载完成", f"新版本已下载到:\n{message}\n\n请关闭本程序后，用该文件替换当前运行的程序，然后重新启动。" ) else: QMessageBox.warning(self, "下载失败", f"无法下载新版本:\n{message}") def quit_app(self): """只清理自己设置的系统代理：用户或其他软件设置的代理不受影响。 本程序异常退出残留的代理设置由下次启动时 _cleanup_stale_proxy() 处理。 """ if self._proxy_set_by_us: try: proxy_off() except Exception: pass # 清理失败也不能阻止退出，残留交给下次启动清理 QApplication.quit() # ── 程序入口 ────────────────────────────────────────── def _set_per_monitor_dpi_aware(): """在创建 QApplication 前尽量设置为每显示器 DPI 感知， 使程序在 2K/4K 高分屏以及多显示器不同缩放比例之间移动时， 由系统通知 WM_DPICHANGED 而不是被位图拉伸。 """ try: # Windows 10 1607+：Per Monitor V2（最优） ctypes.windll.user32.SetThreadDpiAwarenessContext(ctypes.c_ssize_t(-4)) return except Exception: pass try: # Windows 8.1+：Per Monitor ctypes.windll.shcore.SetProcessDpiAwareness(2) return except Exception: pass try: # 最低兜底：System DPI Aware ctypes.windll.user32.SetProcessDPIAware() except Exception: pass _SINGLE_INSTANCE_MUTEX = None def _acquire_single_instance(): """通过命名互斥体保证单实例运行；已有实例时返回 False。""" global _SINGLE_INSTANCE_MUTEX try: # 必须用 use_last_error 捕获真实错误码：直接调 GetLastError 可能被 # ctypes 内部调用重置（曾因此漏检已运行实例，导致双开互抢端口） kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) kernel32.CreateMutexW.restype = ctypes.c_void_p _SINGLE_INSTANCE_MUTEX = kernel32.CreateMutexW(None, True, r"Local\RRPP_SingleInstance") return ctypes.get_last_error() != 183 # ERROR_ALREADY_EXISTS except Exception: return True # 互斥体创建失败时放行，避免把用户锁在门外 def main(): """主入口：检查管理员权限、依赖文件，然后启动 GUI。""" _set_per_monitor_dpi_aware() # 单实例：第二个实例直接退出（且不能执行下面的残留代理清理， # 否则会把第一个实例正在使用的系统代理清掉） if not _acquire_single_instance(): app = QApplication(sys.argv) QMessageBox.information(None, "提示", "程序已在运行中。") sys.exit(0) # 上次异常退出可能残留指向本地转发端口的系统代理（转发器已随进程消失）， # 不清理会导致全网不通，启动时先清掉 _cleanup_stale_proxy() # 任务栏图标：显式设置 AppUserModelID， # 避免 Windows 依赖 exe 图标缓存推断（冷缓存下任务栏会显示空白/默认图标） try: ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID( "RRPP.RoyirinRosmontisPrivateProxy") except Exception: pass # Qt5 默认把缩放因子四舍五入到整数倍：2K 屏常见的 150% 会被取整为 2x， # 界面直接翻倍过大。改为 PassThrough，按实际比例（1.5x）缩放。 os.environ.setdefault("QT_SCALE_FACTOR_ROUNDING_POLICY", "PassThrough") try: QApplication.setHighDpiScaleFactorRoundingPolicy( Qt.HighDpiScaleFactorRoundingPolicy.PassThrough) except Exception: pass QApplication.setAttribute(Qt.AA_EnableHighDpiScaling, True) QApplication.setAttribute(Qt.AA_UseHighDpiPixmaps, True) app = QApplication(sys.argv) icon = load_icon() app.setWindowIcon(icon) app.setQuitOnLastWindowClosed(True) window = MainWindow() window.show() # 开机自启等场景下任务栏/Explorer 可能尚未就绪， # 延迟重设一次图标兜底，避免首次启动任务栏图标丢失 QTimer.singleShot(2000, lambda: window.setWindowIcon(icon)) # 退出必须彻底：此前点 X 后出现过进程残留（解释器收尾被残留线程/信号 # 竞态挂住），os._exit 直接终止进程——所有清理已在 quit_app 里完成 ret = app.exec_() os._exit(ret) if __name__ == "__main__": main()