fix: 修正发布版本时间戳

.gitignore

@@ -7,7 +7,3 @@ lite/.claude/
 lite/backups/
 lite/build/
 lite/dist/
-
-# Runtime PAC downloads generated by the UI smoke test.
-lite/test/pac_direct_domains.json
-lite/test/pac_domains.json

lite/MODULE_SPLIT_PLAN.md

@@ -1,148 +0,0 @@
-# RRPP Lite 模块拆分计划
-
-## 目标与边界
-
-将当前 `main.py`（约 3,480 行）渐进拆成职责清晰、可独立测试的模块，同时保持以下行为不变：
-
-- PAC、全局代理、智能分流和本地 TLS 转发的现有接口与优先级；
-- 节点订阅、DPAPI 加密存储、登录及本地运行数据位置；
-- 源码运行、PyInstaller 单文件版和 Nuitka 版的启动方式；
-- 现有 `lite/test/test_pac_routing.py` 的全部回归用例。
-
-本轮拆分期间**不提交、不推送到主仓库、不打包发布**。每个阶段均只在本地工作区验证；出现回归时优先回退到拆分前备份。
-
-## 开始前的保护措施
-
-1. 记录当前 Git 状态与当前 HEAD，不触碰既有未跟踪的构建产物、备份和用户文件。
-2. 在 `lite/backups/` 创建带时间戳的源码备份，例如：
-   `proxy_lite.py.bak_YYYYMMDD_HHMMSS_before_module_split`。
-3. 首次拆分前运行一次完整基线：
-
-   ```powershell
-   wsl.exe -e bash -lc "cd '/mnt/d/Royirin Rosmontis Private Proxy' && python3 -m py_compile lite/main.py lite/test/test_pac_routing.py && python3 lite/test/test_pac_routing.py"
-   ```
-
-4. 每完成一个阶段，先做编译检查，再跑完整 PAC 回归；不通过则在进入下一阶段前修复或回退。
-
-## 当前实施进度（2026-09-16）
-
-- [x] 已创建拆分前备份：`lite/backups/proxy_lite.py.bak_20260916_220240_before_module_split`。
-- [x] 已完成一次拆分前基线与多次拆分后全量 PAC 回归。
-- [x] UI 主题、控件和三个对话框已迁入 `rrpp_ui/theme.py`、`rrpp_ui/widgets.py`、
-      `rrpp_ui/dialogs.py`；通过显式服务对象调用业务层，不反向导入入口文件。
-- [x] 已迁入核心包的无状态功能：CIDR/Trie 与 PAC 规则解析、PAC 拉取和原子写入、
-      `settings.json` 原子存储、DPAPI 节点/凭据存储、HTTP 转发报文解析与认证注入、
-      DNS/私网判断与 TTL 缓存、流量统计、节点认证请求协议。
-- [ ] 节点同步服务、带缓存的完整路由决策、转发器 socket 生命周期、系统代理注册表控制
-      和主窗口仍在入口文件，待后续阶段继续迁移。
-- [x] 已获得用户授权，在完成构建与验证后执行 `git commit`、`git push` 和 Release 发布。
-
-## 目标目录
-
-```text
-lite/
-├─ main.py                       # 轻量入口、组装依赖、main
-├─ rrpp_core/
-│  ├─ __init__.py
-│  ├─ runtime.py                 # 路径、常量、日志与共享运行时上下文
-│  ├─ config_store.py            # 普通配置、节点、DPAPI 凭据
-│  ├─ pac_data.py                # PAC 在线数据下载、校验、缓存失效
-│  ├─ routing.py                 # 域名/IP 分流、白名单、CIDR/Trie
-│  ├─ forwarder.py               # 本地 TLS 转发、HTTP CONNECT、认证注入
-│  └─ system_proxy.py            # PAC 服务、Windows 系统代理状态
-├─ rrpp_ui/
-│  ├─ __init__.py
-│  ├─ theme.py                   # UI 色彩与样式常量
-│  ├─ widgets.py                 # NodeLabel、开关、状态点等可复用控件
-│  └─ dialogs.py                 # 登录、白名单、设置对话框
-└─ test/
-   └─ test_pac_routing.py
-```
-
-目录是目标形态，而不是一次性迁移要求。入口已统一为 `main.py`；测试和构建脚本均以该文件为准。
-
-## 分阶段实施
-
-### 阶段 0：基线与备份
-
-- 创建源码备份并保存当前测试输出。
-- 核验当前核心测试：PAC 规则优先级、IPv4/IPv6、私网、CONNECT 解析、认证注入、启动连通性校验及其取消流程。
-- 产物：可随时恢复的单文件版本；无业务代码变更。
-
-### 阶段 1：抽取无状态 UI 组件
-
-- 将 `NodeLabel`、`ToggleSwitch`、`StatusDot` 及图标辅助逻辑移到 `rrpp_ui/widgets.py`。
-- 主窗口继续从原入口导入/调用，保持控件信号、样式和构造参数不变。
-- 风险低：不涉及代理线程、注册表或路由决策。
-- 验收：`py_compile`、PAC 全量测试、一次 Windows GUI 手工启动。
-
-### 阶段 2：建立共享运行时与配置存储
-
-- 抽取路径常量、配置读写、节点读写、DPAPI 加解密、认证保存和订阅节点同步至 `rrpp_core/runtime.py`、`config_store.py`。
-- 明确依赖方向：模块不得反向导入 UI；日志通过注入函数或统一运行时上下文提供。
-- 保持 `%LOCALAPPDATA%\\RRPP`、配置文件名、加密格式和节点选择恢复逻辑完全兼容。
-- 风险中等：涉及用户已有数据；只做机械迁移，不改变存储格式。
-- 验收：现有节点可读取；登录/同步至少做一次人工冒烟；全部自动测试通过。
-
-### 阶段 3：抽取 PAC 数据与路由引擎
-
-- 将 PAC 数据下载/校验/写入和缓存失效移至 `rrpp_core/pac_data.py`。
-- 将白名单、强制代理、明确直连域名、CIDR、IPv6、DNS 解析、Trie 匹配和直连决策移至 `rrpp_core/routing.py`。
-- 使用显式的 `RoutingContext` 或等价配置对象替代跨模块全局变量；规则优先级必须锁定为：
-
-  ```text
-  用户白名单 > 强制代理 > 明确直连域名 > LAN/私网 > GeoIP 智能分流
-  ```
-
-- 风险最高的核心阶段；禁止顺便修改规则语义或远程 PAC 格式。
-- 验收：扩充并运行分流回归测试，覆盖 Google/OpenAI/Anthropic、国内直连、IPv6、无 DNS 结果与缓存刷新。
-
-### 阶段 4：抽取转发器与系统代理控制
-
-- 将流量统计、本地 TLS 监听、CONNECT/绝对 URL 解析、认证注入与连接回退移至 `rrpp_core/forwarder.py`。
-- 将 PAC HTTP 服务、注册表备份/恢复、全局/PAC 开关和陈旧状态清理移至 `rrpp_core/system_proxy.py`。
-- 以明确的 start/stop 生命周期对象管理线程与端口，避免模块级状态互相覆盖。
-- 风险中高：涉及注册表和后台线程；本阶段不改变任何系统代理策略。
-- 验收：本地端口监听、PAC 开关、全局开关、取消启动、退出清理和重复启停的人工测试；完整回归测试。
-
-### 阶段 5：抽取对话框与收敛入口
-
-- 将 `LoginDialog`、`WhitelistDialog`、`SettingsDialog` 移至 `rrpp_ui/dialogs.py`，`MainWindow` 可保留在入口或迁到 `rrpp_ui/main_window.py`。
-- `main.py` 收敛为：初始化 DPI/单实例、创建运行时、装配 UI、调用 `main()`。
-- 保留必要的兼容导出，待构建脚本和测试完全迁移后再移除。
-- 验收：源码 GUI 手工走查、全量自动测试、PyInstaller/Nuitka 各做一次仅本地的构建与启动冒烟。
-
-### 阶段 6：清理与交付前评审
-
-- 删除已确认无引用的兼容层，补齐模块文档和最小单元测试。
-- 检查循环导入、隐式全局状态、未关闭线程、路径在冻结环境下的差异。
-- 对比拆分前后的 PAC 输出和系统代理状态变更。
-- 仅在用户确认后，另行决定提交、推送、打包和发布；它们不属于本次拆分的自动后续动作。
-
-## 依赖与架构约束
-
-```text
-main.py  →  rrpp_ui  →  rrpp_core
-                    │           │
-                    └───────────┘（仅通过显式上下文/接口协作）
-```
-
-- `rrpp_core` 不依赖 PyQt；这样 PAC 与转发逻辑可在无 GUI 环境测试。
-- UI 不直接读写注册表、节点文件或 socket；统一通过核心服务对象操作。
-- 禁止新的模块级可变全局状态，已有状态在迁移时集中到上下文或生命周期对象。
-- 配置、加密和 PAC 文件的文件名、字段、URL 与优先级，在本次重构内视为兼容性契约。
-
-## 验收清单
-
-- [x] 源码 `py_compile` 通过。
-- [x] `test_pac_routing.py` 全部通过。
-- [ ] PAC 启动/取消、全局模式、退出恢复系统代理均可用。
-- [ ] 既有节点和已加密凭据可读取，订阅同步不回退格式。
-- [ ] `chatgpt.com`、`auth.openai.com`、`claude.ai`、Google CDN 仍命中强制代理；国内显式直连规则仍生效。
-- [x] PyInstaller 与 Nuitka 均已完成本地构建；Release 附件将使用版本化文件名和 SHA-256 校验。
-- [x] 发布暂存仅包含本轮源码、测试、构建脚本、图标和模板；构建产物、账号数据和备份文件已排除。
-- [ ] 待提交、推送并上传 Release 附件。
-
-## 回滚策略
-
-任一阶段出现无法在当前阶段定位的回归时：停止继续迁移，将该阶段涉及文件恢复到阶段开始前的本地备份或 Git 基线；保留失败现场的测试输出以便定位。不得通过覆盖用户的运行配置、订阅或构建目录来“修复”回归。

lite/build_pyinstaller.bat

@@ -1,12 +0,0 @@
-@echo off
-cd /d "%~dp0"
-call ..\.venv\Scripts\python.exe -m PyInstaller RRPP_Lite.spec
-if errorlevel 1 (
-    echo.
-    echo Build failed.
-    pause
-    exit /b 1
-)
-echo.
-echo Build complete. Output: dist\Royirin Rosmontis Private Proxy Lite.exe
-pause

lite/claude.bat

@@ -1,12 +0,0 @@
-@echo off
-setlocal
-rem Convert current Windows dir to a WSL path (lowercase drive + forward slashes)
-set "WSLDIR=%CD:\=/%"
-set "WSLDIR=%WSLDIR::=%"
-set "DRIVE=%WSLDIR:~0,1%"
-for %%C in (a b c d e f g h i j k l m n o p q r s t u v w x y z) do if /i "%%C"=="%DRIVE%" set "WSLDIR=%%C%WSLDIR:~1%"
-set "WSLDIR=/mnt/%WSLDIR%"
-rem -i is essential: in a non-interactive shell PS1 is empty, so ~/.bashrc's
-rem interactive guard returns early and the model-mapping env vars are lost
-wsl -e bash -ic "cd '%WSLDIR%' && /root/.local/bin/claude --continue"
-pause

lite/config.json

@@ -1,6 +1,6 @@
 {
-  "version": "",
+  "version": "999909171702",
-  "update_url": "",
+  "pac_base_url": "https://cdn.rrrosmontis.icu/RRPP/PAC/",
-  "pac_base_url": "",
+  "nodes_url": "https://api.rrrosmontis.icu/v1/rrpp/subscription?method=rrpp_client",
-  "nodes_url": ""
+  "update_url": "https://git.rrrosmontis.icu/api/v1/repos/RoyirinRosmontis/Private-Proxy/releases/latest"
 }

lite/main.py

@@ -1,12 +1,12 @@
 # -*- coding: utf-8 -*-
 
-import sys, os, subprocess, time, ctypes, threading, functools, base64, bisect, re
+import sys, os, subprocess, time, ctypes, threading, functools, base64, bisect, re, json, uuid
 from pathlib import Path
 
 from PyQt5.QtWidgets import (
     QApplication, QWidget, QLabel, QPushButton,
     QVBoxLayout, QHBoxLayout, QFrame, QMessageBox, QTextEdit,
-    QCheckBox, QMenu, QDialog,
+    QCheckBox, QMenu, QDialog, QProgressDialog,
 )
 from PyQt5.QtCore import Qt, QTimer, pyqtSignal
 from PyQt5.QtGui import QFont, QIcon, QPixmap, QImage, QColor
@@ -14,7 +14,7 @@ from PyQt5.QtGui import QFont, QIcon, QPixmap, QImage, QColor
 from rrpp_ui.theme import C_TITLE, C_GREEN, C_YELLOW, C_RED, C_BG, C_CARD, C_BORDER, C_TEXT, C_SUBTEXT
 from rrpp_ui.widgets import NodeLabel, StatusDot, ToggleSwitch
 from rrpp_ui.dialogs import (
-    LoginDialog, LoginServices, SettingsDialog, SettingsServices,
+    LoginDialog, LoginServices, SettingsDialog, SettingsServices, ThirdPartyDialog, ThirdPartyServices,
     WhitelistDialog, WhitelistServices,
 )
 from rrpp_core.routing_primitives import (
@@ -45,6 +45,10 @@ from rrpp_core.node_api import (
     b64url as _b64url,
     fetch_nodes_auth as _fetch_nodes_auth,
 )
+from rrpp_core.third_party_subscriptions import (
+    fetch_nodes as _fetch_third_party_nodes,
+    mask_url as _mask_subscription_url,
+)
 from rrpp_core.network_primitives import (
     is_private_host as _is_private_host,
     is_private_ipv4_int as _is_private_int,
@@ -61,10 +65,21 @@ _BASE_WINDOW_HEIGHT = 470  # 内容区 + 加大日志区 + 底部两行提示小
 # 已被打包为 EXE（PyInstaller 的 sys.frozen / Nuitka 的 __compiled__）：
 #   BASE_DIR   -> 内置数据目录：PyInstaller 用 _MEIPASS，Nuitka onefile 用
 #                 __file__ 所在解压临时目录（无 _MEIPASS 时回退）
-#   PERSIST_DIR -> EXE 所在目录，服务安装等需要持久化的操作都放这里
+#   PERSIST_DIR -> EXE 所在目录，服务安装等需要持久化的操作都放这里。
+# Nuitka onefile 的 sys.executable 指向 TEMP\onefile_* 解包子进程，不能用它
+# 定位用户启动的原始 EXE；优先使用 __compiled__.containing_dir，旧版 Nuitka
+# 再回退到 sys.argv[0]（官方保证它指向原始 onefile 路径）。
+def _frozen_executable_dir():
+    if "__compiled__" in globals():
+        compiled_dir = getattr(globals()["__compiled__"], "containing_dir", None)
+        if compiled_dir:
+            return Path(compiled_dir).resolve()
+        return Path(sys.argv[0]).resolve().parent
+    return Path(sys.executable).resolve().parent
+
 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
+    PERSIST_DIR = _frozen_executable_dir()
 else:
     BASE_DIR = Path(__file__).resolve().parent
     PERSIST_DIR = BASE_DIR
@@ -99,39 +114,203 @@ def _load_config():
 def _config_val(key, default=""):
     return _load_config().get(key, default)
 
-# ── 节点配置（nodes.json）────────────────────────────
+# ── 节点配置（官方 nodes.json + 第三方订阅）────────────────
-# 节点列表，当前固定取第一个（后续可扩展选择）
+# 官方节点和第三方节点分别保存；加载时才合并，避免登录更新覆盖第三方缓存。
 _NODES = None
+_THIRD_PARTY_FILE = "third_party_subscriptions.json"
+_THIRD_PARTY_LOCK = threading.RLock()
+
+
+def _third_party_path():
+    return _USER_DIR / _THIRD_PARTY_FILE
+
+
+def _load_third_party_subscriptions():
+    """读取本机 DPAPI 加密的第三方订阅与最近一次成功缓存。"""
+    with _THIRD_PARTY_LOCK:
+        try:
+            path = _third_party_path()
+            if not path.is_file():
+                return []
+            data = json.loads(_nodes_decrypt(path.read_bytes()))
+            records = data.get("subscriptions", []) if isinstance(data, dict) else []
+            return [record for record in records if isinstance(record, dict) and record.get("id")]
+        except Exception as error:
+            log_red(f"第三方订阅配置读取失败: {error}")
+            return []
+
+
+def _save_third_party_subscriptions(records):
+    """原子保存加密订阅配置；URL 与节点认证信息不进入明文 settings.json。"""
+    with _THIRD_PARTY_LOCK:
+        _USER_DIR.mkdir(parents=True, exist_ok=True)
+        text = json.dumps({"version": 1, "subscriptions": records}, ensure_ascii=False, indent=2) + "\n"
+        path = _third_party_path()
+        temporary = path.with_suffix(path.suffix + ".tmp")
+        temporary.write_bytes(_nodes_encrypt(text))
+        os.replace(temporary, path)
+
+
+def _subscription_nodes(records):
+    result = []
+    for record in records:
+        source_id = str(record.get("id") or "")
+        source_name = str(record.get("name") or "第三方订阅")
+        nodes = record.get("nodes") or []
+        if not isinstance(nodes, list):
+            continue
+        for node in nodes:
+            if not isinstance(node, dict) or not str(node.get("domain") or "").strip():
+                continue
+            item = dict(node)
+            item["_rrpp_source_id"] = source_id
+            item["_rrpp_source_name"] = source_name
+            item["_rrpp_source_kind"] = "third_party"
+            result.append(item)
+    return result
+
+
+def _tag_official_nodes(nodes):
+    result = []
+    for node in nodes:
+        if not isinstance(node, dict) or not str(node.get("domain") or "").strip():
+            continue
+        item = dict(node)
+        item["_rrpp_source_id"] = "official"
+        item["_rrpp_source_name"] = "官方订阅"
+        item["_rrpp_source_kind"] = "official"
+        result.append(item)
+    return result
+
+
+def _node_reference(node):
+    """选中节点的稳定标识，避免不同订阅的同域名互相覆盖。"""
+    method = (node.get("methods") or {}).get("http") or {}
+    port = method.get("port", node.get("port", ""))
+    return "|".join((
+        str(node.get("_rrpp_source_id") or "official"),
+        str(node.get("domain") or ""), str(port or ""),
+        str(node.get("country") or ""), str(node.get("remark") or ""),
+    ))
+
 
 def _load_nodes():
-    """加载 nodes.json 节点列表（用户数据目录优先）；失败或缺失时返回空列表。"""
+    """加载并合并官方节点与每个第三方订阅的上次成功缓存。"""
     global _NODES
     if _NODES is not None:
         return _NODES
-    _NODES = []
+    official = []
     try:
-        import json
         path = _user_path("nodes.json")
-        if path is None:
+        if path is not None:
-            log_red("nodes.json 缺失，无法加载节点配置")
+            data = json.loads(_nodes_decrypt(Path(path).read_bytes()))
-            return _NODES
+            if not isinstance(data, list):
-        data = json.loads(_nodes_decrypt(Path(path).read_bytes()))
+                raise ValueError("nodes.json 顶层应为节点数组")
-        if not isinstance(data, list):
+            official = _tag_official_nodes(data)
-            raise ValueError("nodes.json 顶层应为节点数组")
+    except Exception as error:
-        _NODES = data
+        log_red(f"官方节点配置读取失败: {error}")
-        if not _SELECTED_RESTORED:
+    third_party = _subscription_nodes(_load_third_party_subscriptions())
-            _restore_selected_node(_NODES)  # 恢复上次使用的节点
+    _NODES = official + third_party
-        log(f"已加载节点配置: {len(_NODES)} 个")
+    if not _SELECTED_RESTORED:
-        if data and not str(data[0].get("domain") or "").strip():
+        _restore_selected_node(_NODES)
-            log_red("节点未填写域名：请配置 config.json 的 nodes_url 在线更新，"
+    if _NODES:
-                    "或使用 others/nodes_tool 导入节点配置")
+        log(f"已加载节点配置: 官方 {len(official)} 个，第三方 {len(third_party)} 个")
-    except Exception as e:
-        log_red(f"nodes.json 加载失败: {e}")
     return _NODES
 
-_SELECTED_NODE = 0  # 当前选中节点下标（默认第一条）
+
-_SELECTED_RESTORED = False  # 启动时是否已按设置恢复过上次选中的节点
+def _third_party_summaries():
-_NODES_SYNC_DONE = threading.Event()  # 订阅同步线程完成信号（启动校验等它）
+    """供 UI 展示，不向界面或日志暴露完整订阅 URL。"""
+    return [{
+        "id": str(record.get("id")),
+        "name": str(record.get("name") or "第三方订阅"),
+        "url": _mask_subscription_url(record.get("url") or ""),
+        "node_count": len(record.get("nodes") or []),
+        "updated_at": str(record.get("updated_at") or "未更新"),
+        "last_error": str(record.get("last_error") or ""),
+    } for record in _load_third_party_subscriptions()]
+
+
+def _subscription_error_text(error):
+    """保留格式错误提示，网络异常不回显可能含令牌的请求地址。"""
+    if isinstance(error, ValueError):
+        return str(error)
+    return "下载失败，请检查网络和订阅地址"
+
+
+def _add_third_party_subscription(name, url):
+    name, url = str(name or "").strip(), str(url or "").strip()
+    if not name:
+        raise ValueError("请填写订阅名称")
+    if not url.lower().startswith(("https://", "http://")):
+        raise ValueError("订阅链接必须以 http:// 或 https:// 开头")
+    with _THIRD_PARTY_LOCK:
+        records = _load_third_party_subscriptions()
+        if any(str(record.get("url") or "") == url for record in records):
+            raise ValueError("该订阅链接已添加")
+        try:
+            nodes = _fetch_third_party_nodes(url)
+        except Exception as error:
+            raise ValueError(_subscription_error_text(error)) from None
+        records.append({
+            "id": uuid.uuid4().hex,
+            "name": name,
+            "url": url,
+            "nodes": nodes,
+            "updated_at": time.strftime("%Y-%m-%d %H:%M:%S"),
+            "last_error": "",
+        })
+        _save_third_party_subscriptions(records)
+    _reload_nodes()
+    log(f"第三方订阅已添加: {name}（{len(nodes)} 个节点）")
+
+
+def _refresh_third_party_subscription(subscription_id):
+    with _THIRD_PARTY_LOCK:
+        records = _load_third_party_subscriptions()
+        record = next((item for item in records if str(item.get("id")) == str(subscription_id)), None)
+        if record is None:
+            raise ValueError("第三方订阅不存在")
+        try:
+            nodes = _fetch_third_party_nodes(str(record.get("url") or ""))
+            record["nodes"] = nodes
+            record["updated_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
+            record["last_error"] = ""
+            _save_third_party_subscriptions(records)
+            _reload_nodes()
+            log(f"第三方订阅已更新: {record.get('name')}（{len(nodes)} 个节点）")
+        except Exception as error:
+            record["last_error"] = _subscription_error_text(error)
+            _save_third_party_subscriptions(records)
+            raise ValueError(record["last_error"]) from None
+
+
+def _remove_third_party_subscription(subscription_id):
+    with _THIRD_PARTY_LOCK:
+        records = _load_third_party_subscriptions()
+        selected = [item for item in records if str(item.get("id")) == str(subscription_id)]
+        if not selected:
+            return
+        _save_third_party_subscriptions([item for item in records if item not in selected])
+    _reload_nodes()
+    log(f"第三方订阅已删除: {selected[0].get('name')}")
+
+
+def _sync_third_party_subscriptions():
+    """启动后更新第三方订阅；失败时保留上次成功的节点缓存。"""
+    try:
+        for record in _load_third_party_subscriptions():
+            try:
+                _refresh_third_party_subscription(record.get("id"))
+            except Exception as error:
+                log(f"第三方订阅更新失败，继续使用本地缓存: {record.get('name')} ({_subscription_error_text(error)})")
+    finally:
+        _THIRD_PARTY_SYNC_DONE.set()
+
+
+_SELECTED_NODE = 0
+_SELECTED_RESTORED = False
+_NODES_SYNC_DONE = threading.Event()  # 官方订阅同步完成信号（启动校验等它）
+_THIRD_PARTY_SYNC_DONE = threading.Event()
 
 def _current_node():
     """当前使用的节点：默认取列表第一个；无节点时返回 None。"""
@@ -156,24 +335,30 @@ def _switch_node(index):
         log("已在当前节点")
         return False
     _SELECTED_NODE = index
-    _save_current_node(nodes[index].get("domain") or "")  # 记录到设置，下次启动恢复
+    _save_current_node(nodes[index])
     return True
 
 def _restore_selected_node(nodes):
-    """首次加载时按设置里记录的 current_node（域名）恢复上次选中的节点。"""
+    """首次加载时按稳定节点标识恢复；兼容旧版仅按域名保存的设置。"""
     global _SELECTED_NODE, _SELECTED_RESTORED
     _SELECTED_RESTORED = True
+    saved_ref = str(_settings_load().get("current_node_ref") or "")
     saved = str(_settings_load().get("current_node") or "")
+    for i, n in enumerate(nodes):
+        if saved_ref and _node_reference(n) == saved_ref:
+            _SELECTED_NODE = i
+            return
     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):
+def _save_current_node(node):
-    """把当前选中节点域名写入设置文件（下次启动自动继续使用）。"""
+    """保存来源与节点标识；保留旧域名字段以兼容已有 settings.json。"""
     data = _settings_load()
-    data["current_node"] = str(domain or "")
+    data["current_node"] = str((node or {}).get("domain") or "")
+    data["current_node_ref"] = _node_reference(node or {})
     _settings_save(data)
 
 # ── IPv6-only 节点可达性（客户端标注用）─────────────────────
@@ -301,16 +486,21 @@ def _first_reachable_index(nodes):
     return 0
 
 def _rematch_saved_domain():
-    """按 settings 的 current_node（域名）在最新节点列表里重匹配下标。
+    """按保存的来源与节点标识重匹配，兼容旧版域名选择记录。"""
-
-    远程同步后服务端可能调整节点顺序：只按下标会静默连到别的节点。
-    返回 True 表示发生了切换。"""
     global _SELECTED_NODE
     nodes = _load_nodes()
     if not nodes:
         return False
+    saved_ref = str(_settings_load().get("current_node_ref") or "")
     saved = str(_settings_load().get("current_node") or "")
     cur = nodes[_SELECTED_NODE] if 0 <= _SELECTED_NODE < len(nodes) else None
+    if saved_ref and (not cur or _node_reference(cur) != saved_ref):
+        match = next((i for i, n in enumerate(nodes)
+                      if _node_reference(n) == saved_ref), None)
+        if match is not None:
+            _SELECTED_NODE = match
+            log(f"节点列表已更新，继续使用上次选择的节点 {nodes[match].get('domain')}")
+            return True
     if saved and (not cur or str(cur.get("domain") or "") != saved):
         match = next((i for i, n in enumerate(nodes)
                       if str(n.get("domain") or "") == saved), None)
@@ -338,7 +528,7 @@ def _verify_selected_node():
     old = str(cur.get("domain") or "")
     new = str(nodes[idx].get("domain") or "")
     _SELECTED_NODE = idx
-    _save_current_node(new)
+    _save_current_node(nodes[idx])
     if new and new != old:
         log_red(f"节点 {old} 在本机不可用，已自动切换到 {new}")
     return True
@@ -381,6 +571,14 @@ def _endpoint_port():
 def _endpoint_country():
     return _node_val("country", "")
 
+
+def _endpoint_label():
+    """主界面节点副标题；无节点时给出可操作的提示。"""
+    domain = _endpoint_domain()
+    if domain == "???":
+        return "无可用节点(请登录)"
+    return f"{domain} ({_endpoint_country()})"
+
 def _proxy_auth_basic():
     """正向代理 Basic 认证凭证：methods.http.{username,password}（新格式）
     或节点顶层同名字段（旧格式）；用于转发器向上游代理注入 Authorization。"""
@@ -1080,16 +1278,8 @@ def _nodes_sync_remote():
         _NODES_SYNC_DONE.set()  # 无论成败都要放行启动校验
 
 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("已创建节点配置文件（加密存储），请配置 config.json 的 "
-                "nodes_url 在线更新，或使用 others/nodes_tool 导入节点配置")
 
 def _user_init_remote(base_url=None):
     """检查并更新 PAC 数据（后台线程）：
@@ -1616,8 +1806,27 @@ def _whitelist_services():
     )
 
 
+def _third_party_services():
+    return ThirdPartyServices(
+        load_icon=load_icon,
+        load_subscriptions=_third_party_summaries,
+        add_subscription=_add_third_party_subscription,
+        refresh_subscription=_refresh_third_party_subscription,
+        remove_subscription=_remove_third_party_subscription,
+    )
+
+
+def _open_login_dialog(parent=None):
+    """从启动流程或设置页登录官方订阅；成功后立即重载合并节点列表。"""
+    dlg = LoginDialog(_login_services(), parent)
+    if dlg.exec_() == QDialog.Accepted:
+        _reload_nodes()
+        return True
+    return False
+
+
 def _logout_user_data():
-    """清除本地登录凭据与节点文件，并让内存节点缓存失效。"""
+    """清除官方登录与节点缓存；第三方订阅保持不变。"""
     global _NODES
     for name in ("auth.json", "nodes.json"):
         try:
@@ -1625,7 +1834,7 @@ def _logout_user_data():
         except Exception:
             pass
     _NODES = None
-    log("已退出登录：凭据与节点配置已清除，下次启动或开启代理时需重新登录")
+    log("已退出登录：官方凭据与节点配置已清除，第三方订阅保持不变")
 
 
 def _settings_services():
@@ -1640,6 +1849,9 @@ def _settings_services():
         set_update_via_proxy=_set_update_via_proxy,
         set_skip_update_check=_set_skip_update_check,
         open_whitelist=lambda parent: WhitelistDialog(_whitelist_services(), parent).exec_(),
+        open_third_party=lambda parent: ThirdPartyDialog(_third_party_services(), parent).exec_(),
+        has_login=lambda: _auth_load() is not None,
+        open_login=_open_login_dialog,
         logout=_logout_user_data,
     )
 
@@ -1652,6 +1864,7 @@ class MainWindow(QWidget):
     _op_done = pyqtSignal(str, bool)
     _update_done = pyqtSignal()
     _update_download_done = pyqtSignal(bool, str)
+    _update_download_progress = pyqtSignal(int, int, int)
 
     _status_ready = pyqtSignal(object)
     _node_switched = pyqtSignal()   # 启动校验时代替用户切换了节点（刷新标题栏）
@@ -1665,6 +1878,7 @@ class MainWindow(QWidget):
         self._start_in_progress = False
         self._proxy_state_lock = threading.RLock()
         self._status_busy = False      # 防止状态轮询线程堆叠
+        self._update_progress_dialog = None
         # 把全局 log() 函数接到 GUI 日志区
         global _log_func
         _log_func = self.log_signal.emit
@@ -1672,6 +1886,7 @@ class MainWindow(QWidget):
         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._update_download_progress.connect(self._on_update_download_progress)
         self._status_ready.connect(self._apply_status)
         self._node_switched.connect(self._on_node_switched)
 
@@ -1691,6 +1906,7 @@ class MainWindow(QWidget):
         threading.Thread(target=_user_init_remote, daemon=True).start()
         # 后台用已保存凭据同步最新节点配置（服务端改动下次启动生效）
         threading.Thread(target=_nodes_sync_remote, daemon=True).start()
+        threading.Thread(target=_sync_third_party_subscriptions, daemon=True).start()
         # 后台校验上次选中的节点在本机是否还可用（IPv6 专用节点本机无 IPv6 时回退）
         threading.Thread(target=self._verify_node_thread, daemon=True).start()
         # 节点未配置时延迟弹出登录框（等窗口显示后）
@@ -1703,6 +1919,7 @@ class MainWindow(QWidget):
         订阅同步可能较慢：等同步完成（最多 15 秒），同步后才校验，
         否则服务端换过节点顺序时校验的是旧列表。"""
         _NODES_SYNC_DONE.wait(15)
+        _THIRD_PARTY_SYNC_DONE.wait(15)
         try:
             if _verify_selected_node():
                 self._node_switched.emit()
@@ -1711,7 +1928,7 @@ class MainWindow(QWidget):
 
     def _on_node_switched(self):
         """节点被自动切换后刷新标题栏（在 UI 线程执行）。"""
-        self.subtitle.setText(f"{_endpoint_domain()} ({_endpoint_country()})")
+        self.subtitle.setText(_endpoint_label())
 
     def init_ui(self):
         """初始化主界面布局与样式。"""
@@ -1745,7 +1962,7 @@ class MainWindow(QWidget):
         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 = NodeLabel(_endpoint_label())
         self.subtitle.clicked.connect(self._show_node_menu)
         node_row.addWidget(self.subtitle)
         node_row.addStretch()
@@ -1940,11 +2157,13 @@ class MainWindow(QWidget):
     def on_start(self):
         if self._start_in_progress:
             return
-        # 节点未配置：先弹出登录框，成功后继续；取消/失败则中止开启
+        # 未配置节点时允许先登录；跳过后仍留在主界面，可到设置页添加第三方订阅。
         if _endpoint_domain() == "???":
             self._show_login_dialog()
             if _endpoint_domain() == "???":
-                log_red("未登录，无法开启代理")
+                message = "未配置可用节点。请在设置中登录，或在「第三方」中添加订阅。"
+                log_red(message)
+                QMessageBox.information(self, "无法开启代理", message)
                 return
         self.cb_pac.setEnabled(False)
         self.cb_bypass_lan.setEnabled(False)
@@ -2079,28 +2298,19 @@ class MainWindow(QWidget):
 
     # ── 设置 ──
     def _show_settings_dialog(self):
-        """打开设置弹窗（代理运行中锁定端口修改）；退出登录后立即触发登录流程。"""
+        """打开设置弹窗；第三方订阅与登录状态可独立管理。"""
         SettingsDialog(_settings_services(), 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(_login_services(), self)
+        if _open_login_dialog(self):
-        if dlg.exec_() == QDialog.Accepted:
-            _reload_nodes()
             self._refresh_subtitle()
 
     # ── 节点切换 ──
@@ -2112,8 +2322,8 @@ class MainWindow(QWidget):
         nodes = _reload_nodes()
         if not nodes:
             QMessageBox.information(self, "提示",
-                                    "nodes.json 中没有可用节点。\n"
+                                    "当前没有可用节点。\n"
-                                    "可在右下角「设置」中退出登录后重新登录获取节点配置。")
+                                    "请在右下角「设置」中登录，或在「第三方」中添加订阅。")
             return
         menu = self._build_node_menu(nodes)
         menu.exec_(self.subtitle.mapToGlobal(self.subtitle.rect().bottomLeft()))
@@ -2146,6 +2356,8 @@ class MainWindow(QWidget):
             text = f"{domain} ({country})"
             if remark and remark != domain:
                 text += f"　{remark}"
+            source_name = n.get("_rrpp_source_name") or "官方订阅"
+            text += f"　[{source_name}]"
             ipv6_only = _node_ipv6_only(n)
             reachable = _node_reachable(n)
             if ipv6_only:
@@ -2175,7 +2387,7 @@ class MainWindow(QWidget):
 
     def _refresh_subtitle(self):
         """刷新节点副标题文字。"""
-        self.subtitle.setText(f"{_endpoint_domain()} ({_endpoint_country()})")
+        self.subtitle.setText(_endpoint_label())
 
     def _update_node_label(self):
         """代理运行中时提示禁止切换节点。"""
@@ -2338,8 +2550,35 @@ class MainWindow(QWidget):
                 QMessageBox.Yes,
             )
             if reply == QMessageBox.Yes:
+                self._show_update_download_progress()
                 threading.Thread(target=self._download_update, daemon=True).start()
 
+    def _show_update_download_progress(self):
+        """在 GUI 线程创建下载进度窗口；网络线程只通过 signal 回传状态。"""
+        dialog = QProgressDialog("正在准备下载更新…", None, 0, 100, self)
+        dialog.setWindowTitle("正在下载更新")
+        dialog.setWindowModality(Qt.WindowModal)
+        dialog.setMinimumDuration(0)
+        dialog.setAutoClose(False)
+        dialog.setAutoReset(False)
+        dialog.setValue(0)
+        dialog.show()
+        self._update_progress_dialog = dialog
+
+    def _on_update_download_progress(self, percent, downloaded, total):
+        """把后台下载进度渲染到进度窗口。"""
+        dialog = self._update_progress_dialog
+        if dialog is None:
+            return
+        if total > 0:
+            dialog.setRange(0, 100)
+            dialog.setValue(max(0, min(100, percent)))
+            dialog.setLabelText(
+                f"正在下载更新… {percent}%（{_fmt_bytes(downloaded)} / {_fmt_bytes(total)}）")
+        else:
+            dialog.setRange(0, 0)
+            dialog.setLabelText(f"正在下载更新… 已下载 {_fmt_bytes(downloaded)}")
+
     def _download_update(self):
         """后台下载新版本 EXE 到 EXE 所在目录。
 
@@ -2357,58 +2596,98 @@ class MainWindow(QWidget):
 
             # 对 URL 进行安全编码，兼容空格或已编码空格
             url = urllib.parse.quote(url, safe=':/?#[]@!$&\'()*+,;=%')
-            save_path = PERSIST_DIR / f"Royirin Rosmontis Private Proxy {self._latest_version}.exe"
+            # 冻结程序必须放到当前 EXE 的目录，便于用户关闭旧版后就地替换；
+            # 源码调试则仍写入源码目录，避免误把测试文件写到 Python 安装目录。
+            if getattr(sys, 'frozen', False) or "__compiled__" in globals():
+                download_dir = _frozen_executable_dir()
+            else:
+                download_dir = PERSIST_DIR
+            save_path = download_dir / f"Royirin Rosmontis Private Proxy {self._latest_version}.exe"
+            part_path = save_path.with_suffix(save_path.suffix + ".part")
             log(f"正在下载新版本: {url}")
             log(f"保存到: {save_path}")
 
-            def fetch(target):
+            use_proxy = _update_via_proxy()
-                # 同更新检查：按「使用代理更新」设置选择隧道或直连
+            started_here = _ensure_forwarder() if use_proxy else False
-                if _update_via_proxy():
+            try:
+                if use_proxy:
                     opener = urllib.request.build_opener(urllib.request.ProxyHandler(
                         {"http": f"{PROXY_HOST}:{PROXY_PORT}",
                          "https": f"{PROXY_HOST}:{PROXY_PORT}"}))
                 else:
                     opener = _pac_no_proxy_opener()
-                with opener.open(urllib.request.Request(target), timeout=120) as resp:
-                    return resp.read()
 
-            started_here = _ensure_forwarder() if _update_via_proxy() else False
+                if part_path.exists():
-            try:
+                    part_path.unlink()
-                data = fetch(url)
+                downloaded = 0
+                hasher = hashlib.sha256()
+                last_percent = -1
+                with opener.open(urllib.request.Request(url), timeout=120) as resp, part_path.open("wb") as fp:
+                    try:
+                        response_size = int(resp.headers.get("Content-Length") or 0)
+                    except (TypeError, ValueError):
+                        response_size = 0
+                    total = self._asset_size or response_size
+                    self._update_download_progress.emit(0, 0, total)
+                    log(f"更新下载进度: 0%（0 / {_fmt_bytes(total)}）" if total else "更新下载进度: 已开始")
+                    while True:
+                        chunk = resp.read(256 * 1024)
+                        if not chunk:
+                            break
+                        fp.write(chunk)
+                        hasher.update(chunk)
+                        downloaded += len(chunk)
+                        if total:
+                            percent = min(100, downloaded * 100 // total)
+                            if percent != last_percent:
+                                last_percent = percent
+                                log(f"更新下载进度: {percent}%（{_fmt_bytes(downloaded)} / {_fmt_bytes(total)}）")
+                                self._update_download_progress.emit(percent, downloaded, total)
+
                 checksum_text = ""
                 if self._checksum_url:
-                    checksum_text = fetch(self._checksum_url).decode("ascii", "replace")
+                    checksum_url = urllib.parse.quote(
+                        self._checksum_url, safe=':/?#[]@!$&\'()*+,;=%')
+                    with opener.open(urllib.request.Request(checksum_url), timeout=120) as resp:
+                        checksum_text = resp.read().decode("ascii", "replace")
             finally:
                 _release_forwarder(started_here)
 
-            if len(data) < 1024:
+            if downloaded < 1024:
-                self._update_download_done.emit(False, "下载文件过小，可能不是有效的 EXE")
+                raise RuntimeError("下载文件过小，可能不是有效的 EXE")
-                return
+            if self._asset_size and downloaded != self._asset_size:
-            if self._asset_size and len(data) != self._asset_size:
+                raise RuntimeError(
-                self._update_download_done.emit(
+                    f"下载不完整：实际 {downloaded} 字节，应为 {self._asset_size} 字节")
-                    False, f"下载不完整：实际 {len(data)} 字节，应为 {self._asset_size} 字节")
-                return
             if checksum_text:
                 want = checksum_text.strip().split()[0].lower()
-                got = hashlib.sha256(data).hexdigest()
+                got = hasher.hexdigest()
                 if len(want) != 64 or want != got:
-                    self._update_download_done.emit(False, "更新包校验失败（SHA-256 不匹配），已放弃")
+                    raise RuntimeError("更新包校验失败（SHA-256 不匹配），已放弃")
-                    return
                 log("更新包 SHA-256 校验通过")
 
-            save_path.write_bytes(data)
+            part_path.replace(save_path)
             log(f"新版本已下载: {save_path}")
             self._update_download_done.emit(True, str(save_path))
         except Exception as e:
+            try:
+                if 'part_path' in locals():
+                    part_path.unlink(missing_ok=True)
+            except Exception:
+                pass
             log(f"下载更新失败: {e}")
             self._update_download_done.emit(False, str(e))
 
     def _on_update_download_done(self, success, message):
         """下载完成后提示用户手动替换 EXE。"""
+        dialog = self._update_progress_dialog
+        self._update_progress_dialog = None
+        if dialog is not None:
+            dialog.close()
+            dialog.deleteLater()
         if success:
             QMessageBox.information(
                 self, "下载完成",
-                f"新版本已下载到:\n{message}\n\n请关闭本程序后，用该文件替换当前运行的程序，然后重新启动。"
+                f"新版本已下载到当前程序所在文件夹:\n{message}\n\n请关闭本程序后，用该文件替换当前运行的程序，然后重新启动。"
             )
         else:
             QMessageBox.warning(self, "下载失败", f"无法下载新版本:\n{message}")

lite/others/nodes_tool/nodes_tool.py

@@ -1,126 +0,0 @@
-# -*- coding: utf-8 -*-
-"""管理用户数据目录里的 nodes.json（Windows DPAPI 加密存储）。
-
-用法（在 Windows 上运行）:
-    python nodes_tool.py encrypt <明文nodes.json>   # 加密导入到 %LOCALAPPDATA%\\RRPP
-    python nodes_tool.py decrypt                    # 打印解密后的内容
-    python nodes_tool.py path                       # 显示存储路径
-
-注意：DPAPI 加密后只有「本机 + 本用户」能解密；文件拷到别的机器/账户无效。
-"""
-import ctypes
-import json
-import os
-import sys
-from pathlib import Path
-
-
-def user_data_dir():
-    if os.name == "nt" and os.environ.get("LOCALAPPDATA"):
-        return Path(os.environ["LOCALAPPDATA"]) / "RRPP"
-    return Path.home() / "RRPP"
-
-
-class _DATA_BLOB(ctypes.Structure):
-    _fields_ = [("cbData", ctypes.c_uint32),
-                ("pbData", ctypes.POINTER(ctypes.c_char))]
-
-
-def dpapi_protect(data: bytes) -> bytes:
-    crypt32 = ctypes.WinDLL("crypt32")
-    kernel32 = ctypes.WinDLL("kernel32")
-    crypt32.CryptProtectData.argtypes = [ctypes.POINTER(_DATA_BLOB), ctypes.c_wchar_p,
-                                         ctypes.POINTER(_DATA_BLOB), ctypes.c_void_p,
-                                         ctypes.c_void_p, ctypes.c_uint32,
-                                         ctypes.POINTER(_DATA_BLOB)]
-    crypt32.CryptProtectData.restype = ctypes.c_bool
-    kernel32.LocalFree.argtypes = [ctypes.c_void_p]
-
-    buf = ctypes.create_string_buffer(data, len(data))
-    blob_in = _DATA_BLOB(len(data), ctypes.cast(buf, ctypes.POINTER(ctypes.c_char)))
-    blob_out = _DATA_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:
-    crypt32 = ctypes.WinDLL("crypt32")
-    kernel32 = ctypes.WinDLL("kernel32")
-    crypt32.CryptUnprotectData.argtypes = [ctypes.POINTER(_DATA_BLOB), ctypes.POINTER(ctypes.c_wchar_p),
-                                           ctypes.POINTER(_DATA_BLOB), ctypes.c_void_p,
-                                           ctypes.c_void_p, ctypes.c_uint32,
-                                           ctypes.POINTER(_DATA_BLOB)]
-    crypt32.CryptUnprotectData.restype = ctypes.c_bool
-    kernel32.LocalFree.argtypes = [ctypes.c_void_p]
-
-    buf = ctypes.create_string_buffer(data, len(data))
-    blob_in = _DATA_BLOB(len(data), ctypes.cast(buf, ctypes.POINTER(ctypes.c_char)))
-    descr = ctypes.c_wchar_p()
-    blob_out = _DATA_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 cmd_encrypt(src: str):
-    plain = Path(src)
-    if not plain.is_file():
-        print(f"未找到 {plain}")
-        sys.exit(1)
-    data = json.loads(plain.read_text("utf-8"))
-    if not (isinstance(data, list) and data and all(isinstance(n, dict) for n in data)):
-        print("nodes.json 应为非空节点数组")
-        sys.exit(1)
-    out_dir = user_data_dir()
-    out_dir.mkdir(parents=True, exist_ok=True)
-    out = out_dir / "nodes.json"
-    text = json.dumps(data, ensure_ascii=False, indent=2) + "\n"
-    blob = dpapi_protect(text.encode("utf-8")) if os.name == "nt" else text.encode("utf-8")
-    out.write_bytes(blob)
-    print(f"已{'加密' if os.name == 'nt' else '写入(非Windows不加密)'}: {out}（{len(data)} 个节点）")
-
-
-def cmd_decrypt():
-    f = user_data_dir() / "nodes.json"
-    if not f.is_file():
-        print(f"未找到 {f}")
-        sys.exit(1)
-    blob = f.read_bytes()
-    if os.name == "nt":
-        try:
-            print(dpapi_unprotect(blob).decode("utf-8"), end="")
-            return
-        except Exception as e:
-            print(f"DPAPI 解密失败: {e}")
-    print(blob.decode("utf-8"), end="")
-
-
-def main():
-    if len(sys.argv) < 2:
-        print(__doc__)
-        sys.exit(1)
-    cmd = sys.argv[1]
-    if cmd == "encrypt" and len(sys.argv) == 3:
-        cmd_encrypt(sys.argv[2])
-    elif cmd == "decrypt":
-        cmd_decrypt()
-    elif cmd == "path":
-        print(user_data_dir())
-    else:
-        print(__doc__)
-        sys.exit(1)
-
-
-if __name__ == "__main__":
-    main()

lite/others/update_pac_domains/update_pac_domains.py

@@ -1,143 +0,0 @@
-# -*- coding: utf-8 -*-
-"""更新 pac_domains.json：从 v2fly/domain-list-community (geosite) 拉取境外服务
-分类，与现有手工条目合并，生成新的强制走隧道域名后缀表。
-
-用法:
-    python update_pac_domains.py            # 拉取并更新（先自动备份）
-    python update_pac_domains.py --dry-run  # 只统计，不写文件
-
-geosite 数据格式（data/<分类>，每行一条规则）:
-    include:xxx      引用其他分类（递归解析）
-    example.com      匹配域名及其子域（与我们的后缀匹配一致）
-    +example.com     同上（additive）
-    .example.com     仅精确匹配（不匹配子域）-> 跳过
-    -example.com     排除 -> 跳过
-    /regex/          正则 -> 跳过
-    example.com @cn  属性标签 -> 去掉标签
-    无点的单标签（如 chrome）-> 无实际主机名会命中，跳过
-"""
-import argparse
-import json
-import sys
-import urllib.request
-from datetime import datetime
-from pathlib import Path
-
-BASE_URL = "https://raw.githubusercontent.com/v2fly/domain-list-community/master/data/"
-# 需要强制走隧道的境外服务分类（geosite 分类名），可按需增删
-CATEGORIES = ["google", "apple", "telegram", "openai", "twitter", "facebook",
-              "github", "netflix", "discord", "spotify", "whatsapp", "paypal"]
-
-SCRIPT_DIR = Path(__file__).resolve().parent
-LITE_DIR = SCRIPT_DIR.parent.parent
-PAC_FILE = LITE_DIR / "pac_domains.json"
-BACKUP_DIR = LITE_DIR / "backups"
-# 本地 pac_domains.json 已不再随项目保留：缺失时从 CDN 拉当前版本作合并基底
-CDN_URL = "https://cdn.rrrosmontis.icu/RRPP/PAC/pac_domains.json"
-
-
-def fetch(cat, tries=2):
-    for i in range(tries):
-        try:
-            with urllib.request.urlopen(BASE_URL + cat, timeout=20) as r:
-                return r.read().decode("utf-8", "replace").splitlines()
-        except Exception:
-            if i == tries - 1:
-                raise
-
-
-def resolve(cat, seen=None):
-    """递归展开 include:，返回原始行列表；循环引用用 seen 防死循环。"""
-    seen = seen or set()
-    if cat in seen:
-        return []
-    seen.add(cat)
-    lines = []
-    try:
-        for raw in fetch(cat):
-            line = raw.strip()
-            if line.startswith("include:"):
-                lines += resolve(line[len("include:"):].strip(), seen)
-            else:
-                lines.append(line)
-    except Exception as e:
-        print(f"  [warn] 拉取 {cat} 失败: {e}")
-    return lines
-
-
-def parse_domains(lines):
-    """过滤 geosite 行，只保留可映射为后缀匹配的纯域名。"""
-    out = set()
-    for line in lines:
-        s = line.strip()
-        if not s or s.startswith("#") or s.startswith("/"):
-            continue
-        if s.startswith("-") or s.startswith("."):
-            continue
-        s = s.lstrip("+").split("@")[0].strip()
-        if "." not in s:
-            continue
-        out.add(s.lower())
-    return out
-
-
-def main():
-    parser = argparse.ArgumentParser(description="更新 pac_domains.json")
-    parser.add_argument("--dry-run", action="store_true", help="只统计，不写文件")
-    args = parser.parse_args()
-
-    if PAC_FILE.is_file():
-        cur = set(json.loads(PAC_FILE.read_text("utf-8"))["domains"])
-    else:
-        # 项目目录已不保留该文件：从 CDN 拉当前版本作合并基底（保留其中的手工条目）
-        print(f"本地未找到 {PAC_FILE}，从 CDN 拉取当前版本作为合并基底")
-        try:
-            with urllib.request.urlopen(CDN_URL, timeout=20) as r:
-                cur = set(json.loads(r.read().decode("utf-8"))["domains"])
-        except Exception as e:
-            print(f"拉取失败: {e}")
-            sys.exit(1)
-
-    print(f"拉取 geosite 分类: {', '.join(CATEGORIES)}")
-    merged = set()
-    for cat in CATEGORIES:
-        doms = parse_domains(resolve(cat))
-        merged |= doms
-        print(f"  {cat:<12} -> {len(doms):>5} 条")
-
-    new = merged - cur
-    kept = cur & merged
-    only_cur = cur - merged
-    print(f"\n合并后总数: {len(merged | cur)}（geosite {len(merged)} + 现有 {len(cur)}）")
-    print(f"新增 {len(new)}，与现有重叠 {len(kept)}，现有手工条目 {len(only_cur)} 条全部保留")
-
-    if args.dry_run:
-        print("--dry-run: 未写文件")
-        return
-
-    final = sorted(merged | cur)
-    now = datetime.now().strftime("%Y-%m-%d %H:%M")
-    comment = (
-        f"强制走代理隧道的域名后缀（IP 智能分流的补充规则）：命中这些域名的连接即使"
-        f"解析到国内 IP 也强制走隧道。匹配规则：host == 条目 或 host 以 \".条目\" 结尾。"
-        f"由 others/update_pac_domains/update_pac_domains.py 从 v2fly geosite "
-        f"({', '.join(CATEGORIES)}) 生成于 {now}，手工条目已保留；"
-        f"客户端按文件内 version/date 从 CDN 在线更新到 %LOCALAPPDATA%\\RRPP。"
-    )
-    out = {"_comment": comment,
-           "version": datetime.now().strftime("%Y%m%d%H%M"),
-           "date": datetime.now().strftime("%Y-%m-%d"),
-           "domains": final}
-
-    BACKUP_DIR.mkdir(exist_ok=True)
-    bak = BACKUP_DIR / f"pac_domains.json.bak_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
-    bak.write_bytes(PAC_FILE.read_bytes())
-    print(f"已备份: {bak}")
-
-    PAC_FILE.write_text(json.dumps(out, ensure_ascii=False, indent=2) + "\n", "utf-8")
-    size = PAC_FILE.stat().st_size / 1024
-    print(f"已写入: {PAC_FILE}（{len(final)} 条，{size:.0f} KB）")
-
-
-if __name__ == "__main__":
-    main()

lite/others/update_pac_ip_cn/update_pac_ip_cn.py

@@ -1,86 +0,0 @@
-# -*- coding: utf-8 -*-
-"""更新 pac_ip_cn.json：从 17mon/china_ip_list 拉取中国 IPv4 段，写入带
-version/date 戳的本地 PAC 数据文件（客户端按 pac_manifest.json 比对在线更新）。
-
-用法:
-    python update_pac_ip_cn.py            # 拉取并更新（先自动备份）
-    python update_pac_ip_cn.py --dry-run  # 只统计，不写文件
-"""
-import argparse
-import ipaddress
-import json
-import sys
-import urllib.request
-from datetime import datetime
-from pathlib import Path
-
-SRC_URL = "https://raw.githubusercontent.com/17mon/china_ip_list/master/china_ip_list.txt"
-
-SCRIPT_DIR = Path(__file__).resolve().parent
-LITE_DIR = SCRIPT_DIR.parent.parent
-PAC_FILE = LITE_DIR / "pac_ip_cn.json"
-BACKUP_DIR = LITE_DIR / "backups"
-
-
-def fetch_lines(tries=2):
-    for i in range(tries):
-        try:
-            with urllib.request.urlopen(SRC_URL, timeout=20) as r:
-                return r.read().decode("utf-8", "replace").splitlines()
-        except Exception:
-            if i == tries - 1:
-                raise
-
-
-def main():
-    parser = argparse.ArgumentParser(description="更新 pac_ip_cn.json")
-    parser.add_argument("--dry-run", action="store_true", help="只统计，不写文件")
-    args = parser.parse_args()
-
-    print(f"拉取: {SRC_URL}")
-    cidrs, skipped = [], 0
-    for line in fetch_lines():
-        s = line.strip()
-        if not s or s.startswith("#"):
-            continue
-        try:
-            net = ipaddress.ip_network(s, strict=False)
-            if net.version == 4:
-                cidrs.append(str(net))
-            else:
-                skipped += 1
-        except ValueError:
-            skipped += 1
-    cidrs.sort(key=lambda c: (int(ipaddress.ip_network(c, strict=False).network_address),
-                              ipaddress.ip_network(c, strict=False).prefixlen))
-    print(f"共 {len(cidrs)} 条 IPv4 CIDR，跳过 {skipped} 条")
-
-    if args.dry_run:
-        print("--dry-run: 未写文件")
-        return
-
-    now = datetime.now()
-    comment = (
-        f"中国 IPv4 地址段（CIDR）。智能分流依据：解析 IP 命中列表 → 直连，"
-        f"未命中 → 走加密隧道。由 others/update_pac_ip_cn/update_pac_ip_cn.py "
-        f"从 17mon/china_ip_list 生成于 {now.strftime('%Y-%m-%d %H:%M')}；"
-        f"客户端按文件内 version/date 从 CDN 在线更新到 %LOCALAPPDATA%\\RRPP。"
-    )
-    out = {"_comment": comment,
-           "version": now.strftime("%Y%m%d%H%M"),
-           "date": now.strftime("%Y-%m-%d"),
-           "cidrs": cidrs}
-
-    BACKUP_DIR.mkdir(exist_ok=True)
-    bak = BACKUP_DIR / f"pac_ip_cn.json.bak_{now.strftime('%Y%m%d_%H%M%S')}"
-    if PAC_FILE.is_file():
-        bak.write_bytes(PAC_FILE.read_bytes())
-        print(f"已备份: {bak}")
-
-    PAC_FILE.write_text(json.dumps(out, ensure_ascii=False, indent=1) + "\n", "utf-8")
-    size = PAC_FILE.stat().st_size / 1024
-    print(f"已写入: {PAC_FILE}（{len(cidrs)} 条，{size:.0f} KB）")
-
-
-if __name__ == "__main__":
-    main()

lite/rrpp_core/third_party_subscriptions.py

@@ -0,0 +1,53 @@
+"""第三方 RRPP Client JSON 订阅的下载、校验与展示辅助。"""
+
+import json
+import urllib.request
+from urllib.parse import urlsplit, urlunsplit
+
+
+def fetch_nodes(url: str, timeout=15):
+    """直连读取一个第三方订阅，避免首次导入受系统代理残留影响。"""
+    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
+    request = urllib.request.Request(
+        url,
+        headers={"Accept": "application/json", "User-Agent": "RRPP-Lite"},
+    )
+    with opener.open(request, timeout=timeout) as response:
+        return validate_nodes(json.loads(response.read().decode("utf-8-sig")))
+
+
+def validate_nodes(data):
+    """校验 RRPP Client 节点数组，返回可安全写入本地缓存的副本。"""
+    if not isinstance(data, list) or not data:
+        raise ValueError("订阅应为非空的 JSON 节点数组")
+    result = []
+    for index, node in enumerate(data, 1):
+        if not isinstance(node, dict):
+            raise ValueError(f"第 {index} 项不是节点对象")
+        domain = str(node.get("domain") or "").strip()
+        method = (node.get("methods") or {}).get("http") or {}
+        port = method.get("port", node.get("port"))
+        try:
+            port = int(port)
+        except (TypeError, ValueError):
+            port = 0
+        if not domain:
+            raise ValueError(f"第 {index} 项缺少 domain")
+        if not 1 <= port <= 65535:
+            raise ValueError(f"第 {index} 项的 HTTP 代理端口无效")
+        result.append(dict(node))
+    return result
+
+
+def mask_url(url: str) -> str:
+    """用于 UI 展示，隐藏查询参数和 URL 用户信息中的敏感令牌。"""
+    try:
+        parts = urlsplit(str(url))
+        host = parts.hostname or ""
+        if parts.port:
+            host = f"{host}:{parts.port}"
+        if parts.username:
+            host = "***@" + host
+        return urlunsplit((parts.scheme, host, parts.path, "…" if parts.query else "", ""))
+    except Exception:
+        return "已隐藏订阅地址"

lite/rrpp_ui/dialogs.py

@@ -103,8 +103,9 @@ class LoginDialog(QDialog):
         hint.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 10px;")
         bottom.addWidget(hint)
         bottom.addStretch()
-        self.btn_cancel = QPushButton("取消")
+        self.btn_cancel = QPushButton("跳过登录")
         self.btn_cancel.setObjectName("btnCancel")
+        self.btn_cancel.setToolTip("可稍后在设置中登录，或添加第三方订阅")
         self.btn_cancel.clicked.connect(self.reject)
         bottom.addWidget(self.btn_cancel)
         self.btn_ok = QPushButton("登录")
@@ -291,6 +292,175 @@ class WhitelistDialog(QDialog):
         self.accept()
 
 
+@dataclass(frozen=True)
+class ThirdPartyServices:
+    """第三方订阅管理框所需的显式业务服务。"""
+
+    load_icon: object
+    load_subscriptions: object
+    add_subscription: object
+    refresh_subscription: object
+    remove_subscription: object
+
+
+class ThirdPartyDialog(QDialog):
+    """管理多个第三方 RRPP Client JSON 订阅。"""
+
+    _done = pyqtSignal(str, bool, str)
+
+    def __init__(self, services: ThirdPartyServices, parent=None):
+        super().__init__(parent)
+        self._services = services
+        self._busy = False
+        self.setWindowTitle("第三方订阅")
+        self.setWindowIcon(services.load_icon())
+        self.setFixedSize(520, 440)
+        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: 5px 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: 5px 7px; 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: 5px 12px; font-size: 12px; font-weight: bold; }}
+            QPushButton#btnOk:hover {{ background-color: #5ed978; }}
+            QPushButton#btnDanger {{ background-color: {C_RED}; color: #fff; border: none;
+                border-radius: 6px; padding: 5px 12px; font-size: 12px; font-weight: bold; }}
+            QPushButton#btnSmall {{ background-color: {C_CARD}; color: {C_SUBTEXT}; border: 1px solid {C_BORDER};
+                border-radius: 6px; padding: 4px 10px; font-size: 11px; }}
+            QPushButton#btnSmall:hover {{ background-color: #e3e8ee; color: {C_TEXT}; }}
+        """)
+        layout = QVBoxLayout(self)
+        layout.setSpacing(9)
+        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("仅接受 RRPP Client 的 JSON 节点数组；订阅地址和节点凭据会加密保存在本机")
+        hint.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 10px;")
+        hint.setAlignment(Qt.AlignCenter)
+        layout.addWidget(hint)
+        self.list_widget = QListWidget()
+        self.list_widget.setToolTip("选择一个订阅后可立即更新或删除")
+        layout.addWidget(self.list_widget, 1)
+        self.ed_name = QLineEdit()
+        self.ed_name.setPlaceholderText("订阅名称，例如：我的备用节点")
+        layout.addWidget(self.ed_name)
+        self.ed_url = QLineEdit()
+        self.ed_url.setPlaceholderText("订阅链接（http:// 或 https://）")
+        self.ed_url.returnPressed.connect(self._add)
+        layout.addWidget(self.ed_url)
+        self.lb_status = QLabel("")
+        self.lb_status.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 11px;")
+        self.lb_status.setFixedHeight(16)
+        layout.addWidget(self.lb_status)
+        actions = QHBoxLayout()
+        self.btn_add = QPushButton("添加")
+        self.btn_add.setObjectName("btnOk")
+        self.btn_add.clicked.connect(self._add)
+        actions.addWidget(self.btn_add)
+        self.btn_refresh = QPushButton("立即更新")
+        self.btn_refresh.setObjectName("btnSmall")
+        self.btn_refresh.clicked.connect(self._refresh)
+        actions.addWidget(self.btn_refresh)
+        self.btn_delete = QPushButton("删除")
+        self.btn_delete.setObjectName("btnDanger")
+        self.btn_delete.clicked.connect(self._delete)
+        actions.addWidget(self.btn_delete)
+        actions.addStretch()
+        self.btn_close = QPushButton("关闭")
+        self.btn_close.setObjectName("btnSmall")
+        self.btn_close.clicked.connect(self.accept)
+        actions.addWidget(self.btn_close)
+        layout.addLayout(actions)
+        self._done.connect(self._on_done)
+        self._reload_list()
+
+    def _reload_list(self, selected_id=None):
+        self.list_widget.clear()
+        for item_data in self._services.load_subscriptions():
+            text = (f"{item_data['name']}  ·  {item_data['node_count']} 个节点"
+                    f"  ·  {item_data.get('updated_at') or '未更新'}\n{item_data['url']}")
+            if item_data.get("last_error"):
+                text += f"  ·  上次更新失败：{item_data['last_error']}"
+            item = QListWidgetItem(text)
+            item.setData(Qt.UserRole, item_data["id"])
+            self.list_widget.addItem(item)
+            if selected_id and item_data["id"] == selected_id:
+                self.list_widget.setCurrentItem(item)
+
+    def _set_busy(self, busy, message=""):
+        self._busy = busy
+        for button in (self.btn_add, self.btn_refresh, self.btn_delete, self.btn_close):
+            button.setEnabled(not busy)
+        self.ed_name.setEnabled(not busy)
+        self.ed_url.setEnabled(not busy)
+        if message:
+            self.lb_status.setText(message)
+            self.lb_status.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 11px;")
+
+    def _add(self):
+        if self._busy:
+            return
+        name, url = self.ed_name.text().strip(), self.ed_url.text().strip()
+        if not name or not url:
+            self.lb_status.setText("请填写订阅名称和链接")
+            self.lb_status.setStyleSheet(f"color: {C_RED}; font-size: 11px;")
+            return
+        self._set_busy(True, "正在下载并校验订阅...")
+        threading.Thread(target=self._worker, args=("add", name, url), daemon=True).start()
+
+    def _selected_id(self):
+        item = self.list_widget.currentItem()
+        return item.data(Qt.UserRole) if item else None
+
+    def _refresh(self):
+        subscription_id = self._selected_id()
+        if not subscription_id or self._busy:
+            self.lb_status.setText("请先选择一个订阅")
+            self.lb_status.setStyleSheet(f"color: {C_RED}; font-size: 11px;")
+            return
+        self._set_busy(True, "正在更新订阅...")
+        threading.Thread(target=self._worker, args=("refresh", subscription_id, ""), daemon=True).start()
+
+    def _delete(self):
+        subscription_id = self._selected_id()
+        if not subscription_id or self._busy:
+            self.lb_status.setText("请先选择一个订阅")
+            self.lb_status.setStyleSheet(f"color: {C_RED}; font-size: 11px;")
+            return
+        self._services.remove_subscription(subscription_id)
+        self._reload_list()
+        self.lb_status.setText("已删除订阅")
+        self.lb_status.setStyleSheet(f"color: {C_GREEN}; font-size: 11px;")
+
+    def _worker(self, action, first, second):
+        try:
+            if action == "add":
+                self._services.add_subscription(first, second)
+                self._done.emit(action, True, "订阅已添加并导入")
+            else:
+                self._services.refresh_subscription(first)
+                self._done.emit(action, True, "订阅已更新")
+        except Exception as error:
+            self._done.emit(action, False, str(error))
+
+    def _on_done(self, action, ok, message):
+        self._set_busy(False)
+        if ok:
+            if action == "add":
+                self.ed_name.clear()
+                self.ed_url.clear()
+            self._reload_list(self._selected_id())
+        self.lb_status.setText(message)
+        color = C_GREEN if ok else C_RED
+        self.lb_status.setStyleSheet(f"color: {color}; font-size: 11px;")
+
+
 @dataclass(frozen=True)
 class SettingsServices:
     """设置框所需的配置与业务回调。"""
@@ -305,6 +475,9 @@ class SettingsServices:
     set_update_via_proxy: object
     set_skip_update_check: object
     open_whitelist: object
+    open_third_party: object
+    has_login: object
+    open_login: object
     logout: object
 
 
@@ -316,7 +489,7 @@ class SettingsDialog(QDialog):
         self._services = services
         self.setWindowTitle("设置")
         self.setWindowIcon(services.load_icon())
-        self.setFixedSize(300, 354)
+        self.setFixedSize(300, 396)
         self.setStyleSheet(f"""
             QDialog {{ background-color: {C_BG}; }}
             QLabel {{ color: {C_TEXT}; font-family: "Microsoft YaHei", "Segoe UI", sans-serif; font-size: 12px; }}
@@ -371,6 +544,19 @@ class SettingsDialog(QDialog):
         whitelist_row.addWidget(self.btn_whitelist)
         layout.addLayout(whitelist_row)
 
+        third_party_row = QHBoxLayout()
+        third_party_label = QLabel("第三方")
+        third_party_label.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;")
+        third_party_label.setToolTip("添加、更新或删除第三方 RRPP Client JSON 订阅")
+        third_party_row.addWidget(third_party_label)
+        third_party_row.addStretch()
+        self.btn_third_party = QPushButton("添加")
+        self.btn_third_party.setCursor(Qt.PointingHandCursor)
+        self.btn_third_party.setStyleSheet(self.btn_whitelist.styleSheet())
+        self.btn_third_party.clicked.connect(self._open_third_party)
+        third_party_row.addWidget(self.btn_third_party)
+        layout.addLayout(third_party_row)
+
         port_row = QHBoxLayout()
         port_label = QLabel("代理转发端口")
         port_label.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;")
@@ -415,13 +601,13 @@ class SettingsDialog(QDialog):
         self.btn_save.setEnabled(not port_locked)
         self.btn_save.clicked.connect(self._save_settings)
         buttons.addWidget(self.btn_save)
-        self.btn_logout = QPushButton("退出登录")
+        self.btn_logout = QPushButton("退出登录" if services.has_login() else "登录")
         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.setToolTip("登录获取官方节点；退出登录不会删除第三方订阅")
         self.btn_logout.clicked.connect(self._logout)
         buttons.addWidget(self.btn_logout)
         buttons.addStretch()
@@ -450,6 +636,9 @@ class SettingsDialog(QDialog):
     def _open_whitelist(self):
         self._services.open_whitelist(self)
 
+    def _open_third_party(self):
+        self._services.open_third_party(self)
+
     def _save_settings(self):
         port = self._services.parse_port(self.ed_port.text())
         if port is None:
@@ -464,5 +653,10 @@ class SettingsDialog(QDialog):
         self.lb_hint.setStyleSheet(f"color: {C_GREEN}; font-size: 11px;")
 
     def _logout(self):
-        self._services.logout()
+        if self._services.has_login():
-        self.accept()
+            self._services.logout()
+            self.btn_logout.setText("登录")
+        else:
+            self._services.open_login(self)
+            if self._services.has_login():
+                self.btn_logout.setText("退出登录")

lite/test.bat

@@ -1,4 +0,0 @@
-@echo off
-cd /d "%~dp0"
-call ..\.venv\Scripts\python.exe main.py
-pause

lite/test/PyQt5/QtCore.py

@@ -1,10 +0,0 @@
-class Qt: pass
-class QTimer: pass
-
-class _Signal:
-    def connect(self, *a, **k): pass
-    def emit(self, *a, **k): pass
-
-def pyqtSignal(*a, **k):
-    return _Signal()
-class QRectF: pass

lite/test/PyQt5/QtGui.py

@@ -1,6 +0,0 @@
-class QFont: pass
-class QIcon: pass
-class QPixmap: pass
-class QImage: pass
-class QColor: pass
-class QPainter: pass

lite/test/PyQt5/QtWidgets.py

@@ -1,28 +0,0 @@
-class QApplication: pass
-class QWidget: pass
-class QLabel:
-    def __init__(self, text=""):
-        self.text = text
-    def setStyleSheet(self, *a): pass
-    def setToolTip(self, *a): pass
-    def setFixedSize(self, *a): pass
-    def setFixedHeight(self, *a): pass
-    def setAlignment(self, *a): pass
-    def setPixmap(self, *a): pass
-    def setText(self, *a): pass
-    def setCursor(self, *a): pass
-    def unsetCursor(self): pass
-    def hide(self): pass
-    def show(self): pass
-class QPushButton: pass
-class QVBoxLayout: pass
-class QHBoxLayout: pass
-class QFrame: pass
-class QMessageBox: pass
-class QTextEdit: pass
-class QCheckBox: pass
-class QMenu: pass
-class QDialog: Accepted = 1
-class QLineEdit: Password = 2
-class QListWidget: pass
-class QListWidgetItem: pass

lite/test/PyQt5/__init__.py

@@ -1,2 +0,0 @@
-# PyQt5 桩模块：仅供 WSL 下 import main 做路由逻辑测试，
-# 决策函数（_decide_direct 等）不依赖任何 Qt 对象。

lite/test/check_node_menu.py

@@ -1,127 +0,0 @@
-# -*- coding: utf-8 -*-
-"""离屏验证节点菜单：IPv6-only 节点的标注与置灰（真实 PyQt5，不出窗口）。
-
-需要 Windows venv 的真实 PyQt5：
-  ..\\.venv\\Scripts\\python.exe test\\check_node_menu.py
-读取 test/nodes.json 作为节点数据，只构建菜单不进事件循环。
-"""
-import os, sys
-os.environ["QT_QPA_PLATFORM"] = "offscreen"
-from pathlib import Path
-
-LITE = Path(__file__).resolve().parent.parent
-# 脚本所在目录（test/）会被解释器自动放到 sys.path 首位，那里有 PyQt5 桩，
-# 会顶掉真实 PyQt5 —— 必须移除，改用本程序自己的 venv
-sys.path = [str(LITE)] + [p for p in sys.path if Path(p).resolve() != Path(__file__).resolve().parent]
-
-sys.stdout.reconfigure(encoding="utf-8")
-
-import main as pp
-from PyQt5.QtWidgets import QApplication
-
-# 构造窗口时不触发用户数据初始化或远程同步；本测试只验证静态菜单和本地节点状态。
-pp._user_init_local = lambda: None
-pp._user_init_remote = lambda *args, **kwargs: None
-pp._nodes_sync_remote = lambda *args, **kwargs: pp._NODES_SYNC_DONE.set()
-pp._NODES_SYNC_DONE.set()
-
-app = QApplication(sys.argv)
-w = pp.MainWindow()
-w.timer.stop()                      # 停掉状态轮询，纯静态检查
-failures = []
-
-pp._USER_DIR = LITE / "test"        # 真实 nodes.json 所在目录
-nodes = pp._reload_nodes()
-print(f"本机有全局 IPv6: {pp._local_has_ipv6()}")
-print(f"节点数: {len(nodes)}")
-
-menu = w._build_node_menu(nodes)
-print("\n菜单项（enabled / 文案 / tooltip）:")
-for act in menu.actions():
-    print(f"  {str(act.isEnabled()):5} | {act.text()}")
-    if act.toolTip():
-        print(f"        └ {act.toolTip()}")
-
-# 点击被禁用的项不应切换节点
-before = pp._selected_index()
-for act in menu.actions():
-    if not act.isEnabled():
-        act.trigger()
-print(f"\n点击禁用项后节点下标: {before} -> {pp._selected_index()} (应不变)")
-if pp._selected_index() != before:
-    failures.append("点击不可用节点后发生了切换")
-
-# ── 上次选中的节点已不可用：启动时自动回退到第一个可达节点 ──────────
-print("\n启动校验（_verify_selected_node）:")
-bad = [i for i, n in enumerate(nodes) if not pp._node_reachable(n)]
-if not bad:
-    print("  本机所有节点均可达，无法验证回退（本机有 IPv6？）")
-else:
-    import json
-    _bak = (pp._USER_DIR / "settings.json").read_bytes() if (pp._USER_DIR / "settings.json").is_file() else None
-    bad_idx = bad[0]
-    # 模拟上次持久化选中了不可用节点：内存下标与 current_node 必须一致，
-    # 否则 _verify_selected_node 会先按真实保存的域名重新匹配，测试并非回退分支。
-    pp._settings_save({**pp._settings_load(), "current_node": nodes[bad_idx].get("domain") or ""})
-    pp._SELECTED_NODE = bad_idx
-    switched = pp._verify_selected_node()
-    expect = pp._first_reachable_index(nodes)
-    ok = switched and pp._selected_index() == expect
-    print(f"  上次选中 {nodes[bad_idx].get('domain')}(下标 {bad_idx}) 不可用 -> "
-          f"回退到下标 {pp._selected_index()} ({nodes[pp._selected_index()].get('domain')}) "
-          f"{'PASS' if ok else 'FAIL'}")
-    if not ok:
-        failures.append("不可用节点未回退到首个可达节点")
-    saved = json.loads((pp._USER_DIR / "settings.json").read_text("utf-8")).get("current_node") if (pp._USER_DIR / "settings.json").is_file() else ""
-    ok2 = saved == str(nodes[expect].get("domain") or "")
-    print(f"  回退结果已写入设置 current_node={saved!r} {'PASS' if ok2 else 'FAIL'}")
-    if not ok2:
-        failures.append("节点回退未写回 current_node")
-    # 再校验一次：已经是可达节点，不应再切换
-    ok3 = pp._verify_selected_node() is False
-    print(f"  再次校验不重复切换 {'PASS' if ok3 else 'FAIL'}")
-    if not ok3:
-        failures.append("已回退节点被重复切换")
-    if _bak is None:
-        (pp._USER_DIR / "settings.json").unlink(missing_ok=True)
-    else:
-        (pp._USER_DIR / "settings.json").write_bytes(_bak)
-
-# ── 服务端调整节点顺序：按域名重匹配而不是死守下标 ──────────────────
-print("\n服务端调序（_rematch_saved_domain）:")
-if len(nodes) < 2:
-    print("  节点不足 2 个，跳过")
-else:
-    import json
-    dom_old = nodes[0].get("domain")
-    ordered = [nodes[1], nodes[0]] + nodes[2:]     # 模拟服务端交换了前两个节点
-    pp._NODES = ordered
-    # 设置里记录的是原下标 0 的域名，重匹配后下标应跟随域名（新列表里是 1）
-    pp._settings_save({**pp._settings_load(), "current_node": dom_old})
-    switched = pp._rematch_saved_domain()
-    ok = switched and ordered[pp._selected_index()].get("domain") == dom_old
-    print(f"  current_node={dom_old} 调序后跟随域名 -> 下标 {pp._selected_index()} "
-          f"{'PASS' if ok else 'FAIL'}")
-    if not ok:
-        failures.append("节点调序后没有按域名重新匹配")
-    # 已是正确域名时再跑一次：不应重复切换
-    ok2 = pp._rematch_saved_domain() is False
-    print(f"  再次重匹配不重复切换 {'PASS' if ok2 else 'FAIL'}")
-    if not ok2:
-        failures.append("节点调序重匹配重复切换")
-    # 域名已不在列表（节点被移除）：不切换，保持现状
-    pp._settings_save({**pp._settings_load(), "current_node": "gone.example.com"})
-    ok3 = pp._rematch_saved_domain() is False
-    print(f"  节点被移除时不动 {'PASS' if ok3 else 'FAIL'}")
-    if not ok3:
-        failures.append("已移除节点仍发生切换")
-    pp._NODES = None   # 还原缓存，后续用例重读文件
-    # 还原 settings.json，避免测试污染
-    if _bak is None:
-        (pp._USER_DIR / "settings.json").unlink(missing_ok=True)
-    else:
-        (pp._USER_DIR / "settings.json").write_bytes(_bak)
-
-if failures:
-    raise SystemExit("FAIL: " + "; ".join(failures))
-print("OK")

lite/test/nodes.json

@@ -1,21 +0,0 @@
-[
-  {
-    "domain": "example.com",
-    "port": 443,
-    "country": "TEST",
-    "remark": "普通测试节点"
-  },
-  {
-    "domain": "example.org",
-    "port": 443,
-    "country": "TEST",
-    "remark": "备用测试节点"
-  },
-  {
-    "domain": "ipv6.test.invalid",
-    "port": 443,
-    "country": "TEST",
-    "ipv6_only": true,
-    "remark": "IPv6 测试节点"
-  }
-]

lite/test/probe_nodes.py

@@ -1,151 +0,0 @@
-# -*- coding: utf-8 -*-
-"""连通性探测：用 test/nodes.json 的真实节点，走 _tls_start 转发逻辑实测。
-
-仅需标准库；不启动 GUI（main 的 GUI 在 __main__ 保护下）。
-用法: python3 test/probe_nodes.py [JP US ...]   不带参数测全部
-"""
-import os, sys, time, json, ssl, socket, urllib.request, urllib.error
-from pathlib import Path
-
-HERE = Path(__file__).resolve().parent
-LITE = HERE.parent
-sys.path.insert(0, str(HERE))   # PyQt5 桩
-sys.path.insert(0, str(LITE))
-
-for k in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY", "all_proxy", "ALL_PROXY"):
-    os.environ.pop(k, None)
-# 不能设 NO_PROXY=*：urllib 会把 127.0.0.1 也当直连，显式 ProxyHandler 被绕过
-os.environ["NO_PROXY"] = ""
-
-import main as pp
-
-NODES = json.loads((HERE / "nodes.json").read_text("utf-8"))
-want = [a.upper() for a in sys.argv[1:]]
-if want:
-    NODES = [n for n in NODES if str(n.get("country", "")).upper() in want]
-
-TIMEOUT = 12
-
-
-def free_port():
-    s = socket.socket(); s.bind(("127.0.0.1", 0))
-    p = s.getsockname()[1]; s.close(); return p
-
-
-def http_method(node):
-    return (node.get("methods") or {}).get("http") or {}
-
-
-def as_flat_node(node):
-    """把 nodes.json 的 methods.http 结构摊平成代码 _node_val() 期望的形状。
-    仅用于探测：暴露「代码读不到真实节点字段」这一问题。"""
-    m = http_method(node)
-    return dict(node, port=int(m.get("port")), username=m.get("username", ""),
-                password=m.get("password", ""))
-
-
-def probe_connect(host, port, sni=None, connect_timeout=8):
-    """直连探测：DNS -> TCP -> TLS -> 发 CONNECT 127.0.0.1:9 看上游代理响应。"""
-    res = {"dns": None, "tcp": None, "tls": None, "proxy": None, "err": None}
-    t0 = time.time()
-    try:
-        infos = socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM)
-        res["dns"] = sorted({i[4][0] for i in infos})
-    except Exception as e:
-        res["err"] = f"DNS: {e}"
-        return res
-    addr = res["dns"][0]
-    af = socket.AF_INET6 if ":" in addr else socket.AF_INET
-    try:
-        s = socket.socket(af, socket.SOCK_STREAM)
-        s.settimeout(connect_timeout)
-        s.connect((addr, port))
-        res["tcp"] = round(time.time() - t0, 2)
-    except Exception as e:
-        res["err"] = f"TCP: {e}"
-        return res
-    try:
-        ctx = ssl.create_default_context()
-        s = ctx.wrap_socket(s, server_hostname=sni or host)
-        res["tls"] = s.version()
-    except Exception as e:
-        res["err"] = f"TLS: {e}"
-        try: s.close()
-        except Exception: pass
-        return res
-    try:
-        s.sendall(b"CONNECT 127.0.0.1:9 HTTP/1.1\r\nHost: 127.0.0.1:9\r\n\r\n")
-        res["proxy"] = s.recv(200).split(b"\r\n", 1)[0].decode("latin-1", "replace")
-    except Exception as e:
-        res["err"] = f"CONNECT: {e}"
-    try: s.close()
-    except Exception: pass
-    return res
-
-
-def probe_via_forwarder(node):
-    """真实转发器链路：_tls_start -> 经 127.0.0.1 代理请求，取出口 IP 与归属地。"""
-    m = http_method(node)
-    if not m:
-        return "（该节点无 http 方法，跳过）"
-    old_node, old_port = pp._current_node, pp._TLS_LOCAL_PORT
-    port = free_port()
-    try:
-        # _proxy_auth_basic() 内部走 _current_node()，固定成本次待测节点
-        pp._current_node = lambda: node
-        pp._TLS_LOCAL_PORT = port
-        pp._tls_stop()
-        pp._tls_start(node["domain"], int(m["port"]), use_tls=True,
-                      bypass_lan=False, smart=False, quiet=True)
-        ph = urllib.request.ProxyHandler({"http": f"127.0.0.1:{port}",
-                                          "https": f"127.0.0.1:{port}"})
-        op = urllib.request.build_opener(ph)
-        t0 = time.time()
-        # 明文 HTTP 代理（走 GET 绝对 URI）与 CONNECT 隧道各测一条
-        with op.open("http://ip-api.com/json/?fields=query,country,regionName,isp",
-                     timeout=TIMEOUT) as r:
-            info = json.loads(r.read(512).decode("utf-8", "replace"))
-        with op.open("https://ifconfig.me/ip", timeout=TIMEOUT) as r:
-            https_ip = r.read(64).decode().strip()
-        dt = round(time.time() - t0, 2)
-        return (f"OK {dt}s  出口IP={info.get('query')} "
-                f"[{info.get('country')} {info.get('regionName')} / {info.get('isp')}]  "
-                f"HTTPS隧道出口={https_ip}")
-    except urllib.error.HTTPError as e:
-        return f"HTTP {e.code}: {e.read(120).decode('utf-8', 'replace').strip()[:100]}"
-    except Exception as e:
-        return f"失败: {type(e).__name__}: {e}"
-    finally:
-        pp._tls_stop()
-        pp._current_node, pp._TLS_LOCAL_PORT = old_node, old_port
-
-
-print(f"节点数: {len(NODES)}")
-n0 = NODES[0]
-print(f"代码 _endpoint_domain()={pp._endpoint_domain()!r} "
-      f"_endpoint_port()={pp._endpoint_port()!r} _endpoint_country()={pp._endpoint_country()!r}")
-print(f"代码 _proxy_auth_basic()={pp._proxy_auth_basic()!r}")
-print(f"nodes.json[0] 顶层键 = {list(n0.keys())}")
-print(f"nodes.json[0] domain={n0.get('domain')!r}  methods.http.port="
-      f"{http_method(n0).get('port')!r}")
-print()
-
-for n in NODES:
-    c = n.get("country", "?")
-    m = http_method(n)
-    print("=" * 78)
-    print(f"[{c}] {n.get('domain')}:{m.get('port')}  provider={n.get('provider')} "
-          f"traffic={n.get('traffic')} level={n.get('available_level')}")
-    if not m:
-        print("    该节点无 http 方法，跳过")
-        continue
-    r = probe_connect(n["domain"], int(m["port"]))
-    print(f"    DNS    : {r['dns']}")
-    print(f"    TCP    : {r['tcp']}s" if r["tcp"] is not None else "    TCP    : -")
-    print(f"    TLS    : {r['tls']}" if r.get("tls") else "    TLS    : -")
-    print(f"    CONNECT: {r['proxy']}" if r.get("proxy") else "    CONNECT: -")
-    if r.get("err"):
-        print(f"    错误   : {r['err'].splitlines()[0][:200]}")
-    print(f"    经转发器: {probe_via_forwarder(n)}")
-    print(f"    经转发器(摊平节点): {probe_via_forwarder(as_flat_node(n))}")
-    print()

lite/test/test_core_modules.py

@@ -1,114 +0,0 @@
-# -*- coding: utf-8 -*-
-"""拆分后 rrpp_core 的独立回归测试（不依赖 Qt 或真实代理连接）。"""
-
-import sys
-import tempfile
-import unittest
-from pathlib import Path
-
-
-LITE_DIR = Path(__file__).resolve().parent.parent
-sys.path.insert(0, str(LITE_DIR))
-
-from rrpp_core.forwarder_primitives import (
-    force_http_close,
-    inject_proxy_auth,
-    parse_connect_authority,
-    parse_http_absolute,
-)
-from rrpp_core.node_api import NodeAuthError, b64url
-from rrpp_core.node_secrets import NodeSecretsStore
-from rrpp_core.network_primitives import is_private_host, is_private_ipv4_int, ttl_cache
-from rrpp_core.pac_data import atomic_write, read_version, validate_document
-from rrpp_core.routing_primitives import (
-    build_suffix_trie,
-    merge_ip_ranges,
-    parse_domain_rules,
-    suffix_trie_matches,
-)
-from rrpp_core.settings_store import JsonSettingsStore
-from rrpp_core.traffic import TrafficStats, format_bytes
-
-
-class CoreModuleTests(unittest.TestCase):
-    def test_domain_rules_and_trie(self):
-        suffixes, exact, regexes, invalid = parse_domain_rules(
-            ["example.com", "full:only.example", "regexp:^edge-[0-9]+\\.example$", "regexp:["]
-        )
-        self.assertEqual(invalid, 1)
-        self.assertIn("example.com", suffixes)
-        self.assertIn("only.example", exact)
-        self.assertTrue(regexes[0].search("edge-42.example"))
-        trie = build_suffix_trie(suffixes)
-        self.assertTrue(suffix_trie_matches("cdn.example.com", trie))
-        self.assertFalse(suffix_trie_matches("example.net", trie))
-        self.assertEqual(
-            merge_ip_ranges(["10.0.0.0/25", "10.0.0.128/25"], 4),
-            [(167772160, 167772415)],
-        )
-
-    def test_forwarder_primitives(self):
-        self.assertEqual(
-            parse_connect_authority("CONNECT [::1]:443 HTTP/1.1"), ("::1", 443)
-        )
-        self.assertEqual(
-            parse_http_absolute("GET https://example.com/path HTTP/1.1"),
-            ("example.com", 443),
-        )
-        request = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n"
-        injected = inject_proxy_auth(request, "token")
-        self.assertIn(b"Authorization: Basic token\r\n", injected)
-        self.assertEqual(inject_proxy_auth(injected, "different"), injected)
-        self.assertIn(b"Connection: close\r\n", force_http_close(request))
-
-    def test_settings_pac_and_node_secrets_storage(self):
-        with tempfile.TemporaryDirectory() as temporary:
-            directory = Path(temporary)
-            settings = JsonSettingsStore(lambda: directory / "settings.json")
-            self.assertEqual(settings.load(), {})
-            settings.save({"local_port": 1080})
-            self.assertEqual(settings.load(), {"local_port": 1080})
-
-            payload = {"version": "1", "domains": ["example.com"]}
-            atomic_write(directory, "pac_domains", payload)
-            self.assertEqual(read_version(directory / "pac_domains.json"), ("1", ""))
-            self.assertTrue(validate_document("pac_domains", payload))
-
-            secrets = NodeSecretsStore(lambda: directory)
-            self.assertEqual(secrets.decrypt_bytes(secrets.encrypt_text("round-trip")), "round-trip")
-            secrets.save_auth("test-user", "test-password")
-            self.assertEqual(
-                secrets.load_auth(), {"username": "test-user", "password": "test-password"}
-            )
-
-    def test_traffic_and_node_protocol_helpers(self):
-        times = iter((10.0, 12.0))
-        stats = TrafficStats(clock=lambda: next(times))
-        stats.reset()
-        stats.add(up=20, down=10)
-        self.assertEqual(stats.rate(), (10, 5, 20, 10))
-        self.assertEqual(format_bytes(1536), "1.5 KB")
-        self.assertEqual(b64url("a/b="), "YS9iPQ")
-        error = NodeAuthError("denied", 403)
-        self.assertEqual((str(error), error.code), ("denied", 403))
-
-    def test_network_primitives_and_ttl_cache(self):
-        self.assertTrue(is_private_ipv4_int(0x0A010203))
-        self.assertTrue(is_private_host("::1"))
-        self.assertFalse(is_private_host("8.8.8.8"))
-        calls = []
-
-        @ttl_cache(300)
-        def probe(value):
-            calls.append(value)
-            return value
-
-        self.assertEqual((probe("x"), probe("x")), ("x", "x"))
-        self.assertEqual(calls, ["x"])
-        probe.cache_clear()
-        probe("x")
-        self.assertEqual(calls, ["x", "x"])
-
-
-if __name__ == "__main__":
-    unittest.main(verbosity=2)

lite/test/test_pac_routing.py

@@ -1,739 +0,0 @@
-# -*- coding: utf-8 -*-
-"""PAC（智能分流）模式路由决策测试。
-
-核心问题：PAC 模式下，目标域名解析为国外 IP 时，是否强制走隧道代理？
-
-判定链（main._decide_direct）：
-  0. 代理白名单命中 -> 强制直连（最优先，PAC/固定模式均生效）
-  1. pac_domains.json 域名表命中 -> 强制隧道（优先于 IP 判定）
-  2. 内网 IP 且勾选 bypass_lan -> 直连
-  3. 智能分流：中国 IP -> 直连；其余（国外 IP）-> 走隧道
-
-直接 import main 调用真实决策函数，数据用真实的 pac_ip_cn.json /
-pac_domains.json；PyQt5 用桩替代（决策逻辑不依赖 Qt）。
-"""
-
-import sys
-import threading
-from pathlib import Path
-
-TEST_DIR = Path(__file__).resolve().parent
-LITE_DIR = TEST_DIR.parent
-
-sys.path.insert(0, str(TEST_DIR))  # 假 PyQt5 桩
-sys.path.insert(0, str(LITE_DIR))
-
-import json
-import socket
-import main as pp
-from main import _decide_direct, _is_china_ip, _is_force_proxy, _is_private_ip, _is_whitelisted
-
-# PAC 数据已不再随项目保留：从 CDN 拉取到临时用户目录（真实数据端到端）
-import tempfile
-pp._USER_DIR = Path(tempfile.mkdtemp()) / "RRPP"
-pp._user_init_remote(base_url="https://cdn.rrrosmontis.icu/RRPP/PAC/")
-
-pp._load_cn_ranges()
-pp._load_force_proxy_suffixes()
-if not pp._CN_RANGES:
-    print("警告: 未能获取 pac_ip_cn.json（CDN 不可达？），中国 IP 判定将失效")
-print(f"中国IP段: {len(pp._CN_RANGES or [])} 段, 强制域名表: {len(pp._FORCE_PROXY_SUFFIXES or [])} 条\n")
-
-
-def resolved_ips(host):
-    try:
-        return sorted({a[4][0] for a in socket.getaddrinfo(host, 443, socket.AF_INET)})
-    except Exception:
-        return []
-
-
-def mark(ip):
-    if _is_private_ip(ip):
-        return "内网"
-    if _is_china_ip(ip):
-        return "中国IP"
-    return "国外IP"
-
-
-PATH = {True: "直连", False: "隧道"}
-MODES = [
-    ("PAC+bypass_lan", dict(bypass_lan=True, smart=True)),   # 默认勾选状态
-    ("PAC无bypass_lan", dict(bypass_lan=False, smart=True)),
-    ("固定代理模式",    dict(bypass_lan=False, smart=False)),  # 全部走隧道
-]
-
-# (名称, host, 期望 PAC+bypass_lan 决策: True=直连 / False=隧道)
-cases = [
-    ("境外域名 google.com",       "google.com",          False),
-    ("境外域名 github.com",       "github.com",          False),
-    ("境外域名 youtube.com",      "youtube.com",         False),
-    ("域名表强制 claude.ai",      "claude.ai",           False),
-    ("域名表强制 *.google 后缀",  "antigravity.google",  False),
-    ("国内域名 www.baidu.com",    "www.baidu.com",       True),
-    ("国内域名 www.qq.com",       "www.qq.com",          True),
-    ("内网 localhost",            "localhost",           True),
-    ("内网 IP 192.168.1.1",       "192.168.1.1",         True),
-    ("境外 IP 8.8.8.8",           "8.8.8.8",             False),
-    ("国内 IP 114.114.114.114",   "114.114.114.114",     True),
-]
-
-failed = 0
-for name, host, expect in cases:
-    ips = resolved_ips(host)
-    force = _is_force_proxy(host)
-    marks = f"[{','.join(mark(ip) for ip in ips)}]" if ips else "[解析失败]"
-    row = f"{name:<22} {host:<18} {str(ips) if ips else '无IP':<45} {marks}"
-    results = []
-    for mode_name, kw in MODES:
-        got = _decide_direct(host, **kw)
-        results.append(f"{mode_name}={PATH[got]}")
-    pac = _decide_direct(host, **MODES[0][1])
-    ok = pac == expect
-    failed += 0 if ok else 1
-    print(f"{row}\n    {'域名表命中' if force else '域名表未命中':<10} | {'  '.join(results)} | 期望 {PATH[expect]} -> {'PASS' if ok else 'FAIL'}")
-
-print(f"\n{'='*70}")
-print(f"结论: {'全部 PASS' if not failed else f'{failed} 个 FAIL'}")
-print("=" * 70)
-print("PAC 模式(智能分流)下: 域名解析结果全部为国外 IP -> 走隧道代理 ✓"
-      if not failed else "存在 FAIL，请检查上方明细")
-
-# ── 附加测试：不在域名表的被墙境外站 ─────────────────────
-# 这类站点不能靠域名表兜底，必须仅凭「解析结果为国外 IP」强制走隧道，
-# 否则会尝试直连 -> 被墙。
-print("\n" + "=" * 70)
-print("附加: 不在 pac_domains.json 的被墙境外站（只能靠 IP 判定走隧道）")
-print("=" * 70)
-
-extra_candidates = [
-    "dailymotion.com", "vimeo.com", "notion.so", "linear.app",
-    "quora.com", "feedly.com", "dropbox.com", "tumblr.com",
-    "flickr.com", "pastebin.com", "fandom.com", "archive.org",
-    "godaddy.com", "imgur.com", "kickstarter.com", "shopify.com",
-]
-
-extra_failed = 0
-skipped = 0
-for host in extra_candidates:
-    if _is_force_proxy(host):
-        print(f"{host:<20} 已在域名表中，跳过")
-        skipped += 1
-        continue
-    ips = resolved_ips(host)
-    if not ips:
-        print(f"{host:<20} 解析失败，跳过")
-        skipped += 1
-        continue
-    foreign = all(not _is_private_ip(ip) and not _is_china_ip(ip) for ip in ips)
-    if not foreign:
-        print(f"{host:<20} 解析到非国外 IP {ips}，跳过（不是有效的境外用例）")
-        skipped += 1
-        continue
-    got = _decide_direct(host, bypass_lan=True, smart=True)  # PAC 模式
-    if got is not False:
-        # 决策函数内部会重新解析域名，CDN 轮询 DNS 可能与其上方展示的解析结果
-        # 不一致：清缓存重试一次，仍不一致则按 DNS 不稳定跳过（非代码问题）
-        _is_china_ip.cache_clear()
-        _is_private_ip.cache_clear()
-        ips = resolved_ips(host)
-        got = _decide_direct(host, bypass_lan=True, smart=True)
-        if got is not False:
-            print(f"{host:<20} 解析/判定随 DNS 轮询不一致 {ips}，跳过（非代码问题）")
-            skipped += 1
-            continue
-        print(f"{host:<20} 重试后走隧道（DNS 瞬时差异） PASS")
-        continue
-    extra_failed += 0
-    print(f"{host:<20} {str(ips):<50} [{'，'.join(mark(ip) for ip in ips)}] -> "
-          f"{PATH[got]} (期望隧道) PASS")
-
-print(f"\n附加测试: {len(extra_candidates)} 个候选, 有效用例 {len(extra_candidates) - skipped} 个, "
-      f"{'全部 PASS' if not extra_failed else f'{extra_failed} 个 FAIL'}")
-if extra_failed:
-    failed += extra_failed
-print("=" * 70)
-
-# ── 代理白名单测试（用户可配置：命中即直连，最优先，含 IPv4/IPv6/CIDR） ─────
-print("\n" + "=" * 70)
-print("代理白名单: 命中规则强制直连（优先于域名表/固定代理，支持 IPv4/IPv6/CIDR）")
-print("=" * 70)
-
-wl_cases = [
-    # (规则, 测试host, 期望命中)
-    ("example.com",         "example.com",          True),
-    ("example.com",         "www.example.com",      True),
-    ("example.com",         "sub.www.example.com",  True),
-    ("example.com",         "example.org",          False),
-    ("claude.ai",           "claude.ai",            True),   # 域名表强制隧道，白名单优先
-    ("localhost",           "localhost",            True),
-    ("8.8.8.8",             "8.8.8.8",              True),   # IPv4 字面量
-    ("8.8.8.8",             "8.8.4.4",              False),
-    ("10.0.0.0/8",          "10.1.2.3",             True),   # IPv4 CIDR
-    ("192.0.2.0/24",        "192.0.2.55",           True),   # IPv4 CIDR
-    ("2001:db8::/32",       "2001:db8::1",          True),   # IPv6 网段
-    ("fe80::1",             "fe80::1",              True),   # IPv6 单地址
-]
-
-# 注入白名单规则（绕过 settings.json 直接改内存缓存；版本 +1 使匹配缓存失效），
-# 另加一条"禁用"的规则验证不参与匹配
-pp._WHITELIST = [{"value": "203.0.113.0/24", "enabled": False}] + \
-                [{"value": r, "enabled": True} for r, _, _ in wl_cases]
-wl_ver = pp._WHITELIST_VERSION + 1
-pp._WHITELIST_VERSION = wl_ver
-
-wl_failed = 0
-for rule, host, expect in wl_cases:
-    got = _is_whitelisted(host, wl_ver)
-    ok = got == expect
-    wl_failed += 0 if ok else 1
-    print(f"规则 {rule:<16} host {host:<22} 命中={got} 期望={expect} -> {'PASS' if ok else 'FAIL'}")
-got = _is_whitelisted("203.0.113.9", wl_ver)  # 禁用的规则不应命中
-print(f"禁用规则 203.0.113.0/24 host 203.0.113.9 -> 命中={got} 期望=False -> {'PASS' if not got else 'FAIL'}")
-wl_failed += 0 if not got else 1
-
-# 白名单在三种模式下都强制直连（固定代理模式原本全部走隧道）
-print("\n白名单在三种模式下的决策（命中白名单应为直连）:")
-for mode_name, kw in MODES:
-    got = _decide_direct("www.example.com", **kw)
-    ok = got is True
-    wl_failed += 0 if ok else 1
-    print(f"  {mode_name:<16} www.example.com -> {PATH[got]} (期望 直连) {'PASS' if ok else 'FAIL'}")
-# 白名单优先于域名表：claude.ai 在 pac_domains.json 中强制隧道，白名单命中应直连
-got = _decide_direct("claude.ai", bypass_lan=True, smart=True)
-ok = got is True
-wl_failed += 0 if ok else 1
-print(f"  {'域名表冲突':<14} claude.ai -> {PATH[got]} (期望 直连，白名单优先) {'PASS' if ok else 'FAIL'}")
-# 未命中白名单的境外域名仍走隧道（白名单不影响其他路由）
-got = _decide_direct("google.com", bypass_lan=True, smart=True)
-ok = got is False
-wl_failed += 0 if ok else 1
-print(f"  {'未命中白名单':<14} google.com -> {PATH[got]} (期望 隧道) {'PASS' if ok else 'FAIL'}")
-
-# 恢复空白名单，避免影响后续用例
-pp._WHITELIST = None
-pp._WHITELIST_VERSION += 1
-if wl_failed:
-    failed += wl_failed
-print("=" * 70)
-
-# ── IPv6-only 节点标注 ─────────────────────────────────────
-# IPv6 专用节点只提供 AAAA 记录，本机没有 IPv6 出口时连不上：
-# 客户端据此在节点菜单标注并置灰。判定三级：ipv6_only 字段 →
-# remark 里的 "(IPv6)"（服务端 rrpp_client 现网唯一的线索）→ 域名解析兜底。
-print("\n" + "=" * 70)
-print("IPv6 专用节点：识别（三级判定）与可达性")
-print("=" * 70)
-
-ipv6_failed = 0
-has_v6 = pp._local_has_ipv6()
-print(f"本机是否有全局 IPv6 地址: {has_v6}")
-
-# 用真实 nodes.json（test 目录）；没有则跳过
-_nodes_file = TEST_DIR / "nodes.json"
-real_nodes = json.loads(_nodes_file.read_text("utf-8")) if _nodes_file.is_file() else []
-if real_nodes:
-    pp._IPV6_OK.clear()   # 清缓存，做真实探测
-    for n in real_nodes:
-        c = n.get("country") or "?"
-        io = pp._node_ipv6_only(n)
-        reach = pp._node_reachable(n)
-        mark = "[IPv6]" if reach else "[不可用：本机无 IPv6]"
-        print(f"  [{c}] {n.get('domain'):<24} ipv6_only={str(io):<6} "
-              f"可达={str(reach):<6} 菜单文案: {n.get('domain')} ({c})"
-              + (f"　{mark}" if io else ""))
-        # ipv6_only 节点在本机无 IPv6 时必须判定为不可达；有 IPv6 时仍取
-        # 实际 TCP 探测结果（本机有 IPv6 不代表每个远端节点都必然在线）。
-        if io:
-            expect = "实际探测" if has_v6 else False
-            ok = isinstance(reach, bool) if has_v6 else reach is False
-            ipv6_failed += 0 if ok else 1
-            print(f"        IPv6 专用节点可达性: {reach} (期望 {expect}) "
-                  f"{'PASS' if ok else 'FAIL'}")
-        else:
-            ok = reach is True
-            ipv6_failed += 0 if ok else 1
-            print(f"        普通节点恒可达: {reach} {'PASS' if ok else 'FAIL'}")
-    # 缓存生效：再次判定不发新探测（结果一致）
-    cached = list(pp._IPV6_OK.items())
-    if cached:
-        print(f"  可达性缓存: {cached}")
-
-# 三级判定：字段 → remark 里的 "(IPv6)" 标记 → 域名解析兜底。
-# 服务端 subscription?method=rrpp_client 只吐旧扁平格式，ipv6_only 字段被丢掉，
-# 只在 remark 里留 "(IPv6)"（后端 countries_to_rrpp_client），故第 2 级是现网主路径。
-synthetic = [
-    ({"domain": "example.com"}, False, "无标记（example.com 有 A 记录）"),
-    ({"domain": "example.com", "ipv6_only": False}, False, "ipv6_only=false"),
-    ({"domain": "example.com", "ipv6_only": True}, True, "ipv6_only=true"),
-    ({"domain": "example.com", "remark": "GlobalVM 350G (IPv6)"}, True, "remark 含 (IPv6)"),
-    ({"domain": "example.com", "remark": "Azure 15G"}, False, "remark 无 IPv6 标记"),
-    ({"domain": "example.com", "remark": "IPv6 专用机"}, True, "remark 含 IPv6（无括号）"),
-    ({"domain": "example.com", "remark": "some_ipv6_thing"}, True, "remark 小写 ipv6"),
-]
-for node, expect, desc in synthetic:
-    got = pp._node_ipv6_only(node)
-    ok = got == expect
-    ipv6_failed += 0 if ok else 1
-    print(f"  {desc:<24} _node_ipv6_only={got} (期望 {expect}) {'PASS' if ok else 'FAIL'}")
-pp._IPV6_OK.clear()   # 合成用例可能污染缓存
-
-if ipv6_failed:
-    failed += ipv6_failed
-print("=" * 70)
-
-# ── CONNECT authority 解析：IPv6 字面量带方括号 ──────────────────
-print("\n" + "=" * 70)
-print("CONNECT authority 解析（IPv6 字面量剥方括号）")
-print("=" * 70)
-
-conn_cases = [
-    ("CONNECT [::1]:443 HTTP/1.1",           ("::1", 443)),
-    ("CONNECT example.com:443 HTTP/1.1",     ("example.com", 443)),
-    ("CONNECT iam.aeolianchi.me:443 HTTP/1.1", ("iam.aeolianchi.me", 443)),
-    ("CONNECT [::1] HTTP/1.1",               None),   # 无端口
-    ("CONNECT example.com HTTP/1.1",         None),   # 无端口
-    ("CONNECT [fe80::1%25eth0]:8443 HTTP/1.1", ("fe80::1%25eth0", 8443)),
-    ("CONNECT 127.0.0.1:1080 HTTP/1.1",      ("127.0.0.1", 1080)),
-]
-conn_failed = 0
-for line, expect in conn_cases:
-    got = pp._parse_connect_authority(line)
-    ok = got == expect
-    conn_failed += 0 if ok else 1
-    print(f"  {line:<45} -> {got} (期望 {expect}) {'PASS' if ok else 'FAIL'}")
-if conn_failed:
-    failed += conn_failed
-print("=" * 70)
-
-# ── 上游代理认证注入：按完整头名精确匹配 ────────────────────────
-print("\n" + "=" * 70)
-print("认证注入：仅精确的 authorization 头名阻止注入")
-print("=" * 70)
-
-def _auth_lines(data: bytes):
-    head = data.split(b"\r\n\r\n", 1)[0]
-    return [l for l in head.split(b"\r\n")[1:] if l.lower().startswith(b"authorization:")]
-
-auth_cases = [
-    # (描述, 请求, 期望)：数字=注入后 authorization 行数；"unchanged"=原样跳过
-    ("无任何认证头",      b"CONNECT iam.aeolianchi.me:443 HTTP/1.1\r\nHost: iam.aeolianchi.me\r\n\r\n", 1),
-    ("Proxy-Authorization", b"CONNECT iam.aeolianchi.me:443 HTTP/1.1\r\nProxy-Authorization: Basic xyz\r\n\r\n", 1),
-    ("已有 Authorization", b"GET http://example.com/ HTTP/1.1\r\nAuthorization: Bearer tok\r\n\r\n", "unchanged"),
-    ("小写 authorization", b"GET http://example.com/ HTTP/1.1\r\nauthorization: basic abc\r\n\r\n", "unchanged"),
-    ("X-Authorization 前缀", b"CONNECT iam.aeolianchi.me:443 HTTP/1.1\r\nX-Authorization-Go: 1\r\n\r\n", 1),
-    ("无头空块",          b"GET http://example.com/ HTTP/1.1\r\n\r\n", 1),
-]
-auth_failed = 0
-for desc, req, expect in auth_cases:
-    out = pp._inject_proxy_auth(req)
-    if expect == "unchanged":
-        ok = out == req
-        print(f"  {desc:<22} 原样跳过 {'PASS' if ok else 'FAIL'}")
-    else:
-        lines = _auth_lines(out)
-        ok = len(lines) == expect and out.split(b"\r\n", 2)[1] == lines[0]
-        # 注入位置必须紧跟请求行（上游取首个 Authorization 时读到我们的认证）
-        print(f"  {desc:<22} authorization 行数={len(lines)} (期望 {expect}) "
-              f"{'PASS' if ok else 'FAIL'}")
-    auth_failed += 0 if ok else 1
-if auth_failed:
-    failed += auth_failed
-print("=" * 70)
-
-# ── 本程序 PAC 地址判定（启动清理只认自己的残留） ────────────────
-print("\n" + "=" * 70)
-print("PAC 地址归属判定（_is_our_pac_url）")
-print("=" * 70)
-
-pac_cases = [
-    ("http://127.0.0.1:39453/proxy.pac", True),     # 本程序 PAC 服务
-    ("http://127.0.0.1:7897/pac?auth=xyz", False),  # 其他代理软件（如 Clash）
-    ("http://127.0.0.1:1080/other.pac", False),     # 路径不符
-    ("http://192.168.1.1:8080/proxy.pac", False),   # 非本机
-    ("", False),
-]
-pac_failed = 0
-for url, expect in pac_cases:
-    got = pp._is_our_pac_url(url)
-    ok = got == expect
-    pac_failed += 0 if ok else 1
-    print(f"  {url:<42} 命中={got} (期望 {expect}) {'PASS' if ok else 'FAIL'}")
-if pac_failed:
-    failed += pac_failed
-print("=" * 70)
-
-# ── 强制代理规则格式：后缀 / full / regexp + 关键域名兜底 ───────────
-print("\n" + "=" * 70)
-print("强制代理规则格式（后缀 / full / regexp / 关键域名兜底）")
-print("=" * 70)
-
-rule_failed = 0
-rule_dir = Path(tempfile.mkdtemp()) / "RRPP"
-rule_dir.mkdir(parents=True)
-(rule_dir / "pac_domains.json").write_text(json.dumps({"domains": [
-    "suffix.example", "full:only.example", "regexp:^edge-[0-9]+\\.example$",
-]}), "utf-8")
-
-# 此段隔离真实 CDN 表，确保三种语义分别受测。
-old_user_dir = pp._USER_DIR
-old_suffixes = pp._FORCE_PROXY_SUFFIXES
-old_exact = pp._FORCE_PROXY_EXACT
-old_regexes = pp._FORCE_PROXY_REGEXES
-old_loaded = pp._FORCE_PROXY_LOADED
-pp._USER_DIR = rule_dir
-pp._FORCE_PROXY_SUFFIXES = None
-pp._FORCE_PROXY_EXACT = None
-pp._FORCE_PROXY_REGEXES = None
-pp._FORCE_PROXY_LOADED = False
-pp._is_force_proxy.cache_clear()
-pp._load_force_proxy_suffixes()
-
-rule_cases = [
-    ("普通后缀匹配子域", "cdn.suffix.example", True),
-    ("full 精确匹配", "only.example", True),
-    ("full 不匹配子域", "cdn.only.example", False),
-    ("regexp 匹配", "edge-42.example", True),
-    ("regexp 不匹配", "edge-x.example", False),
-    ("关键 Google CDN 兜底", "assets.googlecdn.com", True),
-    ("无关域名", "unmatched.example", False),
-]
-for desc, host, expect in rule_cases:
-    got = pp._is_force_proxy(host)
-    ok = got == expect
-    rule_failed += 0 if ok else 1
-    print(f"  {desc:<22} {host:<30} 命中={got} (期望 {expect}) {'PASS' if ok else 'FAIL'}")
-
-pp._USER_DIR = old_user_dir
-pp._FORCE_PROXY_SUFFIXES = old_suffixes
-pp._FORCE_PROXY_EXACT = old_exact
-pp._FORCE_PROXY_REGEXES = old_regexes
-pp._FORCE_PROXY_LOADED = old_loaded
-pp._is_force_proxy.cache_clear()
-if rule_failed:
-    failed += rule_failed
-print("=" * 70)
-
-# ── 明确直连域名：优先级 / Trie / 固定代理隔离 ───────────────────
-print("\n" + "=" * 70)
-print("明确直连域名（后缀 / full / regexp / 强制代理优先 / 固定代理隔离）")
-print("=" * 70)
-
-direct_failed = 0
-direct_dir = Path(tempfile.mkdtemp()) / "RRPP"
-direct_dir.mkdir(parents=True)
-(direct_dir / "pac_direct_domains.json").write_text(json.dumps({"domains": [
-    "direct.example", "full:only-direct.example", "regexp:^edge-direct-[0-9]+\\.example$",
-    "claude.ai",
-]}), "utf-8")
-old_user_dir = pp._USER_DIR
-old_direct_trie = pp._DIRECT_SUFFIX_TRIE
-old_direct_exact = pp._DIRECT_EXACT
-old_direct_regexes = pp._DIRECT_REGEXES
-old_direct_loaded = pp._DIRECT_LOADED
-old_data_path = pp._data_path
-try:
-    pp._USER_DIR = direct_dir
-    pp._data_path = lambda name: None  # 隔离仓库内置种子，覆盖到临时用户数据。
-    pp._DIRECT_SUFFIX_TRIE = None
-    pp._DIRECT_EXACT = None
-    pp._DIRECT_REGEXES = None
-    pp._DIRECT_LOADED = False
-    pp._is_direct_domain.cache_clear()
-    pp._load_direct_domains()
-    direct_cases = [
-        ("后缀匹配子域", "cdn.direct.example", True),
-        ("full 精确匹配", "only-direct.example", True),
-        ("full 不匹配子域", "cdn.only-direct.example", False),
-        ("regexp 匹配", "edge-direct-42.example", True),
-        ("regexp 不匹配", "edge-direct-x.example", False),
-    ]
-    for desc, host, expect in direct_cases:
-        got = pp._is_direct_domain(host)
-        ok = got == expect
-        direct_failed += 0 if ok else 1
-        print(f"  {desc:<22} {host:<30} 命中={got} (期望 {expect}) {'PASS' if ok else 'FAIL'}")
-    smart_direct = pp._decide_direct("cdn.direct.example", True, True)
-    fixed_proxy = not pp._decide_direct("cdn.direct.example", True, False)
-    force_wins = not pp._decide_direct("claude.ai", True, True)
-    direct_target = pp._direct_connect_target("cdn.direct.example", 443, True, True)
-    target_keeps_host = direct_target == ("cdn.direct.example", 443)
-    print(f"  PAC 模式命中明确直连: {smart_direct} {'PASS' if smart_direct else 'FAIL'}")
-    print(f"  固定代理模式不使用明确直连表: {fixed_proxy} {'PASS' if fixed_proxy else 'FAIL'}")
-    print(f"  强制代理优先于明确直连: {force_wins} {'PASS' if force_wins else 'FAIL'}")
-    print(f"  明确直连保持原始主机名: {target_keeps_host} {'PASS' if target_keeps_host else 'FAIL'}")
-    direct_failed += 0 if smart_direct else 1
-    direct_failed += 0 if fixed_proxy else 1
-    direct_failed += 0 if force_wins else 1
-    direct_failed += 0 if target_keeps_host else 1
-    pp._pac_write("pac_direct_domains", {"domains": ["updated-direct.example"]})
-    pp._invalidate_pac_routing_cache("pac_direct_domains")
-    direct_reloaded = (not pp._is_direct_domain("direct.example")
-                       and pp._is_direct_domain("updated-direct.example"))
-    print(f"  明确直连表热更新立即生效: {direct_reloaded} "
-          f"{'PASS' if direct_reloaded else 'FAIL'}")
-    direct_failed += 0 if direct_reloaded else 1
-finally:
-    pp._USER_DIR = old_user_dir
-    pp._data_path = old_data_path
-    pp._DIRECT_SUFFIX_TRIE = old_direct_trie
-    pp._DIRECT_EXACT = old_direct_exact
-    pp._DIRECT_REGEXES = old_direct_regexes
-    pp._DIRECT_LOADED = old_direct_loaded
-    pp._is_direct_domain.cache_clear()
-if direct_failed:
-    failed += direct_failed
-print("=" * 70)
-
-# ── PAC 更新立即生效 + DNS 解析一致性 ──────────────────────────
-print("\n" + "=" * 70)
-print("PAC 热更新缓存失效 + 国内直连使用同一次 DNS 解析结果")
-print("=" * 70)
-
-hot_failed = 0
-hot_dir = Path(tempfile.mkdtemp()) / "RRPP"
-hot_dir.mkdir(parents=True)
-old_user_dir = pp._USER_DIR
-old_cn_ranges = pp._CN_RANGES
-old_cn_ranges_v6 = pp._CN_RANGES_V6
-old_cn_loaded = pp._CN_RANGES_LOADED
-old_suffixes = pp._FORCE_PROXY_SUFFIXES
-old_exact = pp._FORCE_PROXY_EXACT
-old_regexes = pp._FORCE_PROXY_REGEXES
-old_force_loaded = pp._FORCE_PROXY_LOADED
-old_getaddrinfo = socket.getaddrinfo
-try:
-    pp._USER_DIR = hot_dir
-    pp._CN_RANGES = None
-    pp._CN_RANGES_V6 = None
-    pp._CN_RANGES_LOADED = False
-    pp._FORCE_PROXY_SUFFIXES = None
-    pp._FORCE_PROXY_EXACT = None
-    pp._FORCE_PROXY_REGEXES = None
-    pp._FORCE_PROXY_LOADED = False
-    pp._is_china_ip.cache_clear()
-    pp._is_force_proxy.cache_clear()
-
-    pp._pac_write("pac_domains", {"domains": ["old-rule.example"]})
-    pp._pac_write("pac_ip_cn", {"cidrs": ["114.114.114.0/24"]})
-    pp._load_force_proxy_suffixes()
-    pp._load_cn_ranges()
-    before = pp._is_force_proxy("old-rule.example")
-    pp._pac_write("pac_domains", {"domains": ["new-rule.example"]})
-    pp._invalidate_pac_routing_cache("pac_domains")
-    domains_reloaded = (not pp._is_force_proxy("old-rule.example")
-                        and pp._is_force_proxy("new-rule.example"))
-    print(f"  域名表热更新立即生效: 初始={before}, 更新后={domains_reloaded} "
-          f"{'PASS' if before and domains_reloaded else 'FAIL'}")
-    hot_failed += 0 if before and domains_reloaded else 1
-
-    initial_cn = pp._is_china_ip("114.114.114.114")
-    pp._pac_write("pac_ip_cn", {"cidrs": ["8.8.8.0/24"]})
-    pp._invalidate_pac_routing_cache("pac_ip_cn")
-    ip_reloaded = (not pp._is_china_ip("114.114.114.114")
-                   and pp._is_china_ip("8.8.8.8"))
-    print(f"  IP 段热更新立即生效: 初始={initial_cn}, 更新后={ip_reloaded} "
-          f"{'PASS' if initial_cn and ip_reloaded else 'FAIL'}")
-    hot_failed += 0 if initial_cn and ip_reloaded else 1
-
-    # 固定本次 DNS 返回值；_direct_connect_target 必须只解析一次并直连该 IP。
-    pp._pac_write("pac_ip_cn", {"cidrs": ["114.114.114.0/24"]})
-    pp._invalidate_pac_routing_cache("pac_ip_cn")
-    dns_calls = []
-    def fake_getaddrinfo(host, *args, **kwargs):
-        dns_calls.append(host)
-        if host == "mixed.example":
-            ips = ["114.114.114.114", "8.8.8.8"]
-        else:
-            ips = ["114.114.114.114"]
-        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (ip, 0)) for ip in ips]
-    socket.getaddrinfo = fake_getaddrinfo
-    target = pp._direct_connect_target("cn-cdn.example", 443, False, True)
-    stable_dns = target == ("114.114.114.114", 443) and dns_calls == ["cn-cdn.example"]
-    mixed = pp._direct_connect_target("mixed.example", 443, False, True)
-    mixed_safe = mixed is None
-    print(f"  国内 CDN 固定解析 IP 且不二次解析: {stable_dns} {'PASS' if stable_dns else 'FAIL'}")
-    print(f"  混合中国/境外 CDN 记录改走隧道: {mixed_safe} {'PASS' if mixed_safe else 'FAIL'}")
-    hot_failed += 0 if stable_dns else 1
-    hot_failed += 0 if mixed_safe else 1
-
-    # IPv6 中国 CIDR 与 IPv6 私网/回环分流。240e::/16 是测试用的公网 IPv6 段。
-    pp._pac_write("pac_ip_cn", {"cidrs": ["240e::/16"]})
-    pp._invalidate_pac_routing_cache("pac_ip_cn")
-    ipv6_cn = (pp._is_china_ip("240e:1234::1")
-               and pp._direct_connect_target("240e:1234::1", 443, False, True)
-                   == ("240e:1234::1", 443))
-    ipv6_private = all(pp._is_private_ip(ip) for ip in ("::1", "fc00::1", "fe80::1"))
-    ipv6_public = not pp._is_private_ip("2001:4860:4860::8888")
-    print(f"  IPv6 中国 CIDR 智能直连: {ipv6_cn} {'PASS' if ipv6_cn else 'FAIL'}")
-    print(f"  IPv6 回环/ULA/链路本地直连: {ipv6_private and ipv6_public} "
-          f"{'PASS' if ipv6_private and ipv6_public else 'FAIL'}")
-    hot_failed += 0 if ipv6_cn else 1
-    hot_failed += 0 if ipv6_private and ipv6_public else 1
-finally:
-    socket.getaddrinfo = old_getaddrinfo
-    pp._USER_DIR = old_user_dir
-    pp._CN_RANGES = old_cn_ranges
-    pp._CN_RANGES_V6 = old_cn_ranges_v6
-    pp._CN_RANGES_LOADED = old_cn_loaded
-    pp._FORCE_PROXY_SUFFIXES = old_suffixes
-    pp._FORCE_PROXY_EXACT = old_exact
-    pp._FORCE_PROXY_REGEXES = old_regexes
-    pp._FORCE_PROXY_LOADED = old_force_loaded
-    pp._is_china_ip.cache_clear()
-    pp._is_force_proxy.cache_clear()
-if hot_failed:
-    failed += hot_failed
-print("=" * 70)
-
-# ── 智能国内直连失败回退上游隧道 ───────────────────────────────
-print("\n" + "=" * 70)
-print("智能国内直连失败：仅该路径回退一次上游隧道")
-print("=" * 70)
-
-fallback_failed = 0
-direct = ("114.114.114.114", 443)
-upstream = ("node.example", 1205)
-calls = []
-remote_obj = object()
-def fake_connect(target, timeout):
-    calls.append((target, timeout))
-    if target == direct:
-        raise OSError("mock direct connect failure")
-    if target == upstream:
-        return remote_obj
-    raise AssertionError(f"unexpected target: {target}")
-
-remote, via_tunnel = pp._connect_route(
-    direct, True, upstream[0], upstream[1], None, False, fake_connect)
-fallback_ok = (remote is remote_obj and via_tunnel is True
-               and calls == [(direct, 10), (upstream, 10)])
-calls.clear()
-try:
-    pp._connect_route(direct, False, upstream[0], upstream[1], None, False, fake_connect)
-    explicit_not_fallback = False
-except OSError:
-    explicit_not_fallback = calls == [(direct, 10)]
-print(f"  智能中国公网直连失败回退隧道: {fallback_ok} {'PASS' if fallback_ok else 'FAIL'}")
-print(f"  显式直连失败不回退: {explicit_not_fallback} "
-      f"{'PASS' if explicit_not_fallback else 'FAIL'}")
-fallback_failed += 0 if fallback_ok else 1
-fallback_failed += 0 if explicit_not_fallback else 1
-if fallback_failed:
-    failed += fallback_failed
-print("=" * 70)
-
-# ── 启动连通性必须验证境外隧道 ─────────────────────────────────
-print("\n" + "=" * 70)
-print("启动连通性：google.com 隧道失败必须阻止“开启成功”")
-print("=" * 70)
-
-class _FakeResponse:
-    def __enter__(self):
-        return self
-    def __exit__(self, *args):
-        return False
-    def read(self, size=-1):
-        return b""
-
-class _FakeOpener:
-    def __init__(self, fail_google=False):
-        self.fail_google = fail_google
-    def open(self, url, timeout):
-        if self.fail_google and "www.google.com" in url:
-            raise OSError("mock tunnel failure")
-        return _FakeResponse()
-
-
-class _CancellingResponse(_FakeResponse):
-    def __init__(self, cancel_event):
-        self.cancel_event = cancel_event
-    def read(self, size=-1):
-        self.cancel_event.set()
-        return b""
-
-
-class _CancellingOpener:
-    def __init__(self, cancel_event):
-        self.cancel_event = cancel_event
-        self.urls = []
-    def open(self, url, timeout):
-        self.urls.append(url)
-        return _CancellingResponse(self.cancel_event)
-
-
-class _GoogleCancellingOpener(_CancellingOpener):
-    def open(self, url, timeout):
-        self.urls.append(url)
-        if "www.google.com" in url:
-            return _CancellingResponse(self.cancel_event)
-        return _FakeResponse()
-
-verify_failed = 0
-ok_results = pp._verify_proxy_connectivity(opener=_FakeOpener())
-ok = ok_results.get("google.com") is True
-try:
-    pp._verify_proxy_connectivity(opener=_FakeOpener(fail_google=True))
-    blocked = False
-except RuntimeError:
-    blocked = True
-cancel_event = threading.Event()
-cancelling_opener = _CancellingOpener(cancel_event)
-try:
-    pp._verify_proxy_connectivity(opener=cancelling_opener, cancel_event=cancel_event)
-    cancel_blocked = False
-except pp._StartCancelled:
-    cancel_blocked = True
-google_not_started = all("www.google.com" not in url for url in cancelling_opener.urls)
-google_cancel_event = threading.Event()
-google_cancelling_opener = _GoogleCancellingOpener(google_cancel_event)
-try:
-    pp._verify_proxy_connectivity(opener=google_cancelling_opener,
-                                  cancel_event=google_cancel_event)
-    google_cancelled = False
-except pp._StartCancelled:
-    google_cancelled = True
-google_started = any("www.google.com" in url for url in google_cancelling_opener.urls)
-print(f"  境外隧道成功可通过: {ok} {'PASS' if ok else 'FAIL'}")
-print(f"  境外隧道失败会阻止开启: {blocked} {'PASS' if blocked else 'FAIL'}")
-print(f"  Baidu 探测中取消会中断后续 Google 探测: {cancel_blocked and google_not_started} "
-      f"{'PASS' if cancel_blocked and google_not_started else 'FAIL'}")
-print(f"  Google 探测中取消会终止本次开启: {google_cancelled and google_started} "
-      f"{'PASS' if google_cancelled and google_started else 'FAIL'}")
-verify_failed += 0 if ok else 1
-verify_failed += 0 if blocked else 1
-verify_failed += 0 if cancel_blocked and google_not_started else 1
-verify_failed += 0 if google_cancelled and google_started else 1
-if verify_failed:
-    failed += verify_failed
-print("=" * 70)
-
-# ── DNS 分流 TTL 缓存 ────────────────────────────────────────
-print("\n" + "=" * 70)
-print("TTL 缓存：cache_clear 兼容 + 缓存命中")
-print("=" * 70)
-
-ttl_failed = 0
-assert hasattr(pp._is_china_ip, "cache_clear"), "cache_clear 缺失"
-assert hasattr(pp._is_private_ip, "cache_clear"), "cache_clear 缺失"
-assert hasattr(pp._is_whitelisted, "cache_clear"), "cache_clear 缺失"
-pp._is_china_ip.cache_clear()
-pp._is_private_ip.cache_clear()
-pp._is_whitelisted.cache_clear()
-a1, a2 = pp._is_private_ip("10.1.2.3"), pp._is_private_ip("10.1.2.3")
-ok = a1 is True and a2 is True
-ttl_failed += 0 if ok else 1
-print(f"  内网 IP 判定缓存命中: {a1}/{a2} {'PASS' if ok else 'FAIL'}")
-b1, b2 = pp._is_china_ip("114.114.114.114"), pp._is_china_ip("114.114.114.114")
-ok = b1 is True and b2 is True
-ttl_failed += 0 if ok else 1
-print(f"  中国 IP 判定缓存命中: {b1}/{b2} {'PASS' if ok else 'FAIL'}")
-pp._is_china_ip.cache_clear()
-pp._is_private_ip.cache_clear()
-print(f"  cache_clear 后重查: {pp._is_china_ip('114.114.114.114')} "
-      f"{'PASS' if pp._is_china_ip('114.114.114.114') is True else 'FAIL'}")
-if ttl_failed:
-    failed += ttl_failed
-print("=" * 70)
-
-print(f"\n最终结论: {'全部 PASS' if not failed else f'{failed} 个 FAIL'}")
-sys.exit(1 if failed else 0)