54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
"""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)
|