DSH Agent
9a64853db4
- 新增 §12：PAC 不能直接下发的原因、三个规则集产物、PAC 更新后的再生成流程、 mihomo/Shadowrocket 两种客户端接入方式与验证口径 - 新增 rrpp-subscription/：API.md（订阅端点/method/rules 参数/规则集说明） 与 gen_rulesets.py（PAC → 客户端规则集转换脚本，幂等、原子替换） - method 新增 shadowrocket（.conf：远程 RULE-SET + FINAL，不猜测 REALITY 参数） - method=clash 新增 &rules=provider|inline|global
218 lines
7.7 KiB
Python
218 lines
7.7 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
RRPP PAC → 规则集转换器
|
||
|
||
把 PAC 数据文件（JSON）转换成 **mihomo(Clash) 与 Shadowrocket 通用**的
|
||
"classical 规则行"文本规则集——每行都含规则类型：
|
||
|
||
pac_direct_domains.json -> rrpp-direct.list （DOMAIN-SUFFIX,<域名>）
|
||
pac_domains.json -> rrpp-proxy.list （DOMAIN-SUFFIX,<域名>）
|
||
pac_ip_cn.json -> rrpp-cn-ip.list （IP-CIDR,<网段>）
|
||
|
||
为什么用这个格式（两端通吃）：
|
||
* mihomo : rule-providers { behavior: classical, format: text }
|
||
* Shadowrocket: RULE-SET,<url>,<策略>
|
||
（官方要求"规则集的内容需包含规则类型"）
|
||
|
||
三种规则集由客户端按各自策略引用：
|
||
rrpp-proxy.list -> 代理（强制代理域名）
|
||
rrpp-direct.list -> 直连（保守直连种子）
|
||
rrpp-cn-ip.list -> 直连（中国 IP 段）
|
||
|
||
用法
|
||
----
|
||
python gen_rulesets.py # 就地生成（内容有变化才写盘）
|
||
python gen_rulesets.py --check # 只检查/报告，不写盘
|
||
python gen_rulesets.py --pac-dir DIR --out-dir DIR
|
||
python gen_rulesets.py --quiet
|
||
|
||
维护
|
||
----
|
||
PAC 更新后重新执行本脚本即可。脚本是**幂等**的：内容与现有文件一致时不会重写；
|
||
写入采用 临时文件 + 原子替换，失败不会损坏已有文件。输出不含注释，只有规则行。
|
||
"""
|
||
import argparse
|
||
import hashlib
|
||
import ipaddress
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
# ---------- 默认路径（bj 生产环境） ----------
|
||
DEFAULT_PAC_DIR = r'C:/BtSoft/nginx/others/htmlResources/cdn/RRPP/PAC'
|
||
DEFAULT_OUT_DIR = DEFAULT_PAC_DIR # 与 PAC 同目录，由 CDN 直接对外提供
|
||
|
||
# 映射：PAC 文件 -> (输出文件, JSON 中的列表字段, 类型)
|
||
# 类型 domains = 域名表（含 v2fly 前缀：full:/keyword:/regexp:/行内注释）
|
||
# 类型 cidrs = IP 网段表
|
||
RULESETS = [
|
||
('pac_domains.json', 'rrpp-proxy.list', 'domains'),
|
||
('pac_direct_domains.json', 'rrpp-direct.list', 'domains'),
|
||
('pac_ip_cn.json', 'rrpp-cn-ip.list', 'cidrs'),
|
||
]
|
||
|
||
|
||
def log(msg, quiet=False):
|
||
if not quiet:
|
||
print(msg, flush=True)
|
||
|
||
|
||
def read_json(path):
|
||
"""读取 JSON；容忍 BOM。"""
|
||
with open(path, 'r', encoding='utf-8-sig') as f:
|
||
return json.load(f)
|
||
|
||
|
||
def _valid_domain(s):
|
||
if not s or len(s) > 253 or '.' not in s:
|
||
return False
|
||
if any(c in s for c in ' /\\:@?#[]'):
|
||
return False
|
||
return True
|
||
|
||
|
||
def convert_domain_entry(v):
|
||
"""把一条域名表条目转成规则行。
|
||
|
||
支持 v2fly geosite 前缀与行内注释（PAC 数据实测存在）：
|
||
full:<域名> -> DOMAIN,<域名> （精确匹配）
|
||
keyword:<关键词> -> DOMAIN-KEYWORD,<关键词>
|
||
regexp:<正则> -> 跳过（两端都无法表达域名正则）
|
||
<域名> # 注释 -> DOMAIN-SUFFIX,<域名>
|
||
返回 (规则行 或 None, 跳过原因 或 None)。
|
||
"""
|
||
s = str(v or '')
|
||
if '#' in s: # 行内注释（如 "apple.xn--czr694b # apple.商标"）
|
||
s = s.split('#', 1)[0]
|
||
s = s.strip().lower()
|
||
if not s:
|
||
return None, 'empty'
|
||
|
||
if s.startswith('regexp:'):
|
||
return None, 'regexp'
|
||
if s.startswith('include:'):
|
||
return None, 'include'
|
||
|
||
if s.startswith('full:'):
|
||
dom = s[5:].strip().lstrip('*.').rstrip('.')
|
||
return (f'DOMAIN,{dom}', None) if _valid_domain(dom) else (None, 'invalid')
|
||
|
||
if s.startswith('keyword:'):
|
||
kw = s[8:].strip()
|
||
return (f'DOMAIN-KEYWORD,{kw}', None) if kw else (None, 'invalid')
|
||
|
||
if s.startswith('domain:'):
|
||
s = s[7:].strip()
|
||
|
||
dom = s.lstrip('*.').rstrip('.') # 通配前缀统一按后缀匹配
|
||
return (f'DOMAIN-SUFFIX,{dom}', None) if _valid_domain(dom) else (None, 'invalid')
|
||
|
||
|
||
def norm_cidr(v):
|
||
"""CIDR 规范化/校验：返回规范化网段，非法返回 None。"""
|
||
s = str(v or '').strip()
|
||
if not s:
|
||
return None
|
||
try:
|
||
net = ipaddress.ip_network(s, strict=False)
|
||
except ValueError:
|
||
return None
|
||
return str(net.network_address) + '/' + str(net.prefixlen)
|
||
|
||
|
||
def build_lines(kind, data, quiet=False):
|
||
"""由 PAC JSON 生成规则行（排序去重），返回 (lines, stats)。"""
|
||
field = 'domains' if kind == 'domains' else 'cidrs'
|
||
raw = data.get(field) if isinstance(data, dict) else None
|
||
if not isinstance(raw, list):
|
||
raise ValueError(f'字段 {field} 不是列表')
|
||
seen, stats, out = set(), {}, []
|
||
for item in raw:
|
||
if kind == 'domains':
|
||
line, why = convert_domain_entry(item)
|
||
else:
|
||
c = norm_cidr(item)
|
||
line, why = (f'IP-CIDR,{c}', None) if c else (None, 'invalid')
|
||
if line is None:
|
||
stats[why] = stats.get(why, 0) + 1
|
||
continue
|
||
if line not in seen:
|
||
seen.add(line)
|
||
out.append(line)
|
||
out.sort()
|
||
if stats:
|
||
detail = ' '.join(f'{k}={v}' for k, v in sorted(stats.items()))
|
||
log(f' ! 跳过条目: {detail}', quiet)
|
||
return out, stats
|
||
|
||
|
||
|
||
def write_if_changed(path, text, check_only=False):
|
||
"""内容有变化才原子写入。返回 (changed, md5)。"""
|
||
data = text.encode('utf-8')
|
||
md5 = hashlib.md5(data).hexdigest()
|
||
old = None
|
||
if os.path.exists(path):
|
||
with open(path, 'rb') as f:
|
||
old = f.read()
|
||
if old == data:
|
||
return False, md5
|
||
if check_only:
|
||
return True, md5
|
||
tmp = path + '.tmp'
|
||
with open(tmp, 'wb') as f: # 二进制写：确保 LF、无 BOM
|
||
f.write(data)
|
||
os.replace(tmp, path)
|
||
return True, md5
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description='RRPP PAC -> 规则集转换器')
|
||
ap.add_argument('--pac-dir', default=DEFAULT_PAC_DIR, help='PAC JSON 所在目录')
|
||
ap.add_argument('--out-dir', default=DEFAULT_OUT_DIR, help='规则集输出目录')
|
||
ap.add_argument('--check', action='store_true', help='只检查，不写盘')
|
||
ap.add_argument('--quiet', action='store_true')
|
||
args = ap.parse_args()
|
||
|
||
if not os.path.isdir(args.pac_dir):
|
||
print(f'错误：PAC 目录不存在: {args.pac_dir}', file=sys.stderr)
|
||
return 2
|
||
os.makedirs(args.out_dir, exist_ok=True)
|
||
|
||
total, changed_cnt, failed = 0, 0, 0
|
||
log(f'PAC 目录 : {args.pac_dir}')
|
||
log(f'输出目录 : {args.out_dir}{" (--check 模式，不写盘)" if args.check else ""}')
|
||
log('')
|
||
|
||
for src_name, out_name, kind in RULESETS:
|
||
src_path = os.path.join(args.pac_dir, src_name)
|
||
out_path = os.path.join(args.out_dir, out_name)
|
||
if not os.path.exists(src_path):
|
||
print(f' [跳过] {src_name} 不存在', file=sys.stderr)
|
||
failed += 1
|
||
continue
|
||
try:
|
||
data = read_json(src_path)
|
||
lines, _stats = build_lines(kind, data, args.quiet)
|
||
text = '\n'.join(lines) + '\n'
|
||
changed, md5 = write_if_changed(out_path, text, args.check)
|
||
size = len(text.encode('utf-8'))
|
||
ver = data.get('version', '-') if isinstance(data, dict) else '-'
|
||
state = ('已更新' if changed else '无变化') if not args.check else ('需更新' if changed else '无变化')
|
||
log(f' {src_name:26s} -> {out_name:18s} {len(lines):5d} 行 {size/1024:7.1f} KB '
|
||
f'ver={ver:12s} md5={md5[:8]} [{state}]')
|
||
total += len(lines)
|
||
changed_cnt += 1 if changed else 0
|
||
except Exception as e:
|
||
print(f' [失败] {src_name}: {type(e).__name__}: {e}', file=sys.stderr)
|
||
failed += 1
|
||
|
||
log('')
|
||
log(f'合计 {total} 条规则，{changed_cnt} 个文件{"需更新" if args.check else "已更新"}，{failed} 个失败')
|
||
return 1 if failed else 0
|
||
|
||
|
||
if __name__ == '__main__':
|
||
sys.exit(main())
|