Compare commits

..
Author SHA1 Message Date
RoyirinRosmontis 8063b6d4e8 feat(lite): 发布多跳代理跳板 9999091814 2026-09-18 14:40:44 +08:00
RoyirinRosmontis aac616eed9 fix: 修正发布版本时间戳 2026-09-17 17:07:52 +08:00
RoyirinRosmontis 9bbfe900c8 删除 lite/rrpp_client_nodes.template.json 2026-09-17 17:05:20 +08:00
RoyirinRosmontis c58ef59889 feat: 支持第三方节点订阅 2026-09-17 16:57:47 +08:00
RoyirinRosmontis 9dbaabd4a9 修复:Nuitka onefile 更新路径 2026-09-16 23:45:40 +08:00
RoyirinRosmontis 6673d28503 改进:显示更新下载进度并就地保存 2026-09-16 23:35:38 +08:00
RoyirinRosmontis 6fb2926627 清理:移除本地辅助脚本与测试工具 2026-09-16 23:20:16 +08:00
RoyirinRosmontis 02d36c72f7 重构:模块化 RRPP Lite 并更新入口 2026-09-16 23:13:28 +08:00
RoyirinRosmontis c37d8af821 新增:PAC 明确直连域名空模板 2026-09-16 21:45:48 +08:00
RoyirinRosmontis cddd3afd18 Merge remote-tracking branch 'origin/main' 2026-09-16 19:13:38 +08:00
RoyirinRosmontis 0603224de2 修复:完善 PAC 分流与启动取消 2026-09-16 19:10:12 +08:00
RoyirinRosmontis 3ce5e57dae 更新 README.md 2026-09-16 19:09:37 +08:00
RoyirinRosmontis 9376e35025 修复:PAC 规则解析与关键域名分流 2026-09-16 17:46:54 +08:00
RRRosmontis 110a526fc6 server_config: 同步伪装页为线上实际版本(md5 4eb73450)
仓库内 fake-site/index.html 为旧版(md5 e0f5cbcb),与四个生产节点实际
服务的版本不一致。已用线上(us 节点)实际抓取版本替换,恢复所有节点
2026-09-11 12:53:56 +08:00
RoyirinRosmontis 9b7b817cb7 更新 README.md 2026-09-11 02:06:47 +08:00
RoyirinRosmontis a39647d46f 更新 README.md 2026-09-11 02:06:21 +08:00
RoyirinRosmontis be02d1b17f 更新 README.md 2026-09-11 02:01:47 +08:00
RoyirinRosmontis 53c36e143f 更新 README.md 2026-09-11 02:01:30 +08:00
RoyirinRosmontis 23ba368737 修复:- 系统代理设置保存/恢复:覆盖前备份原 ProxyEnable/ProxyServer/ProxyOverride/AutoConfigURL(内存+settings.json 防崩溃丢失),关闭/退出时恢复用户原有代理而不是清空;启动残留清理只认本程序精确地址(127.0.0.1:1080 精确匹配 + PAC URL 正则),不再误删其他代理软件(如 Clash)在 127.0.0.1 上的 PAC
- CONNECT 目标解析支持 IPv6 字面量:剥掉方括号再拆 host:port,[::1]:443 不再解析成 "[::1]"(直连/白名单/内网绕过判定失效的根因)
- 更新包校验加固:release 附件按 RRPP_Lite_<tag>.exe 名字精确匹配(不再固定取第一个);下载后与 API 记录的 size 比对;发布流程附带 .sha256 边车时校验 SHA-256
- 上游代理认证注入按完整头名精确匹配:Proxy-Authorization 不再误判为已有 Authorization;注入位置紧跟请求行保证上游读到我们的认证
- 远程同步后节点按 settings 记录的域名重匹配:服务端调整节点顺序时不再静默连到别的节点;启动校验等待订阅同步完成(最多 15 秒)后再执行
- DNS 分流判定缓存加 300 秒 TTL:域名换 IP 后不再长期沿用旧的直连/隧道判断(_is_private_ip/_is_china_ip/_is_whitelisted 换 _ttl_cache,保留 cache_clear 兼容)
- is_proxy_on 状态判定同步收紧为本程序 PAC 地址(其他软件的 PAC 不再让状态灯误亮)
2026-09-11 01:56:04 +08:00
25 changed files with 3300 additions and 1275 deletions
+10
View File
@@ -1,2 +1,12 @@
__pycache__/
*.pyc
# Local development state and reproducible build outputs.
.claude/
lite/.claude/
lite/backups/
lite/build/
lite/dist/
lite/RELEASE_WORKFLOW.md
*.bak
*.bak-*
+10 -4
View File
@@ -1,11 +1,17 @@
# Royirin Rosmontis Private Proxy
Supported by Global Rosmanka Cloud Nexus Inc. —— https://rosmanka.rrrosmontis.icu
## 警告
- 包含 WireGuard 的 common 版本转发器永久不再维护,推荐使用 Lite 版本
> common 版本现在有许多未修复的 bug (如部分 UI 的显示问题) 、令人难以忍受的代码逻辑 (如 PAC 规则直接写在代码里)和安全性问题 (认证凭据的暴露与证书忽略),而 Lite 版本修复了这些问题,并会收到持续的更新
- 本程序完全没有 Linux 与 CLI 功能兼容,默认构建为带有图形界面的 Windows .exe
- 本程序完全没有 Linux 与 CLI 功能兼容,默认构建为带有图形界面的 Windows 应用 (.exe)
## 如何区分 common 版本与 Lite 版本
### 如何区分 common 版本与 Lite 版本
- Lite 版本的版本号前四位固定为 9999,后续为从月份开始的构建时间戳,而 common 版本的版本号为包含年份的完整的构建时间戳
## 服务器侧配置手册 (部分内容过时)
https://rrrosmontis.icu/articles/index.html?id=00005
- https://rrrosmontis.icu/articles/index.html?id=00005
- 更推荐参考本仓库 server_config 文件夹中的 md 文件手册
## 若要直接使用仓库客户端对应的服务
请前往 https://sso.rrrosmontis.icu 注册 Rosmanka ID,然后与管理员商定使用细则
royirin@rrrosmontis.icu
+85
View File
@@ -0,0 +1,85 @@
# -*- mode: python ; coding: utf-8 -*-
from pathlib import Path
from PyInstaller.utils.hooks import collect_all
# 强制收集 PyQt5 全部模块及二进制文件
qt_datas, qt_binaries, qt_hiddenimports = collect_all('PyQt5')
# 只内置 config.json(普通设置)。nodes.json / pac_*.json 属用户数据:
# 运行时读 %LOCALAPPDATA%\RRPP(首启自动创建模板 / 从 CDN 拉取),不进 EXE。
_JSON_DATAS = [(str(Path(SPECPATH) / 'config.json'), '.')]
print(f"[spec] 打包 json 数据文件: {[Path(src).name for src, _ in _JSON_DATAS]}")
# 自定义 manifest:普通权限运行 + 声明 PerMonitorV2 DPI 感知,
# 避免程序在不同缩放比例的显示器之间移动时被 Windows 位图拉伸。
_MANIFEST = r'''<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<assembly xmlns="urn:schemas-microsoft-com:asm.v1" manifestVersion="1.0">
<assemblyIdentity version="1.0.0.0" name="RRPP_Lite" type="win32"/>
<description>Royirin Rosmontis Private Proxy Lite</description>
<trustInfo xmlns="urn:schemas-microsoft-com:asm.v2">
<security>
<requestedPrivileges>
<requestedExecutionLevel level="asInvoker" uiAccess="false"/>
</requestedPrivileges>
</security>
</trustInfo>
<compatibility xmlns="urn:schemas-microsoft-com:compatibility.v1">
<application>
<supportedOS Id="{8e0f7a12-bfb3-4fe8-b9a5-48fd50a15a9a}"/>
<supportedOS Id="{1f676c76-80e1-4239-95bb-83d0f6d0da78}"/>
</application>
</compatibility>
<application xmlns="urn:schemas-microsoft-com:asm.v3">
<windowsSettings>
<dpiAwareness xmlns="http://schemas.microsoft.com/SMI/2016/WindowsSettings">PerMonitorV2, PerMonitor, system</dpiAwareness>
<dpiAware xmlns="http://schemas.microsoft.com/SMI/2005/WindowsSettings">true/pm</dpiAware>
</windowsSettings>
</application>
<dependency>
<dependentAssembly>
<assemblyIdentity type="win32" name="Microsoft.Windows.Common-Controls" version="6.0.0.0" processorArchitecture="*" publicKeyToken="6595b64144ccf1df" language="*"/>
</dependentAssembly>
</dependency>
</assembly>
'''
a = Analysis(
['main.py'],
pathex=['..\\.venv\\Lib\\site-packages'],
binaries=qt_binaries,
datas=[('ico.ico', '.')] + _JSON_DATAS + qt_datas,
hiddenimports=qt_hiddenimports,
hookspath=[],
hooksconfig={},
runtime_hooks=[],
# 排除 Nuitka 构建链相关模块(venv 里装了 Nuitka,但 PyInstaller 产物不需要)
excludes=['nuitka', 'zstandard', 'ordered_set'],
noarchive=False,
optimize=0,
)
pyz = PYZ(a.pure)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.datas,
[],
name='Royirin Rosmontis Private Proxy Lite Python',
debug=False,
bootloader_ignore_signals=False,
strip=False,
# UPX 会压缩资源段(含图标),冷缓存下 Explorer 可能提取不到图标导致任务栏空白
upx=False,
upx_exclude=[],
runtime_tmpdir=None,
console=False,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
icon=['ico.ico'],
manifest=_MANIFEST,
)
+19
View File
@@ -0,0 +1,19 @@
@echo off
cd /d "%~dp0"
call ..\.venv\Scripts\python.exe -m PyInstaller --version >nul 2>&1
if errorlevel 1 (
echo PyInstaller not installed. Run first:
echo ..\.venv\Scripts\python.exe -m pip install pyinstaller
pause
exit /b 1
)
call ..\.venv\Scripts\python.exe -m PyInstaller RRPP_Lite.spec --noconfirm
if errorlevel 1 (
echo.
echo Build failed.
pause
exit /b 1
)
echo.
echo Build complete. Output: dist\Royirin Rosmontis Private Proxy Lite Python.exe
pause
+42
View File
@@ -0,0 +1,42 @@
@echo off
cd /d "%~dp0"
call ..\.venv\Scripts\python.exe -m nuitka --version >nul 2>&1
if errorlevel 1 (
echo Nuitka not installed. Run first:
echo ..\.venv\Scripts\python.exe -m pip install nuitka zstandard
pause
exit /b 1
)
rem Remove prior artifacts and the SCons incremental state before every build.
rem The constants blob is embedded outside normal dependency tracking, so reuse
rem can leave new UI strings and settings keys out of the executable.
rd /s /q "dist\main.build" 2>nul
rd /s /q "dist\main.dist" 2>nul
rd /s /q "dist\main.onefile-build" 2>nul
rem Use Nuitka COFF objects for embedded constants to avoid compiler cache reuse.
rem Zig C23 embed caches do not reliably track embedded blob changes.
set NUITKA_RESOURCE_MODE=coff_obj
rem Convert config.json version to Windows file-version segments (each <= 65535).
for /f %%v in ('..\.venv\Scripts\python.exe -c "import json;v=json.load(open('config.json',encoding='utf-8'))['version'];print(f'9999.{int(v[4:8])}.{int(v[8:12])}.0')"') do set "FILEVER=%%v"
call ..\.venv\Scripts\python.exe -m nuitka ^
--standalone --onefile ^
--enable-plugin=pyqt5 ^
--include-data-files=config.json=config.json ^
--windows-icon-from-ico=ico.ico ^
--windows-console-mode=disable ^
--company-name="Global Rosmanka Cloud Nexus Inc" ^
--product-name="R.R.P.P." ^
--file-version=%FILEVER% ^
--output-dir=dist ^
--output-filename="Royirin Rosmontis Private Proxy Lite.exe" ^
--assume-yes-for-downloads ^
main.py
if errorlevel 1 (
echo.
echo Build failed.
pause
exit /b 1
)
echo.
echo Build complete. Output: dist\Royirin Rosmontis Private Proxy Lite.exe
pause
+4 -4
View File
@@ -1,6 +1,6 @@
{
"version": "",
"update_url": "",
"pac_base_url": "",
"nodes_url": ""
"version": "999909181435",
"pac_base_url": "https://cdn.rrrosmontis.icu/RRPP/PAC/",
"nodes_url": "https://api.rrrosmontis.icu/v1/rrpp/subscription?method=rrpp_client",
"update_url": "https://git.rrrosmontis.icu/api/v1/repos/RoyirinRosmontis/Private-Proxy/releases/latest"
}
BIN
View File
Binary file not shown.

After

Width:  |  Height:  |  Size: 121 KiB

File diff suppressed because it is too large Load Diff
+6
View File
@@ -0,0 +1,6 @@
{
"version": "",
"date": "",
"description": "PAC 智能分流的明确直连域名模板。强制代理规则和用户白名单优先级更高。",
"domains": []
}
+1
View File
@@ -0,0 +1 @@
"""RRPP Lite 的无 GUI 核心逻辑。"""
+91
View File
@@ -0,0 +1,91 @@
"""本地转发器的无状态报文解析与改写函数。"""
from urllib.parse import urlsplit
HTTP_METHODS = (
b"CONNECT ", b"GET ", b"POST ", b"PUT ", b"DELETE ",
b"HEAD ", b"OPTIONS ", b"PATCH ",
)
def starts_with_http_method(data: bytes) -> bool:
return any(data.startswith(method) for method in HTTP_METHODS)
def inject_proxy_auth(data: bytes, basic_auth: str) -> bytes:
"""为完整 HTTP 代理请求注入上游 Basic Authorization。"""
if not starts_with_http_method(data):
return data
end = data.find(b"\r\n\r\n")
if end == -1:
return data
head = data[:end]
first_end = head.find(b"\r\n")
request_line = head if first_end == -1 else head[:first_end]
headers = b"" if first_end == -1 else head[first_end + 2:]
for line in headers.split(b"\r\n"):
if b":" not in line:
continue
name = line[:line.index(b":")].strip().lower()
if name == b"authorization":
return data
auth_line = f"Authorization: Basic {basic_auth}\r\n".encode()
return request_line + b"\r\n" + auth_line + headers + data[end:]
def parse_connect_authority(first_line: str):
"""从 CONNECT 首行提取 ``(host, port)``,失败时返回 ``None``。"""
parts = first_line.split(" ")
if len(parts) < 2:
return None
authority = parts[1]
if authority.startswith("[") and "]" in authority:
try:
host = authority[1:authority.index("]")]
tail = authority[authority.index("]") + 1:]
if tail.startswith(":") and tail[1:].isdigit():
return host, int(tail[1:])
except ValueError:
pass
if ":" not in authority:
return None
host, _, port_s = authority.rpartition(":")
if not host or not port_s.isdigit():
return None
return host, int(port_s)
def parse_http_absolute(first_line: str):
"""从明文 HTTP 代理首行提取 ``(host, port)``,失败时返回 ``None``。"""
parts = first_line.split(" ")
if len(parts) < 2 or not parts[1].lower().startswith(("http://", "https://")):
return None
parsed = urlsplit(parts[1])
if not parsed.hostname:
return None
port = parsed.port or (443 if parsed.scheme == "https" else 80)
return parsed.hostname, port
def force_http_close(buffer: bytes) -> bytes:
"""直连明文 HTTP 时强制 Connection: close,避免错复用连接。"""
end = buffer.find(b"\r\n\r\n")
if end == -1:
return buffer
head, tail = buffer[:end], buffer[end:]
lines = head.split(b"\r\n")
out = [lines[0]]
conn_seen = False
for line in lines[1:]:
low = line.lower()
if low.startswith(b"connection:"):
out.append(b"Connection: close")
conn_seen = True
elif low.startswith(b"proxy-connection:"):
continue
else:
out.append(line)
if not conn_seen:
out.append(b"Connection: close")
return b"\r\n".join(out) + tail
+82
View File
@@ -0,0 +1,82 @@
"""路由、白名单和转发器共用的网络基础函数。"""
import functools
import ipaddress
import socket
import threading
import time
def ttl_cache(seconds: int, maxsize: int = 16384):
"""带 TTL 的小型线程安全缓存,并保留 ``cache_clear`` 兼容接口。"""
def decorator(function):
store = {}
lock = threading.Lock()
@functools.wraps(function)
def wrapper(*args, **kwargs):
key = (args, tuple(sorted(kwargs.items())))
now = time.time()
with lock:
hit = store.get(key)
if hit and now - hit[0] < seconds:
return hit[1]
value = function(*args, **kwargs)
with lock:
store[key] = (now, value)
if len(store) > maxsize:
cutoff = now - seconds
for cached_key, (timestamp, _) in list(store.items()):
if timestamp < cutoff:
del store[cached_key]
if len(store) > maxsize:
store.clear()
return value
def cache_clear():
with lock:
store.clear()
wrapper.cache_clear = cache_clear
return wrapper
return decorator
def is_private_ipv4_int(value: int) -> bool:
"""整数形式 IPv4 是否属于常见私网、回环或链路本地段。"""
return (
(value & 0xFF000000) == 0x0A000000
or (value & 0xFFF00000) == 0xAC100000
or (value & 0xFFFF0000) == 0xC0A80000
or (value & 0xFF000000) == 0x7F000000
or (value & 0xFFFF0000) == 0xA9FE0000
)
def resolve_host_ips(host: str):
"""解析主机名的 IPv4/IPv6 集合;字面量 IP 返回自身,失败返回空集合。"""
try:
return {ipaddress.ip_address(host)}
except ValueError:
pass
try:
return {
ipaddress.ip_address(address[4][0])
for address in socket.getaddrinfo(host, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
}
except Exception:
return set()
def is_private_host(host: str) -> bool:
"""目标是否命中私网、回环或 IPv6 ULA/链路本地地址。"""
for address in resolve_host_ips(host):
if address.version == 4 and is_private_ipv4_int(int(address)):
return True
if address.version == 6 and (
address.is_loopback
or address.is_link_local
or address in ipaddress.ip_network("fc00::/7")
):
return True
return False
+36
View File
@@ -0,0 +1,36 @@
"""节点订阅的认证请求协议。"""
import base64
import json
import urllib.error
import urllib.request
class NodeAuthError(Exception):
"""订阅服务拒绝请求时携带的 HTTP 状态和用户可读原因。"""
def __init__(self, message, code=0):
super().__init__(message)
self.code = code
def b64url(value: str) -> str:
"""SSO 查询参数使用的无 padding URL-safe Base64。"""
return base64.urlsafe_b64encode(str(value).encode("utf-8")).decode("ascii").rstrip("=")
def fetch_nodes_auth(url, user, password, timeout=10):
"""直连拉取需 SSO 查询参数认证的 nodes.json。"""
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
separator = "&" if "?" in url else "?"
request_url = f"{url}{separator}user={b64url(user)}&password={b64url(password)}"
try:
with opener.open(request_url, timeout=timeout) as response:
return json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as error:
body = ""
try:
body = error.read().decode("utf-8", "replace").strip()
except Exception:
pass
raise NodeAuthError(body or f"请求被拒绝 (HTTP {error.code})", error.code) from None
+95
View File
@@ -0,0 +1,95 @@
"""节点与登录凭据的本机加密存储。"""
import ctypes
import json
import os
class _DpapiBlob(ctypes.Structure):
_fields_ = [
("cbData", ctypes.c_uint32),
("pbData", ctypes.POINTER(ctypes.c_char)),
]
class NodeSecretsStore:
"""使用 Windows DPAPI 保存节点和 SSO 凭据;非 Windows 保持明文兼容。"""
def __init__(self, user_dir_getter):
self._user_dir_getter = user_dir_getter
@staticmethod
def protect_bytes(data: bytes) -> bytes:
crypt32 = ctypes.WinDLL("crypt32")
kernel32 = ctypes.WinDLL("kernel32")
crypt32.CryptProtectData.argtypes = [
ctypes.POINTER(_DpapiBlob), ctypes.c_wchar_p, ctypes.POINTER(_DpapiBlob),
ctypes.c_void_p, ctypes.c_void_p, ctypes.c_uint32, ctypes.POINTER(_DpapiBlob),
]
crypt32.CryptProtectData.restype = ctypes.c_bool
kernel32.LocalFree.argtypes = [ctypes.c_void_p]
buffer = ctypes.create_string_buffer(data, len(data))
blob_in = _DpapiBlob(len(data), ctypes.cast(buffer, ctypes.POINTER(ctypes.c_char)))
blob_out = _DpapiBlob()
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)
@staticmethod
def unprotect_bytes(data: bytes) -> bytes:
crypt32 = ctypes.WinDLL("crypt32")
kernel32 = ctypes.WinDLL("kernel32")
crypt32.CryptUnprotectData.argtypes = [
ctypes.POINTER(_DpapiBlob), ctypes.POINTER(ctypes.c_wchar_p), ctypes.POINTER(_DpapiBlob),
ctypes.c_void_p, ctypes.c_void_p, ctypes.c_uint32, ctypes.POINTER(_DpapiBlob),
]
crypt32.CryptUnprotectData.restype = ctypes.c_bool
kernel32.LocalFree.argtypes = [ctypes.c_void_p]
buffer = ctypes.create_string_buffer(data, len(data))
blob_in = _DpapiBlob(len(data), ctypes.cast(buffer, ctypes.POINTER(ctypes.c_char)))
description = ctypes.c_wchar_p()
blob_out = _DpapiBlob()
if not crypt32.CryptUnprotectData(ctypes.byref(blob_in), ctypes.byref(description), 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 description.value:
kernel32.LocalFree(description)
def encrypt_text(self, text: str) -> bytes:
if os.name != "nt":
return text.encode("utf-8")
return self.protect_bytes(text.encode("utf-8"))
def decrypt_bytes(self, blob: bytes) -> str:
if os.name != "nt":
return blob.decode("utf-8")
try:
return self.unprotect_bytes(blob).decode("utf-8")
except Exception:
return blob.decode("utf-8")
def save_auth(self, username, password):
directory = self._user_dir_getter()
directory.mkdir(parents=True, exist_ok=True)
text = json.dumps({"username": username, "password": password}, ensure_ascii=False)
(directory / "auth.json").write_bytes(self.encrypt_text(text))
def load_auth(self):
try:
path = self._user_dir_getter() / "auth.json"
if not path.is_file():
return None
data = json.loads(self.decrypt_bytes(path.read_bytes()))
if not (isinstance(data, dict) and data.get("username") and data.get("password")):
return None
return data
except Exception:
return None
+53
View File
@@ -0,0 +1,53 @@
"""PAC 数据文件的网络传输与安全落盘。"""
import json
import os
import urllib.request
from pathlib import Path
def read_version(path):
"""读取 PAC JSON 的 version/date;无效文件返回空字符串。"""
try:
data = json.loads(Path(path).read_text("utf-8"))
return str(data.get("version") or ""), str(data.get("date") or "")
except Exception:
return "", ""
def no_proxy_opener():
"""创建忽略环境/系统代理的直连 HTTP opener。"""
return urllib.request.build_opener(urllib.request.ProxyHandler({}))
def fetch_json(url, timeout=15, tries=2):
"""以直连方式拉取 JSON;最后一次失败时保留原始异常。"""
opener = no_proxy_opener()
for attempt in range(tries):
try:
with opener.open(url, timeout=timeout) as response:
return json.loads(response.read().decode("utf-8"))
except Exception:
if attempt == tries - 1:
raise
def validate_document(name, data):
"""校验各 PAC 文件的最小结构,避免错误响应覆盖有效缓存。"""
if not isinstance(data, dict):
return False
if name in ("pac_domains", "pac_direct_domains"):
return isinstance(data.get("domains"), list)
if name == "pac_ip_cn":
return isinstance(data.get("cidrs"), list)
return False
def atomic_write(directory, name, data):
"""将 PAC JSON 以 tmp + replace 原子写入指定用户目录。"""
directory = Path(directory)
directory.mkdir(parents=True, exist_ok=True)
target = directory / f"{name}.json"
temporary_path = directory / f"{name}.json.tmp"
temporary_path.write_text(json.dumps(data, ensure_ascii=False), "utf-8")
os.replace(temporary_path, target)
+81
View File
@@ -0,0 +1,81 @@
"""PAC 路由引擎使用的纯算法,不依赖文件、网络或 Qt。"""
import ipaddress
import re
def merge_ip_ranges(cidrs, version):
"""将指定 IP 版本的 CIDR 合并为整数闭区间,供二分查找。"""
ranges = []
for cidr in cidrs:
network = ipaddress.ip_network(str(cidr).strip(), strict=False)
if network.version == version:
ranges.append((int(network.network_address), int(network.broadcast_address)))
ranges.sort()
merged = []
for start, end in ranges:
if merged and start <= merged[-1][1] + 1:
if end > merged[-1][1]:
merged[-1] = (merged[-1][0], end)
else:
merged.append((start, end))
return merged
def build_suffix_trie(suffixes):
"""按反向域名标签建立 Trie,适合大规模后缀规则的 O(标签数) 匹配。"""
root = {}
for suffix in suffixes:
node = root
for label in reversed(suffix.split(".")):
node = node.setdefault(label, {})
node[""] = True
return root
def suffix_trie_matches(host: str, trie) -> bool:
"""判断主机名是否命中一个后缀 Trie。"""
node = trie
for label in reversed(host.rstrip(".").lower().split(".")):
node = node.get(label)
if node is None:
return False
if node.get(""):
return True
return bool(node and node.get(""))
def parse_domain_rules(raw_rules, seed_suffixes=()):
"""解析后缀、``full:`` 与 ``regexp:`` 三种 PAC 域名规则。
返回 ``(suffixes, exact, regexes, invalid_regex_count)``。调用方决定
这些规则是“强制代理”还是“明确直连”,从而避免解析层携带业务策略。
"""
suffixes = set(seed_suffixes)
exact = set()
regexes = []
invalid_regexes = 0
for raw in raw_rules or ():
raw_rule = str(raw).strip()
rule = raw_rule.lower()
if not raw_rule:
continue
if rule.startswith("full:"):
host = rule[len("full:"):].rstrip(".")
if host:
exact.add(host)
continue
if rule.startswith("regexp:"):
pattern = raw_rule[len("regexp:"):]
if not pattern:
invalid_regexes += 1
continue
try:
regexes.append(re.compile(pattern, re.IGNORECASE))
except re.error:
invalid_regexes += 1
continue
suffix = rule.lstrip("*.").rstrip(".")
if suffix:
suffixes.add(suffix)
return suffixes, frozenset(exact), tuple(regexes), invalid_regexes
+34
View File
@@ -0,0 +1,34 @@
"""用户设置的 JSON 存储实现。"""
import json
import os
class JsonSettingsStore:
"""读取与原子写回一个 JSON 对象。
``path_getter`` 在每次操作时调用,保证测试或运行时切换用户目录后不会
意外写到旧目录。
"""
def __init__(self, path_getter):
self._path_getter = path_getter
def load(self):
path = self._path_getter()
try:
if not path.is_file():
return {}
data = json.loads(path.read_text("utf-8"))
return data if isinstance(data, dict) else {}
except Exception:
return {}
def save(self, data):
path = self._path_getter()
path.parent.mkdir(parents=True, exist_ok=True)
temporary_path = path.with_suffix(path.suffix + ".tmp")
temporary_path.write_text(
json.dumps(data, ensure_ascii=False, indent=2) + "\n", "utf-8"
)
os.replace(temporary_path, path)
@@ -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 "已隐藏订阅地址"
+55
View File
@@ -0,0 +1,55 @@
"""线程安全的本地转发流量统计。"""
import threading
import time
class TrafficStats:
"""记录单次代理会话的上下行总量与采样速率。"""
def __init__(self, clock=time.time):
self._clock = clock
self._lock = threading.Lock()
self._up = 0
self._down = 0
self._prev_up = 0
self._prev_down = 0
self._prev_ts = 0.0
def add(self, up=0, down=0):
with self._lock:
self._up += up
self._down += down
def reset(self):
with self._lock:
self._up = self._down = 0
self._prev_up = self._prev_down = 0
self._prev_ts = self._clock()
def rate(self):
with self._lock:
now = self._clock()
if self._prev_ts > 0 and now > self._prev_ts:
elapsed = now - self._prev_ts
up_rate = int((self._up - self._prev_up) / elapsed)
down_rate = int((self._down - self._prev_down) / elapsed)
else:
up_rate = down_rate = 0
self._prev_up = self._up
self._prev_down = self._down
self._prev_ts = now
return up_rate, down_rate, self._up, self._down
def snapshot(self):
with self._lock:
return self._up, self._down
def format_bytes(value):
"""将字节数格式化为 UI 展示文本。"""
number = float(value)
for unit in ("B", "KB", "MB", "GB", "TB"):
if number < 1024 or unit == "TB":
return f"{int(number)} {unit}" if unit == "B" else f"{number:.1f} {unit}"
number /= 1024
+4
View File
@@ -0,0 +1,4 @@
"""RRPP Lite 的 Qt 界面层。
核心代理与 PAC 逻辑不得依赖本包,以便后续在无 GUI 环境中独立测试。
"""
+878
View File
@@ -0,0 +1,878 @@
"""需要业务服务的 Qt 对话框。"""
import json
import threading
from dataclasses import dataclass
from PyQt5.QtCore import Qt, pyqtSignal
from PyQt5.QtWidgets import (
QDialog, QHBoxLayout, QLabel, QLineEdit, QListWidget, QListWidgetItem,
QPushButton, QVBoxLayout, QWidget,
)
from .theme import C_BG, C_BORDER, C_CARD, C_GREEN, C_RED, C_SUBTEXT, C_TEXT, C_TITLE
from .widgets import NodeLabel, ToggleSwitch
@dataclass(frozen=True)
class LoginServices:
"""登录框所需的显式业务依赖,避免 UI 反向导入程序入口。"""
load_icon: object
load_auth: object
get_nodes_url: object
fetch_nodes: object
user_dir: object
encrypt_nodes: object
save_auth: object
auth_error: type
log: object
class LoginDialog(QDialog):
"""使用 SSO 获取并加密保存节点配置的登录框。"""
_done = pyqtSignal(bool, str)
def __init__(self, services: LoginServices, parent=None):
super().__init__(parent)
self._services = services
self.setWindowTitle("节点配置登录")
self.setWindowIcon(services.load_icon())
self.setFixedSize(420, 170)
self.setStyleSheet(f"""
QDialog {{ background-color: {C_BG}; }}
QLabel {{ color: {C_TEXT}; font-family: "Microsoft YaHei", "Segoe UI", sans-serif; font-size: 12px; }}
QLineEdit {{
background-color: {C_CARD}; border: 1px solid {C_BORDER}; border-radius: 6px;
padding: 6px 8px; color: {C_TEXT}; font-size: 12px;
}}
QPushButton#btnOk {{
background-color: {C_GREEN}; color: #fff; border: none; border-radius: 6px;
padding: 4px 12px; font-size: 12px; font-weight: bold;
}}
QPushButton#btnOk:hover {{ background-color: #5ed978; }}
QPushButton#btnOk:disabled {{ background-color: #2a3a2e; color: #666; }}
QPushButton#btnCancel {{
background-color: {C_CARD}; color: {C_SUBTEXT}; border: 1px solid {C_BORDER};
border-radius: 6px; padding: 4px 12px; font-size: 12px;
}}
""")
root = QHBoxLayout(self)
root.setSpacing(14)
root.setContentsMargins(20, 18, 20, 16)
left = QVBoxLayout()
left.setSpacing(6)
left.addStretch()
logo = QLabel()
logo.setPixmap(services.load_icon().pixmap(80, 80))
logo.setFixedSize(80, 80)
logo.setAlignment(Qt.AlignCenter)
left.addWidget(logo, 0, Qt.AlignHCenter)
brand = QLabel("R.R.P.P.")
brand.setStyleSheet(f"color: {C_TITLE}; font-size: 10px;")
brand.setAlignment(Qt.AlignCenter)
left.addWidget(brand, 0, Qt.AlignHCenter)
left.addStretch()
root.addLayout(left)
form = QVBoxLayout()
form.setSpacing(6)
self.ed_user = QLineEdit()
self.ed_user.setPlaceholderText("SSO 邮箱")
form.addWidget(self.ed_user)
self.ed_pass = QLineEdit()
self.ed_pass.setPlaceholderText("密码")
self.ed_pass.setEchoMode(QLineEdit.Password)
self.ed_pass.returnPressed.connect(self._submit)
form.addWidget(self.ed_pass)
credentials = services.load_auth()
if credentials:
self.ed_user.setText(credentials["username"])
self.ed_pass.setText(credentials["password"])
self.lb_err = QLabel("")
self.lb_err.setStyleSheet(f"color: {C_RED}; font-size: 11px;")
self.lb_err.setFixedHeight(14)
self.lb_err.hide()
form.addWidget(self.lb_err)
bottom = QHBoxLayout()
bottom.setSpacing(10)
hint = QLabel("登录以获取节点配置")
hint.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 10px;")
bottom.addWidget(hint)
bottom.addStretch()
self.btn_cancel = QPushButton("跳过登录")
self.btn_cancel.setObjectName("btnCancel")
self.btn_cancel.setToolTip("可稍后在设置中登录,或添加第三方订阅")
self.btn_cancel.clicked.connect(self.reject)
bottom.addWidget(self.btn_cancel)
self.btn_ok = QPushButton("登录")
self.btn_ok.setObjectName("btnOk")
self.btn_ok.clicked.connect(self._submit)
bottom.addWidget(self.btn_ok)
form.addLayout(bottom)
box = QWidget()
box.setLayout(form)
root.addWidget(box, 1, Qt.AlignVCenter)
self._done.connect(self._on_done)
def _submit(self):
user = self.ed_user.text().strip()
password = self.ed_pass.text()
if not user or not password:
self.lb_err.setText("请输入用户名和密码")
self.lb_err.show()
return
self.btn_ok.setEnabled(False)
self.btn_cancel.setEnabled(False)
self.btn_ok.setText("登录中")
threading.Thread(target=self._worker, args=(user, password), daemon=True).start()
def _worker(self, user, password):
try:
url = str(self._services.get_nodes_url() or "").strip()
if not url:
raise RuntimeError("config.json 未配置 nodes_url")
data = self._services.fetch_nodes(url, user, password)
if not (isinstance(data, list) and data and all(isinstance(node, dict) for node in data)
and str(data[0].get("domain") or "").strip()):
raise ValueError("返回的节点数据无效")
text = json.dumps(data, ensure_ascii=False, indent=2) + "\n"
directory = self._services.user_dir()
directory.mkdir(parents=True, exist_ok=True)
(directory / "nodes.json").write_bytes(self._services.encrypt_nodes(text))
self._services.save_auth(user, password)
self._done.emit(True, str(len(data)))
except self._services.auth_error as error:
self._done.emit(False, str(error))
except Exception as error:
self._done.emit(False, f"登录失败: {error}")
def _on_done(self, ok, info):
self.btn_ok.setEnabled(True)
self.btn_cancel.setEnabled(True)
self.btn_ok.setText("登录")
if ok:
self._services.log(f"登录成功,已获取 {info} 个节点(nodes.json 已加密保存)")
self.accept()
else:
self.lb_err.setText(info)
self.lb_err.show()
@dataclass(frozen=True)
class WhitelistServices:
"""白名单编辑框所需的规则服务。"""
load_icon: object
load_rules: object
validate_rule: object
save_rules: object
class WhitelistDialog(QDialog):
"""编辑强制直连白名单;规则数据通过显式服务读写。"""
def __init__(self, services: WhitelistServices, parent=None):
super().__init__(parent)
self._services = services
self.setWindowTitle("代理白名单")
self.setWindowIcon(services.load_icon())
self.setFixedSize(380, 400)
self.setStyleSheet(f"""
QDialog {{ background-color: {C_BG}; }}
QLabel {{ color: {C_TEXT}; font-family: "Microsoft YaHei", "Segoe UI", sans-serif; font-size: 12px; }}
QLineEdit {{ background-color: {C_CARD}; border: 1px solid {C_BORDER}; border-radius: 6px;
padding: 6px 8px; color: {C_TEXT}; font-size: 12px; }}
QListWidget {{ background-color: {C_CARD}; border: 1px solid {C_BORDER}; border-radius: 6px;
padding: 4px; color: {C_TEXT}; font-size: 12px; }}
QListWidget::item {{ padding: 3px 6px; border-radius: 4px; }}
QListWidget::item:selected {{ background-color: #e3e8ee; color: {C_TEXT}; }}
QPushButton#btnOk {{ background-color: {C_GREEN}; color: #fff; border: none;
border-radius: 6px; padding: 4px 12px; font-size: 12px; font-weight: bold; }}
QPushButton#btnOk:hover {{ background-color: #5ed978; }}
QPushButton#btnCancel {{ background-color: {C_CARD}; color: {C_SUBTEXT}; border: 1px solid {C_BORDER};
border-radius: 6px; padding: 4px 12px; font-size: 12px; }}
QPushButton#btnSmall {{ background-color: {C_CARD}; color: {C_SUBTEXT}; border: 1px solid {C_BORDER};
border-radius: 6px; padding: 3px 10px; font-size: 11px; }}
QPushButton#btnSmall:hover {{ background-color: #e3e8ee; color: {C_TEXT}; }}
""")
layout = QVBoxLayout(self)
layout.setSpacing(10)
layout.setContentsMargins(20, 16, 20, 16)
title = QLabel("代理白名单")
title.setStyleSheet(f"color: {C_TITLE}; font-size: 15px; font-weight: bold;")
title.setAlignment(Qt.AlignCenter)
layout.addWidget(title)
hint = QLabel("命中以下规则的连接将强制直连(不经过代理)")
hint.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 11px;")
hint.setAlignment(Qt.AlignCenter)
layout.addWidget(hint)
self.lb_err = QLabel("")
self.lb_err.setStyleSheet(f"color: {C_RED}; font-size: 11px;")
self.lb_err.setFixedHeight(14)
self.lb_err.hide()
layout.addWidget(self.lb_err)
self.list_widget = QListWidget()
self.list_widget.setToolTip("勾选启用规则;可多选后删除")
for rule in services.load_rules():
self._add_item(rule["value"], rule.get("enabled", True))
layout.addWidget(self.list_widget, 1)
add_row = QHBoxLayout()
add_row.setSpacing(8)
self.ed_add = QLineEdit()
self.ed_add.setPlaceholderText("输入域名 / IP / CIDR,如 example.com")
self.ed_add.returnPressed.connect(self._add_rule)
add_row.addWidget(self.ed_add, 1)
self.btn_add = QPushButton("添加")
self.btn_add.setObjectName("btnOk")
self.btn_add.setStyleSheet(f"""
QPushButton {{ background-color: {C_GREEN}; color: #fff; border: none; border-radius: 6px;
padding: 6px 14px; font-size: 12px; font-weight: bold; }}
QPushButton:hover {{ background-color: #5ed978; }}
""")
self.btn_add.clicked.connect(self._add_rule)
add_row.addWidget(self.btn_add)
layout.addLayout(add_row)
middle = QHBoxLayout()
self.btn_delete = QPushButton("删除选中")
self.btn_delete.setObjectName("btnSmall")
self.btn_delete.clicked.connect(self._delete_selected)
middle.addWidget(self.btn_delete)
middle.addStretch()
layout.addLayout(middle)
buttons = QHBoxLayout()
buttons.setSpacing(10)
buttons.addStretch()
self.btn_cancel = QPushButton("取消")
self.btn_cancel.setObjectName("btnCancel")
self.btn_cancel.clicked.connect(self.reject)
buttons.addWidget(self.btn_cancel)
self.btn_ok = QPushButton("保存")
self.btn_ok.setObjectName("btnOk")
self.btn_ok.clicked.connect(self._save)
buttons.addWidget(self.btn_ok)
layout.addLayout(buttons)
def _add_item(self, value, enabled):
item = QListWidgetItem(value)
item.setFlags(item.flags() | Qt.ItemIsUserCheckable)
item.setCheckState(Qt.Checked if enabled else Qt.Unchecked)
self.list_widget.addItem(item)
def _add_rule(self):
value, error = self._services.validate_rule(self.ed_add.text())
if error:
self.lb_err.setText(error)
self.lb_err.show()
return
for index in range(self.list_widget.count()):
if self.list_widget.item(index).text().lower() == value.lower():
self.lb_err.setText("该规则已存在")
self.lb_err.show()
return
self.lb_err.hide()
self._add_item(value, True)
self.ed_add.clear()
def _delete_selected(self):
for item in self.list_widget.selectedItems():
self.list_widget.takeItem(self.list_widget.row(item))
def _save(self):
rules = []
for index in range(self.list_widget.count()):
item = self.list_widget.item(index)
value = str(item.text()).strip()
if value:
rules.append({"value": value, "enabled": item.checkState() == Qt.Checked})
self._services.save_rules(rules)
self.accept()
@dataclass(frozen=True)
class JumpServices:
"""代理跳板管理所需的节点摘要与配置回调。"""
load_icon: object
load_profiles: object
available_nodes: object
add_profile: object
remove_profile: object
class JumpDialog(QDialog):
"""配置最多三跳的节点链路;设置中只保存节点稳定标识。"""
def __init__(self, services: JumpServices, parent=None):
super().__init__(parent)
self._services = services
self.setWindowTitle("代理跳板")
self.setWindowIcon(services.load_icon())
self.setFixedSize(560, 520)
self.setStyleSheet(f"""
QDialog {{ background-color: {C_BG}; }}
QLabel {{ color: {C_TEXT}; font-family: "Microsoft YaHei", "Segoe UI", sans-serif; 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(8)
layout.setContentsMargins(18, 14, 18, 14)
title = QLabel("代理跳板")
title.setStyleSheet(f"color: {C_TITLE}; font-size: 15px; font-weight: bold;")
title.setAlignment(Qt.AlignCenter)
layout.addWidget(title)
hint = QLabel("按顺序加入节点,至少 2 个、最多 3 个;最后一个节点是出口")
hint.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 11px;")
hint.setAlignment(Qt.AlignCenter)
layout.addWidget(hint)
saved_label = QLabel("已保存跳板")
saved_label.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 11px;")
layout.addWidget(saved_label)
self.profiles = QListWidget()
self.profiles.setFixedHeight(68)
self.profiles.setToolTip("选择已保存跳板后可删除;新链路在下方配置")
layout.addWidget(self.profiles)
columns = QHBoxLayout()
columns.setSpacing(8)
left = QVBoxLayout()
left.addWidget(QLabel("当前可用节点"))
self.available = QListWidget()
self.available.setToolTip("选择节点后点击“加入链路”")
left.addWidget(self.available, 1)
columns.addLayout(left, 1)
middle = QVBoxLayout()
middle.addStretch()
self.btn_add_node = QPushButton("加入链路 →")
self.btn_add_node.setObjectName("btnSmall")
self.btn_add_node.clicked.connect(self._add_node)
middle.addWidget(self.btn_add_node)
self.btn_remove_node = QPushButton("← 移除")
self.btn_remove_node.setObjectName("btnSmall")
self.btn_remove_node.clicked.connect(self._remove_node)
middle.addWidget(self.btn_remove_node)
middle.addStretch()
columns.addLayout(middle)
right = QVBoxLayout()
right.addWidget(QLabel("新跳板链路"))
self.route = QListWidget()
self.route.setToolTip("顺序即连接顺序;可用上下按钮调整")
right.addWidget(self.route, 1)
reorder = QHBoxLayout()
self.btn_up = QPushButton("上移")
self.btn_up.setObjectName("btnSmall")
self.btn_up.clicked.connect(lambda: self._move_node(-1))
reorder.addWidget(self.btn_up)
self.btn_down = QPushButton("下移")
self.btn_down.setObjectName("btnSmall")
self.btn_down.clicked.connect(lambda: self._move_node(1))
reorder.addWidget(self.btn_down)
right.addLayout(reorder)
columns.addLayout(right, 1)
layout.addLayout(columns, 1)
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_profile = QPushButton("添加")
self.btn_add_profile.setObjectName("btnOk")
self.btn_add_profile.clicked.connect(self._add_profile)
actions.addWidget(self.btn_add_profile)
self.btn_delete_profile = QPushButton("删除选中跳板")
self.btn_delete_profile.setObjectName("btnDanger")
self.btn_delete_profile.clicked.connect(self._delete_profile)
actions.addWidget(self.btn_delete_profile)
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._reload()
def _reload(self):
self.available.clear()
for node in self._services.available_nodes():
item = QListWidgetItem(node["label"])
item.setData(Qt.UserRole, node["ref"])
self.available.addItem(item)
self.profiles.clear()
for profile in self._services.load_profiles():
state = "可用" if profile.get("available") else "节点不可用"
item = QListWidgetItem(f"{profile['name']} · {profile['route']} · {state}")
item.setData(Qt.UserRole, profile["id"])
self.profiles.addItem(item)
self.route.clear()
self.lb_status.setText("")
def _add_node(self):
item = self.available.currentItem()
if item is None:
self._status("请先选择一个节点", error=True)
return
ref = item.data(Qt.UserRole)
refs = {self.route.item(i).data(Qt.UserRole) for i in range(self.route.count())}
if ref in refs:
self._status("同一条链路不能重复使用节点", error=True)
return
if self.route.count() >= 3:
self._status("最多只能配置 3 个节点", error=True)
return
copy = QListWidgetItem(item.text())
copy.setData(Qt.UserRole, ref)
self.route.addItem(copy)
self._status(f"当前链路:{self.route.count()} 个节点")
def _remove_node(self):
item = self.route.currentItem()
if item is not None:
self.route.takeItem(self.route.row(item))
def _move_node(self, offset):
row = self.route.currentRow()
new_row = row + offset
if row < 0 or not 0 <= new_row < self.route.count():
return
item = self.route.takeItem(row)
self.route.insertItem(new_row, item)
self.route.setCurrentRow(new_row)
def _add_profile(self):
if not 2 <= self.route.count() <= 3:
self._status("请先配置 2-3 个节点的链路", error=True)
return
refs = [self.route.item(i).data(Qt.UserRole) for i in range(self.route.count())]
try:
self._services.add_profile(refs)
except Exception as error:
self._status(str(error), error=True)
return
self._reload()
self._status("跳板已添加")
def _delete_profile(self):
item = self.profiles.currentItem()
if item is None:
self._status("请先选择要删除的跳板", error=True)
return
try:
self._services.remove_profile(item.data(Qt.UserRole))
except Exception as error:
self._status(str(error), error=True)
return
self._reload()
self._status("跳板已删除")
def _status(self, message, error=False):
self.lb_status.setText(message)
color = C_RED if error else C_GREEN
self.lb_status.setStyleSheet(f"color: {color}; font-size: 11px;")
@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:
"""设置框所需的配置与业务回调。"""
load_icon: object
version: str
get_port: object
get_update_via_proxy: object
get_skip_update_check: object
parse_port: object
set_port: object
set_update_via_proxy: object
set_skip_update_check: object
open_whitelist: object
open_jump: object
open_third_party: object
has_login: object
open_login: object
logout: object
class SettingsDialog(QDialog):
"""代理端口、更新方式、白名单与登录状态设置框。"""
def __init__(self, services: SettingsServices, parent=None, port_locked=False):
super().__init__(parent)
self._services = services
self.setWindowTitle("设置")
self.setWindowIcon(services.load_icon())
self.setFixedSize(300, 430)
self.setStyleSheet(f"""
QDialog {{ background-color: {C_BG}; }}
QLabel {{ color: {C_TEXT}; font-family: "Microsoft YaHei", "Segoe UI", sans-serif; font-size: 12px; }}
QLineEdit {{ background-color: {C_CARD}; border: 1px solid {C_BORDER}; border-radius: 6px;
padding: 2px 8px; color: {C_TEXT}; font-size: 12px; }}
""")
layout = QVBoxLayout(self)
layout.setSpacing(12)
layout.setContentsMargins(20, 16, 20, 16)
title = QLabel("设置")
title.setStyleSheet(f"color: {C_TITLE}; font-size: 15px; font-weight: bold;")
title.setAlignment(Qt.AlignCenter)
layout.addWidget(title)
version_row = QHBoxLayout()
version_row.addStretch()
version_label = QLabel(f"版本号: {services.version}")
version_label.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 11px;")
version_row.addWidget(version_label)
version_row.addSpacing(4)
self.link_check_update = NodeLabel(
"检查更新", tooltip="手动检查更新",
base=f"color: {C_SUBTEXT}; font-size: 11px;",
hover=f"color: {C_TEXT}; font-size: 11px; text-decoration: underline;",
)
self.link_check_update.clicked.connect(self._manual_check_update)
version_row.addWidget(self.link_check_update)
version_row.addStretch()
layout.addLayout(version_row)
self.lb_hint = QLabel("代理运行中,关闭代理后可修改" if port_locked else "保存更改后下次开启代理生效")
self.lb_hint.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 11px;")
self.lb_hint.setFixedHeight(14)
self.lb_hint.setAlignment(Qt.AlignCenter)
layout.addWidget(self.lb_hint)
layout.addStretch()
whitelist_row = QHBoxLayout()
whitelist_label = QLabel("代理白名单")
whitelist_label.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;")
whitelist_label.setToolTip("命中列表的域名 / IP 将强制直连(不经过代理),每条规则可单独启用")
whitelist_row.addWidget(whitelist_label)
whitelist_row.addStretch()
self.btn_whitelist = QPushButton("配置")
self.btn_whitelist.setCursor(Qt.PointingHandCursor)
self.btn_whitelist.setStyleSheet(f"""
QPushButton {{ background-color: {C_CARD}; color: {C_SUBTEXT}; border: 1px solid {C_BORDER};
border-radius: 6px; padding: 2px 10px; font-size: 11px; }}
QPushButton:hover {{ background-color: #e3e8ee; color: {C_TEXT}; }}
""")
self.btn_whitelist.clicked.connect(self._open_whitelist)
whitelist_row.addWidget(self.btn_whitelist)
layout.addLayout(whitelist_row)
jump_row = QHBoxLayout()
jump_label = QLabel("代理跳板")
jump_label.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;")
jump_label.setToolTip("把当前可用节点串成 2-3 跳链路;最后一个节点作为对外出口")
jump_row.addWidget(jump_label)
jump_row.addStretch()
self.btn_jump = QPushButton("添加")
self.btn_jump.setCursor(Qt.PointingHandCursor)
self.btn_jump.setStyleSheet(self.btn_whitelist.styleSheet())
self.btn_jump.clicked.connect(self._open_jump)
jump_row.addWidget(self.btn_jump)
layout.addLayout(jump_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;")
port_row.addWidget(port_label)
port_row.addStretch()
self.ed_port = QLineEdit(str(services.get_port()))
self.ed_port.setFixedSize(54, 22)
self.ed_port.setEnabled(not port_locked)
port_row.addWidget(self.ed_port)
layout.addLayout(port_row)
update_row = QHBoxLayout()
update_label = QLabel("使用代理更新")
update_label.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;")
update_row.addWidget(update_label)
update_row.addStretch()
self.cb_update_proxy = ToggleSwitch(checked=services.get_update_via_proxy())
self.cb_update_proxy.setToolTip("开启后检查更新与下载走节点隧道(CONNECT);默认直连仓库")
update_row.addWidget(self.cb_update_proxy)
layout.addLayout(update_row)
skip_row = QHBoxLayout()
skip_label = QLabel("跳过检查更新")
skip_label.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;")
skip_row.addWidget(skip_label)
skip_row.addStretch()
self.cb_skip_update = ToggleSwitch(checked=services.get_skip_update_check())
self.cb_skip_update.setToolTip("跳过 PAC 数据与程序版本检查(节点更新与登录不受影响),被跳过的检查不输出日志")
skip_row.addWidget(self.cb_skip_update)
layout.addLayout(skip_row)
layout.addStretch()
buttons = QHBoxLayout()
buttons.setSpacing(10)
buttons.addStretch()
self.btn_save = QPushButton("保存更改")
self.btn_save.setStyleSheet(f"""
QPushButton {{ background-color: {C_GREEN}; color: #fff; border: none; border-radius: 6px;
padding: 5px 14px; font-size: 12px; font-weight: bold; }}
QPushButton:disabled {{ background-color: #2a3a2e; color: #666; }}
""")
self.btn_save.setEnabled(not port_locked)
self.btn_save.clicked.connect(self._save_settings)
buttons.addWidget(self.btn_save)
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.clicked.connect(self._logout)
buttons.addWidget(self.btn_logout)
buttons.addStretch()
layout.addLayout(buttons)
def _manual_check_update(self):
parent = self.parent()
if not (hasattr(parent, "_check_update") and hasattr(parent, "_update_done")):
return
self.lb_hint.setText("正在检查更新...")
self.lb_hint.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 11px;")
try:
parent._update_done.connect(self._on_check_done)
except TypeError:
pass
threading.Thread(target=parent._check_update, kwargs={"manual": True}, daemon=True).start()
def _on_check_done(self):
self.lb_hint.setText("检查完成,结果见主界面日志")
self.lb_hint.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 11px;")
try:
self.parent()._update_done.disconnect(self._on_check_done)
except Exception:
pass
def _open_whitelist(self):
self._services.open_whitelist(self)
def _open_jump(self):
self._services.open_jump(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:
self.lb_hint.setText("端口应为 1-65535 的数字")
self.lb_hint.setStyleSheet(f"color: {C_RED}; font-size: 11px;")
return
self._services.set_port(port)
self._services.set_update_via_proxy(bool(self.cb_update_proxy.isChecked()))
self._services.set_skip_update_check(bool(self.cb_skip_update.isChecked()))
self.ed_port.setText(str(port))
self.lb_hint.setText("已保存,下次开启代理生效")
self.lb_hint.setStyleSheet(f"color: {C_GREEN}; font-size: 11px;")
def _logout(self):
if self._services.has_login():
self._services.logout()
self.btn_logout.setText("登录")
else:
self._services.open_login(self)
if self._services.has_login():
self.btn_logout.setText("退出登录")
+11
View File
@@ -0,0 +1,11 @@
"""RRPP Lite 界面的共享色彩常量。"""
C_TITLE = "#8899aa"
C_GREEN = "#51cf66"
C_YELLOW = "#ffd43b"
C_RED = "#ff6b6b"
C_BG = "#ffffff"
C_CARD = "#f0f2f5"
C_BORDER = "#d0d5dd"
C_TEXT = "#333333"
C_SUBTEXT = "#889099"
+104
View File
@@ -0,0 +1,104 @@
"""无业务状态的可复用 Qt 控件。"""
from PyQt5.QtCore import Qt, QRectF, pyqtSignal
from PyQt5.QtGui import QColor, QPainter
from PyQt5.QtWidgets import QCheckBox, QLabel
from .theme import C_BORDER, C_GREEN, C_RED, C_TITLE
_NL_BASE = f"color: {C_TITLE}; font-size: 12px;"
class NodeLabel(QLabel):
"""可点击标签:hover 显示下划线和手型光标,点击发出 clicked 信号。"""
clicked = pyqtSignal()
def __init__(self, text="", tooltip="点击切换节点", base=None, hover=None):
super().__init__(text)
self._base = base or _NL_BASE
self._hover = hover or self._base + " text-decoration: underline;"
self.setStyleSheet(self._base)
if tooltip:
self.setToolTip(tooltip)
def enterEvent(self, event):
self.setStyleSheet(self._hover)
self.setCursor(Qt.PointingHandCursor)
super().enterEvent(event)
def leaveEvent(self, event):
self.setStyleSheet(self._base)
self.unsetCursor()
super().leaveEvent(event)
def mouseReleaseEvent(self, event):
if event.button() == Qt.LeftButton:
self.clicked.emit()
super().mouseReleaseEvent(event)
class ToggleSwitch(QCheckBox):
"""苹果风格开关:绿/灰胶囊加滑动圆钮,自绘且不依赖图片资源。"""
def __init__(self, checked=False, parent=None):
super().__init__(parent)
self.setChecked(checked)
self.setFixedSize(42, 22)
self.setCursor(Qt.PointingHandCursor)
def paintEvent(self, event):
painter = QPainter(self)
painter.setRenderHint(QPainter.Antialiasing)
painter.setPen(Qt.NoPen)
painter.setBrush(QColor(C_GREEN) if self.isChecked() else QColor("#c8ccd4"))
painter.drawRoundedRect(QRectF(0, 0, 42, 22), 11, 11)
knob_x = 20.0 if self.isChecked() else 2.0
painter.setBrush(QColor("#ffffff"))
painter.drawEllipse(QRectF(knob_x, 2.0, 18.0, 18.0))
painter.end()
def mousePressEvent(self, event):
if event.button() == Qt.LeftButton:
self.setChecked(not self.isChecked())
self.repaint()
event.accept()
return
super().mousePressEvent(event)
class StatusDot(QLabel):
"""代理状态小圆点。"""
def __init__(self):
super().__init__()
self.setFixedSize(14, 14)
self.set_off()
def set_on(self):
self.setStyleSheet(f"""
QLabel {{
background-color: {C_GREEN};
border-radius: 7px;
border: 1px solid rgba(81, 207, 102, 50);
}}
""")
def set_off(self):
self.setStyleSheet(f"""
QLabel {{
background-color: {C_RED};
border-radius: 7px;
border: 1px solid rgba(255, 107, 107, 50);
}}
""")
def set_disabled(self):
self.setStyleSheet(f"""
QLabel {{
background-color: {C_BORDER};
border-radius: 7px;
border: 1px solid rgba(136, 153, 170, 50);
}}
""")
-75
View File
@@ -1,75 +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 proxy_lite as pp
from PyQt5.QtWidgets import QApplication
# 构造窗口时不写用户数据目录(避免污染),构造后再指向真实 nodes.json
pp._user_init_local = lambda: None
app = QApplication(sys.argv)
w = pp.MainWindow()
w.timer.stop() # 停掉状态轮询,纯静态检查
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()} (应不变)")
# ── 上次选中的节点已不可用:启动时自动回退到第一个可达节点 ──────────
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]
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'}")
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'}")
# 再校验一次:已经是可达节点,不应再切换
ok3 = pp._verify_selected_node() is False
print(f" 再次校验不重复切换 {'PASS' if ok3 else 'FAIL'}")
if _bak is None:
(pp._USER_DIR / "settings.json").unlink(missing_ok=True)
else:
(pp._USER_DIR / "settings.json").write_bytes(_bak)
print("OK")
+2 -2
View File
@@ -26,7 +26,7 @@
<span class="code-label">Error code 200</span>
</h1>
<div>
Visit <a href="https://example.com" target="_blank" rel="noopener noreferrer">example.com</a> for more information.
Visit <a href="https://rrrosmontis.icu" target="_blank" rel="noopener noreferrer">rrrosmontis.icu</a> for more information.
</div>
<div class="mt-3">2026-06-28 05:00:19 UTC</div>
</header>
@@ -90,7 +90,7 @@
<span class="cf-footer-separator sm:hidden">&bull;</span>
</span>
<span class="cf-footer-item sm:block sm:mb-1"><span>Performance &amp; security by</span> <a rel="noopener noreferrer" href="https://example.com" id="brand_link" target="_blank">Example Site</a></span>
<span class="cf-footer-item sm:block sm:mb-1"><span>Performance &amp; security by</span> <a rel="noopener noreferrer" href="https://rrrosmontis.icu" id="brand_link" target="_blank">Royirin Rosmontis</a></span>
</p>
</div><!-- /.error-footer -->
</div>