#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 规则库构建脚本 把 blackmatrix7 上游规则 +你的私有定制 合成一套自建规则，输出到 rule-vault/ 上游更新后重跑本脚本即可同步（建议配 cron 每日自动更新） 设计要点： - Clash 版用 .yaml 分离载荷（behavior: domain / ipcidr），与 mihomo 匹配 - Loon 版用 .list 混合格式 - 私有定制（法院隔离网 / Tailscale / 自建域名）单独成一个高优先级规则集 """ import os import sys import urllib.request import ssl import time import concurrent.futures from datetime import datetime, timezone, timedelta OUT = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'rule-vault') CDN = 'https://cdn.jsdelivr.net/gh/blackmatrix7/ios_rule_script@master/rule' B = 'https://cdn.jsdelivr.net/gh/blackmatrix7/ios_rule_script@master/rule' CST = timezone(timedelta(hours=8)) ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE # ---------- 要同步的上游规则集 ---------- # key: (客户端, 上游相对路径, 输出文件名, 行为类型说明) UPSTREAM = [ # ---- 广告（Clash 用 AdvertisingLite，误杀少）---- ('Clash', 'Clash/AdvertisingLite/AdvertisingLite.yaml', 'AdvertisingLite.yaml', 'classical'), ('Loon', 'Loon/AdvertisingLite/AdvertisingLite.list', 'AdvertisingLite.list', 'classical'), # ---- 国内平台（强制直连，防境外IP 静默降级）---- ('Clash', 'Clash/DouYin/DouYin.list', 'DouYin.list', 'classical'), ('Clash', 'Clash/XiaoHongShu/XiaoHongShu.list', 'XiaoHongShu.list', 'classical'), ('Clash', 'Clash/WeChat/WeChat.list', 'WeChat.list', 'classical'), ('Clash', 'Clash/DingTalk/DingTalk.list', 'DingTalk.list', 'classical'), ('Loon', 'Loon/DouYin/DouYin.list', 'DouYin.list', 'classical'), ('Loon', 'Loon/XiaoHongShu/XiaoHongShu.list', 'XiaoHongShu.list', 'classical'), ('Loon', 'Loon/WeChat/WeChat.list', 'WeChat.list', 'classical'), ('Loon', 'Loon/DingTalk/DingTalk.list', 'DingTalk.list', 'classical'), # ---- AI 服务 ---- ('Clash', 'Clash/OpenAI/OpenAI.list', 'OpenAI.list', 'classical'), ('Clash', 'Clash/Claude/Claude.list', 'Claude.list', 'classical'), ('Clash', 'Clash/Gemini/Gemini.list', 'Gemini.list', 'classical'), ('Loon', 'Loon/OpenAI/OpenAI.list', 'OpenAI.list', 'classical'), ('Loon', 'Loon/Claude/Claude.list', 'Claude.list', 'classical'), ('Loon', 'Loon/Gemini/Gemini.list', 'Gemini.list', 'classical'), # ---- 开发工具 ---- ('Clash', 'Clash/GitHub/GitHub.list', 'GitHub.list', 'classical'), ('Clash', 'Clash/Notion/Notion.list', 'Notion.list', 'classical'), ('Clash', 'Clash/Docker/Docker.list', 'Docker.list', 'classical'), ('Loon', 'Loon/GitHub/GitHub.list', 'GitHub.list', 'classical'), ('Loon', 'Loon/Notion/Notion.list', 'Notion.list', 'classical'), ('Loon', 'Loon/Docker/Docker.list', 'Docker.list', 'classical'), # ---- 流媒体 ---- ('Clash', 'Clash/YouTube/YouTube.list', 'YouTube.list', 'classical'), ('Clash', 'Clash/Netflix/Netflix.list', 'Netflix.list', 'classical'), ('Clash', 'Clash/Disney/Disney.list', 'Disney.list', 'classical'), ('Clash', 'Clash/Spotify/Spotify.list', 'Spotify.list', 'classical'), ('Clash', 'Clash/Telegram/Telegram.list', 'Telegram.list', 'classical'), ('Clash', 'Clash/Google/Google.list', 'Google.list', 'classical'), ('Clash', 'Clash/Apple/Apple.list', 'Apple.list', 'classical'), ('Loon', 'Loon/YouTube/YouTube.list', 'YouTube.list', 'classical'), ('Loon', 'Loon/Netflix/Netflix.list', 'Netflix.list', 'classical'), ('Loon', 'Loon/Disney/Disney.list', 'Disney.list', 'classical'), ('Loon', 'Loon/Spotify/Spotify.list', 'Spotify.list', 'classical'), ('Loon', 'Loon/Telegram/Telegram.list', 'Telegram.list', 'classical'), ('Loon', 'Loon/Google/Google.list', 'Google.list', 'classical'), ('Loon', 'Loon/Apple/Apple.list', 'Apple.list', 'classical'), # ---- 私有网络 ---- ('Clash', 'Clash/Lan/Lan.list', 'Lan.list', 'classical'), ('Loon', 'Loon/Lan/Lan.list', 'Lan.list', 'classical'), # ---- 境外兜底 ---- ('Clash', 'Clash/Proxy/Proxy.list', 'Proxy.list', 'classical'), ('Loon', 'Loon/Proxy/Proxy.list', 'Proxy.list', 'classical'), # ---- 中国大陆（分离载荷，关键：behavior 必须配对）---- ('Clash', 'Clash/ChinaMax/ChinaMax_Domain.yaml', 'ChinaMax_Domain.yaml', 'domain'), ('Clash', 'Clash/ChinaMax/ChinaMax_IP.yaml', 'ChinaMax_IP.yaml', 'ipcidr'), ('Loon', 'Loon/ChinaMax/ChinaMax.list', 'ChinaMax.list', 'classical'), ] def fetch(url, timeout=90, retries=4): """带指数退避的重试。CDN 偶发 SSL EOF，重试即可成功。""" last = None for attempt in range(retries): try: req = urllib.request.Request(url, headers={'User-Agent': 'curl/8'}) with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r: return r.read() except Exception as e: last = e if attempt < retries - 1: time.sleep(2 * (attempt + 1)) raise last def strip_comments(text): """ 去掉注释行与上游的 payload: 声明行，只保留裸规则值。 上游 Clash/*.yaml 本身已是「payload: + 裸值」结构， 直接取payload 下的值即可，不能把 payload: 当成一条规则写进去。 """ body = text.replace('\r\n', '\n').replace('\r', '\n') lines = [] in_payload = False for line in body.splitlines(): s = line.strip() if not s or s.startswith('#'): continue if s == 'payload:': in_payload = True continue # 兼容上游写成 "payload:\n - xxx" 或直接 "- xxx" if s.startswith('- '): lines.append(s[2:].strip()) else: lines.append(s) return lines def quote(v): """值若已带引号则原样输出，避免出现 ''x'' 这种双引号。""" v = v.strip() if (v.startswith("'") and v.endswith("'")) or (v.startswith('"') and v.endswith('"')): return v if ':' in v or v.startswith('*') or v.startswith('&') or v.startswith('#'): return f"'{v}'" return v def build_one(client, src, dst_name, behavior, force=False): src_url = f'{B}/{src}' dst = os.path.join(OUT, client, dst_name) os.makedirs(os.path.dirname(dst), exist_ok=True) try: data = fetch(src_url).decode('utf-8', errors='ignore') except Exception as e: return (client, dst_name, False, f'下载失败: {e}') lines = strip_comments(data) stamp = datetime.now(CST).strftime('%Y-%m-%d %H:%M:%S') header = [ f'# 自建规则库 | 来源: blackmatrix7/ios_rule_script', f'# 上游: {src_url}', f'# 同步时间: {stamp} (UTC+8)', f'# behavior: {behavior}', f'# 规则数: {len(lines)}', ] if dst_name.endswith('.yaml'): # Clash yaml 载荷格式：裸值列表 + payload: header.append('payload:') body = [' - ' + quote(v) for v in lines] else: body = lines content = '\n'.join(header + body) + '\n' with open(dst, 'w', encoding='utf-8', newline='\n') as f: f.write(content) return (client, dst_name, True, f'{len(lines)} 条') def main(): print(f'规则库同步开始 {datetime.now(CST):%Y-%m-%d %H:%M:%S}') results = [] with concurrent.futures.ThreadPoolExecutor(8) as ex: futs = [ex.submit(build_one, *a) for a in UPSTREAM] for f in concurrent.futures.as_completed(futs): results.append(f.result()) results.sort(key=lambda x: (x[0], x[1])) ok = sum(1 for r in results if r[2]) print(f'\n{"客户端":<8}{"输出文件":<26}{"结果"}') print('-' * 60) for client, name, good, msg in results: print(f'{client:<8}{name:<26}{"OK" if good else "FAIL"} {msg}') print(f'\n完成 {ok}/{len(results)}') return 0 if ok == len(results) else 1 if __name__ == '__main__': sys.exit(main())