RoyirinRosmontis/Private-Proxy
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 21 Wiki Activity

feat: 支持第三方节点订阅

Browse Source
This commit is contained in:
RoyirinRosmontis
2026-09-17 16:57:47 +08:00
parent 9dbaabd4a9
commit c58ef59889
5 changed files with 530 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lite/main.py
+263 -65
View File
@@ -1,6 +1,6 @@
# -*- 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 (
@@ -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,
@@ -110,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:
log_red("nodes.json 缺失，无法加载节点配置")
return _NODES
data = json.loads(_nodes_decrypt(Path(path).read_bytes()))
if not isinstance(data, list):
raise ValueError("nodes.json 顶层应为节点数组")
_NODES = data
if not _SELECTED_RESTORED:
_restore_selected_node(_NODES) # 恢复上次使用的节点
log(f"已加载节点配置: {len(_NODES)} 个")
if data and not str(data[0].get("domain") or "").strip():
log_red("节点未填写域名：请配置 config.json 的 nodes_url 在线更新，"
"或使用 others/nodes_tool 导入节点配置")
except Exception as e:
log_red(f"nodes.json 加载失败: {e}")
if path is not None:
data = json.loads(_nodes_decrypt(Path(path).read_bytes()))
if not isinstance(data, list):
raise ValueError("nodes.json 顶层应为节点数组")
official = _tag_official_nodes(data)
except Exception as error:
log_red(f"官方节点配置读取失败: {error}")
third_party = _subscription_nodes(_load_third_party_subscriptions())
_NODES = official + third_party
if not _SELECTED_RESTORED:
_restore_selected_node(_NODES)
if _NODES:
log(f"已加载节点配置: 官方 {len(official)} 个，第三方 {len(third_party)} 个")
return _NODES
_SELECTED_NODE = 0 # 当前选中节点下标（默认第一条）
_SELECTED_RESTORED = False # 启动时是否已按设置恢复过上次选中的节点
_NODES_SYNC_DONE = threading.Event() # 订阅同步线程完成信号（启动校验等它）
def _third_party_summaries():
"""供 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。"""
@@ -167,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 节点可达性（客户端标注用）─────────────────────
@@ -312,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)
@@ -349,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
@@ -392,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。"""
@@ -1091,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 数据（后台线程）：
@@ -1627,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:
@@ -1636,7 +1834,7 @@ def _logout_user_data():
except Exception:
pass
_NODES = None
log("已退出登录：凭据与节点配置已清除，下次启动或开启代理时需重新登录")
log("已退出登录：官方凭据与节点配置已清除，第三方订阅保持不变")
def _settings_services():
@@ -1651,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,
)
@@ -1705,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()
# 节点未配置时延迟弹出登录框（等窗口显示后）
@@ -1717,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()
@@ -1725,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):
"""初始化主界面布局与样式。"""
@@ -1759,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()
@@ -1954,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)
@@ -2093,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 dlg.exec_() == QDialog.Accepted:
_reload_nodes()
if _open_login_dialog(self):
self._refresh_subtitle()
# ── 节点切换 ──
@@ -2126,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()))
@@ -2160,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:
@@ -2189,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):
"""代理运行中时提示禁止切换节点。"""