54 lines
2.0 KiB
Python
54 lines
2.0 KiB
Python
"""第三方 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 "已隐藏订阅地址"
|