Leonid PershinandClaude Opus 5
61c0ea3558
Домены и сети Discord в geosite.dat и geoip.dat для Xray/3x-ui, плюс sing-box, Clash/mihomo, AdGuard, текст и готовый фрагмент routing. Рукописная база живёт в data/discord-base.txt и data/ip/discord-base.txt, остальное tools/sync.py тянет из пяти публичных апстримов: JSON-исходники парсятся напрямую, скомпилированные .srs распаковываются через sing-box. Каждый источник кладётся в свою категорию, недоступный источник не обнуляет данные, файлы без шапки GENERATED скрипт не трогает. Ограничения по протоколу и портам (директивы !network / !port) доезжают до sing-box rule-set и до xray/routing.json. Такие категории намеренно не входят в сводную discord: там сети вроде 172.64.0.0/13, и без привязки к UDP-портам правило утащило бы в туннель чужой трафик. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1029 lines
41 KiB
Python
1029 lines
41 KiB
Python
#!/usr/bin/env python3
|
||
"""Собирает geosite/geoip-артефакты из data/ в несколько форматов.
|
||
|
||
Два независимых трека:
|
||
|
||
data/*.txt — домены → geosite.dat (тег = имя файла)
|
||
data/ip/*.txt — IP и CIDR → geoip.dat (тег = имя файла)
|
||
|
||
Оба .dat — protobuf v2ray/Xray, ровно то, что кладут рядом с x-ui.
|
||
Дополнительно: plain text, JSON, sing-box rule-set, Clash/mihomo, AdGuard
|
||
и готовый фрагмент routing для Xray/3x-ui.
|
||
|
||
Зависимостей нет — только стандартная библиотека. Внешние бинарники
|
||
(sing-box, mihomo) нужны лишь для компиляции .srs/.mrs и опциональны.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import ipaddress
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
DATA_DIR = ROOT / "data"
|
||
IP_SUBDIR = "ip"
|
||
|
||
TYPE_DOMAIN = "domain"
|
||
TYPE_FULL = "full"
|
||
TYPE_KEYWORD = "keyword"
|
||
TYPE_REGEXP = "regexp"
|
||
|
||
# Значения enum Domain.Type из v2ray router protobuf.
|
||
PB_TYPE = {TYPE_KEYWORD: 0, TYPE_REGEXP: 1, TYPE_DOMAIN: 2, TYPE_FULL: 3}
|
||
# Порядок вывода: сначала суффиксы, потом точные, потом keyword/regexp.
|
||
TYPE_ORDER = {TYPE_DOMAIN: 0, TYPE_FULL: 1, TYPE_KEYWORD: 2, TYPE_REGEXP: 3}
|
||
|
||
LABEL_RE = r"(?!-)[a-z0-9_-]{1,63}(?<!-)"
|
||
DOMAIN_RE = re.compile(rf"^{LABEL_RE}(\.{LABEL_RE})+$")
|
||
ATTR_RE = re.compile(r"^[a-z0-9_-]+$")
|
||
CATEGORY_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
|
||
PORT_RE = re.compile(r"^(\d{1,5})(?:[-:](\d{1,5}))?$")
|
||
|
||
NETWORKS = ("tcp", "udp")
|
||
|
||
# Шапка, по которой sync.py отличает свои файлы от рукописных.
|
||
GENERATED_MARK = "GENERATED"
|
||
|
||
|
||
class BuildError(Exception):
|
||
"""Ошибка в исходных данных — сборка должна упасть."""
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Модель данных
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Rule:
|
||
"""Доменное правило geosite."""
|
||
|
||
type: str
|
||
value: str
|
||
attrs: tuple[str, ...] = ()
|
||
|
||
@property
|
||
def sort_key(self) -> tuple[int, str]:
|
||
return (TYPE_ORDER[self.type], self.value)
|
||
|
||
def __str__(self) -> str:
|
||
base = self.value if self.type == TYPE_DOMAIN else f"{self.type}:{self.value}"
|
||
return " ".join([base, *(f"@{a}" for a in self.attrs)])
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Net:
|
||
"""IP-сеть geoip: упакованный адрес + длина префикса."""
|
||
|
||
packed: bytes
|
||
prefix: int
|
||
|
||
@property
|
||
def version(self) -> int:
|
||
return 4 if len(self.packed) == 4 else 6
|
||
|
||
@property
|
||
def sort_key(self) -> tuple[int, bytes, int]:
|
||
return (len(self.packed), self.packed, self.prefix)
|
||
|
||
@property
|
||
def network(self) -> ipaddress.IPv4Network | ipaddress.IPv6Network:
|
||
return ipaddress.ip_network(str(self))
|
||
|
||
def __str__(self) -> str:
|
||
return f"{ipaddress.ip_address(self.packed)}/{self.prefix}"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Options:
|
||
"""Ограничения категории: протокол и порты.
|
||
|
||
Пустые поля означают «без ограничений». Выразимы только в sing-box и в
|
||
routing-правилах Xray: ни geoip.dat, ни Clash-ipcidr портов не знают.
|
||
"""
|
||
|
||
networks: tuple[str, ...] = ()
|
||
ports: tuple[tuple[int, int], ...] = ()
|
||
|
||
@property
|
||
def restricted(self) -> bool:
|
||
return bool(self.networks or self.ports)
|
||
|
||
@property
|
||
def xray_ports(self) -> str:
|
||
return ",".join(f"{a}-{b}" if a != b else str(a) for a, b in self.ports)
|
||
|
||
def __str__(self) -> str:
|
||
parts = []
|
||
if self.networks:
|
||
parts.append("/".join(self.networks))
|
||
if self.ports:
|
||
parts.append(self.xray_ports)
|
||
return " ".join(parts) or "без ограничений"
|
||
|
||
|
||
def merge_ports(ports: list[tuple[int, int]]) -> tuple[tuple[int, int], ...]:
|
||
"""Схлопывает пересекающиеся и смежные диапазоны — вывод детерминирован."""
|
||
merged: list[tuple[int, int]] = []
|
||
for start, end in sorted(ports):
|
||
if merged and start <= merged[-1][1] + 1:
|
||
merged[-1] = (merged[-1][0], max(merged[-1][1], end))
|
||
else:
|
||
merged.append((start, end))
|
||
return tuple(merged)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Разбор исходников: домены
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def source_path(category: str) -> Path:
|
||
return DATA_DIR / f"{category}.txt"
|
||
|
||
|
||
def ip_source_path(category: str) -> Path:
|
||
return DATA_DIR / IP_SUBDIR / f"{category}.txt"
|
||
|
||
|
||
def list_categories() -> list[str]:
|
||
return sorted(p.stem for p in DATA_DIR.glob("*.txt"))
|
||
|
||
|
||
def list_ip_categories() -> list[str]:
|
||
return sorted(p.stem for p in (DATA_DIR / IP_SUBDIR).glob("*.txt"))
|
||
|
||
|
||
def strip_comment(raw: str) -> str:
|
||
return raw.split("#", 1)[0].strip()
|
||
|
||
|
||
def parse_token(token: str, where: str) -> Rule:
|
||
prefix, sep, value = token.partition(":")
|
||
if sep and prefix.lower() in (TYPE_DOMAIN, TYPE_FULL, TYPE_KEYWORD, TYPE_REGEXP):
|
||
rtype = prefix.lower()
|
||
else:
|
||
rtype, value = TYPE_DOMAIN, token
|
||
|
||
if not value:
|
||
raise BuildError(f"{where}: пустое значение в '{token}'")
|
||
|
||
if rtype in (TYPE_DOMAIN, TYPE_FULL):
|
||
try:
|
||
value = value.encode("idna").decode("ascii")
|
||
except UnicodeError:
|
||
pass # ниже отловит DOMAIN_RE
|
||
value = value.lower().strip(".")
|
||
if not DOMAIN_RE.match(value):
|
||
raise BuildError(f"{where}: невалидный домен '{value}'")
|
||
elif rtype == TYPE_KEYWORD:
|
||
value = value.lower()
|
||
if any(c.isspace() for c in value):
|
||
raise BuildError(f"{where}: keyword не может содержать пробелы")
|
||
elif rtype == TYPE_REGEXP:
|
||
try:
|
||
re.compile(value)
|
||
except re.error as exc:
|
||
raise BuildError(f"{where}: невалидный regexp '{value}': {exc}") from exc
|
||
|
||
return Rule(rtype, value)
|
||
|
||
|
||
def parse_include(token: str, where: str) -> str:
|
||
target = token[len("include:") :]
|
||
if not CATEGORY_RE.match(target):
|
||
raise BuildError(f"{where}: невалидное имя категории '{target}'")
|
||
return target
|
||
|
||
|
||
def parse_category(category: str) -> list[tuple[str, Rule | str]]:
|
||
"""Возвращает список ('rule', Rule) / ('include', category) в порядке файла."""
|
||
path = source_path(category)
|
||
if not path.is_file():
|
||
raise BuildError(f"категория '{category}' не найдена: {path}")
|
||
|
||
entries: list[tuple[str, Rule | str]] = []
|
||
for lineno, raw in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
||
line = strip_comment(raw)
|
||
if not line:
|
||
continue
|
||
where = f"{path.name}:{lineno}"
|
||
|
||
token, *rest = line.split()
|
||
attrs = []
|
||
for part in rest:
|
||
if not part.startswith("@"):
|
||
raise BuildError(f"{where}: ожидался атрибут '@name', получено '{part}'")
|
||
attr = part[1:].lower()
|
||
if not ATTR_RE.match(attr):
|
||
raise BuildError(f"{where}: невалидный атрибут '@{attr}'")
|
||
attrs.append(attr)
|
||
|
||
if token.startswith("include:"):
|
||
if attrs:
|
||
raise BuildError(f"{where}: include не поддерживает атрибуты")
|
||
entries.append(("include", parse_include(token, where)))
|
||
continue
|
||
|
||
rule = parse_token(token, where)
|
||
entries.append(("rule", Rule(rule.type, rule.value, tuple(sorted(set(attrs))))))
|
||
|
||
return entries
|
||
|
||
|
||
def collapse_rules(rules: list[Rule]) -> list[Rule]:
|
||
"""Убирает домены, уже покрытые суффиксом из этого же набора.
|
||
|
||
domain:discord.gg покрывает и сам домен, и любой поддомен, поэтому
|
||
full:atlanta1068.discord.gg рядом с ним — чистый балласт. Правила с
|
||
атрибутами не трогаем: '@attr' выделяет подмножество, и покрывающий
|
||
суффикс без атрибута в это подмножество не входит.
|
||
"""
|
||
suffixes = {r.value for r in rules if r.type == TYPE_DOMAIN and not r.attrs}
|
||
kept = []
|
||
for rule in rules:
|
||
if rule.type in (TYPE_DOMAIN, TYPE_FULL) and not rule.attrs:
|
||
# У domain: свой же value — это он сам, поэтому смотрим только на
|
||
# родителей; у full: суффикс может совпасть с ним целиком.
|
||
parent = rule.value if rule.type == TYPE_FULL else rule.value.partition(".")[2]
|
||
while parent:
|
||
if parent in suffixes:
|
||
break
|
||
parent = parent.partition(".")[2]
|
||
if parent:
|
||
continue
|
||
kept.append(rule)
|
||
return sorted(set(kept), key=lambda r: r.sort_key)
|
||
|
||
|
||
def collapse_nets(nets: list[Net]) -> list[Net]:
|
||
"""Убирает сети, вложенные в другие сети этого же набора."""
|
||
kept: list[Net] = []
|
||
for version in (4, 6):
|
||
ordered = sorted(
|
||
{n for n in nets if n.version == version},
|
||
key=lambda n: (int.from_bytes(n.packed, "big"), n.prefix),
|
||
)
|
||
# CIDR-сети либо вложены, либо не пересекаются, поэтому в таком порядке
|
||
# покрывающая всегда идёт раньше покрытых — хватает одной «текущей».
|
||
current: ipaddress.IPv4Network | ipaddress.IPv6Network | None = None
|
||
for net in ordered:
|
||
if current is not None and net.network.subnet_of(current): # type: ignore[arg-type]
|
||
continue
|
||
kept.append(net)
|
||
current = net.network
|
||
return sorted(kept, key=lambda n: n.sort_key)
|
||
|
||
|
||
def resolve(category: str, stack: tuple[str, ...] = ()) -> list[Rule]:
|
||
"""Разворачивает include-и, убирает дубликаты, сортирует детерминированно."""
|
||
if category in stack:
|
||
chain = " -> ".join([*stack, category])
|
||
raise BuildError(f"циклический include: {chain}")
|
||
|
||
collected: dict[Rule, None] = {}
|
||
for kind, payload in parse_category(category):
|
||
if kind == "include":
|
||
for rule in resolve(str(payload), (*stack, category)):
|
||
collected.setdefault(rule, None)
|
||
else:
|
||
collected.setdefault(payload, None) # type: ignore[arg-type]
|
||
|
||
return collapse_rules(list(collected))
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Разбор исходников: IP
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def parse_net(token: str, where: str) -> Net:
|
||
try:
|
||
network = ipaddress.ip_network(token, strict=True)
|
||
except ValueError as exc:
|
||
raise BuildError(f"{where}: невалидная сеть '{token}': {exc}") from exc
|
||
return Net(network.network_address.packed, network.prefixlen)
|
||
|
||
|
||
def parse_directive(line: str, where: str) -> tuple[str, list[str]]:
|
||
key, sep, value = line[1:].partition(":")
|
||
if not sep:
|
||
raise BuildError(f"{where}: директива без значения, ожидалось '!ключ: значение'")
|
||
values = [v for v in value.replace(",", " ").split() if v]
|
||
if not values:
|
||
raise BuildError(f"{where}: пустое значение директивы '!{key.strip()}'")
|
||
return key.strip().lower(), values
|
||
|
||
|
||
def parse_ip_category(category: str) -> tuple[list[tuple[str, Net | str]], Options]:
|
||
"""Возвращает записи файла и его ограничения (!network / !port)."""
|
||
path = ip_source_path(category)
|
||
if not path.is_file():
|
||
raise BuildError(f"IP-категория '{category}' не найдена: {path}")
|
||
|
||
entries: list[tuple[str, Net | str]] = []
|
||
networks: list[str] = []
|
||
ports: list[tuple[int, int]] = []
|
||
|
||
for lineno, raw in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
||
line = strip_comment(raw)
|
||
if not line:
|
||
continue
|
||
where = f"{IP_SUBDIR}/{path.name}:{lineno}"
|
||
|
||
if line.startswith("!"):
|
||
key, values = parse_directive(line, where)
|
||
if key == "network":
|
||
for value in values:
|
||
if value.lower() not in NETWORKS:
|
||
raise BuildError(f"{where}: !network принимает tcp/udp, получено '{value}'")
|
||
networks.append(value.lower())
|
||
elif key == "port":
|
||
for value in values:
|
||
match = PORT_RE.match(value)
|
||
if not match:
|
||
raise BuildError(f"{where}: невалидный порт '{value}', ожидалось N или A-B")
|
||
start = int(match.group(1))
|
||
end = int(match.group(2) or start)
|
||
if not 0 < start <= end <= 65535:
|
||
raise BuildError(f"{where}: порт вне диапазона 1..65535: '{value}'")
|
||
ports.append((start, end))
|
||
else:
|
||
raise BuildError(f"{where}: неизвестная директива '!{key}'")
|
||
continue
|
||
|
||
token, *rest = line.split()
|
||
if rest:
|
||
raise BuildError(f"{where}: лишнее после сети: '{' '.join(rest)}'")
|
||
|
||
if token.startswith("include:"):
|
||
entries.append(("include", parse_include(token, where)))
|
||
continue
|
||
|
||
entries.append(("net", parse_net(token, where)))
|
||
|
||
options = Options(tuple(sorted(set(networks))), merge_ports(ports))
|
||
return entries, options
|
||
|
||
|
||
def resolve_ip(category: str, stack: tuple[str, ...] = ()) -> tuple[list[Net], Options]:
|
||
if category in stack:
|
||
chain = " -> ".join([*stack, category])
|
||
raise BuildError(f"циклический include: {chain}")
|
||
|
||
entries, options = parse_ip_category(category)
|
||
collected: dict[Net, None] = {}
|
||
for kind, payload in entries:
|
||
if kind == "include":
|
||
nets, child = resolve_ip(str(payload), (*stack, category))
|
||
# Ограничения не наследуются и не объединяются: слить «udp:50000-50099»
|
||
# с безусловным списком можно только расширив одно из них — молча
|
||
# менять смысл правила нельзя.
|
||
if child != options:
|
||
raise BuildError(
|
||
f"{IP_SUBDIR}/{category}.txt: include:{payload} — ограничения не совпадают "
|
||
f"({options} против {child}); держите такие списки разными категориями"
|
||
)
|
||
for net in nets:
|
||
collected.setdefault(net, None)
|
||
else:
|
||
collected.setdefault(payload, None) # type: ignore[arg-type]
|
||
|
||
return collapse_nets(list(collected)), options
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# protobuf: geosite.dat / geoip.dat
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def _varint(value: int) -> bytes:
|
||
out = bytearray()
|
||
while True:
|
||
chunk = value & 0x7F
|
||
value >>= 7
|
||
if value:
|
||
out.append(chunk | 0x80)
|
||
else:
|
||
out.append(chunk)
|
||
return bytes(out)
|
||
|
||
|
||
def _len_field(field: int, payload: bytes) -> bytes:
|
||
return _varint(field << 3 | 2) + _varint(len(payload)) + payload
|
||
|
||
|
||
def _str_field(field: int, value: str) -> bytes:
|
||
return _len_field(field, value.encode("utf-8"))
|
||
|
||
|
||
def _varint_field(field: int, value: int) -> bytes:
|
||
return _varint(field << 3 | 0) + _varint(value)
|
||
|
||
|
||
def _read_varint(buf: bytes, pos: int) -> tuple[int, int]:
|
||
value = shift = 0
|
||
while True:
|
||
byte = buf[pos]
|
||
pos += 1
|
||
value |= (byte & 0x7F) << shift
|
||
if not byte & 0x80:
|
||
return value, pos
|
||
shift += 7
|
||
|
||
|
||
def _fields(buf: bytes):
|
||
pos = 0
|
||
while pos < len(buf):
|
||
key, pos = _read_varint(buf, pos)
|
||
field, wire = key >> 3, key & 7
|
||
if wire == 0:
|
||
value, pos = _read_varint(buf, pos)
|
||
yield field, value
|
||
elif wire == 2:
|
||
length, pos = _read_varint(buf, pos)
|
||
yield field, buf[pos : pos + length]
|
||
pos += length
|
||
else:
|
||
raise BuildError(f"неподдерживаемый wire type {wire}")
|
||
|
||
|
||
def encode_geosite_dat(categories: dict[str, list[Rule]]) -> bytes:
|
||
"""GeoSiteList{ repeated GeoSite entry = 1 }."""
|
||
out = bytearray()
|
||
for name in sorted(categories):
|
||
entry = bytearray(_str_field(1, name.upper())) # GeoSite.country_code
|
||
for rule in categories[name]:
|
||
domain = bytearray()
|
||
pb_type = PB_TYPE[rule.type]
|
||
if pb_type: # proto3 опускает значение по умолчанию (0 == Plain)
|
||
domain += _varint_field(1, pb_type)
|
||
domain += _str_field(2, rule.value)
|
||
for attr in rule.attrs:
|
||
# Attribute{ key = 1; bool_value = 2 }
|
||
domain += _len_field(3, _str_field(1, attr) + _varint_field(2, 1))
|
||
entry += _len_field(2, bytes(domain)) # GeoSite.domain
|
||
out += _len_field(1, bytes(entry))
|
||
return bytes(out)
|
||
|
||
|
||
def decode_geosite_dat(blob: bytes) -> dict[str, list[Rule]]:
|
||
"""Обратный разбор — используется для самопроверки собранного файла."""
|
||
pb_to_type = {v: k for k, v in PB_TYPE.items()}
|
||
|
||
result: dict[str, list[Rule]] = {}
|
||
for field, payload in _fields(blob):
|
||
if field != 1:
|
||
continue
|
||
name, rules = "", []
|
||
for f, p in _fields(payload):
|
||
if f == 1:
|
||
name = p.decode("utf-8")
|
||
elif f == 2:
|
||
rtype, value, attrs = TYPE_KEYWORD, "", []
|
||
for df, dp in _fields(p):
|
||
if df == 1:
|
||
rtype = pb_to_type[dp]
|
||
elif df == 2:
|
||
value = dp.decode("utf-8")
|
||
elif df == 3:
|
||
for af, ap in _fields(dp):
|
||
if af == 1:
|
||
attrs.append(ap.decode("utf-8"))
|
||
rules.append(Rule(rtype, value, tuple(attrs)))
|
||
result[name] = rules
|
||
return result
|
||
|
||
|
||
def encode_geoip_dat(categories: dict[str, list[Net]]) -> bytes:
|
||
"""GeoIPList{ repeated GeoIP entry = 1 }, GeoIP{ code = 1; repeated CIDR cidr = 2 }."""
|
||
out = bytearray()
|
||
for name in sorted(categories):
|
||
entry = bytearray(_str_field(1, name.upper())) # GeoIP.country_code
|
||
for net in categories[name]:
|
||
cidr = bytearray(_len_field(1, net.packed)) # CIDR.ip
|
||
if net.prefix: # proto3 опускает 0
|
||
cidr += _varint_field(2, net.prefix) # CIDR.prefix
|
||
entry += _len_field(2, bytes(cidr))
|
||
out += _len_field(1, bytes(entry))
|
||
return bytes(out)
|
||
|
||
|
||
def decode_geoip_dat(blob: bytes) -> dict[str, list[Net]]:
|
||
result: dict[str, list[Net]] = {}
|
||
for field, payload in _fields(blob):
|
||
if field != 1:
|
||
continue
|
||
name, nets = "", []
|
||
for f, p in _fields(payload):
|
||
if f == 1:
|
||
name = p.decode("utf-8")
|
||
elif f == 2:
|
||
packed, prefix = b"", 0
|
||
for cf, cp in _fields(p):
|
||
if cf == 1:
|
||
packed = cp
|
||
elif cf == 2:
|
||
prefix = cp
|
||
nets.append(Net(packed, prefix))
|
||
result[name] = nets
|
||
return result
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Текстовые форматы
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def write(path: Path, content: str) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(content.replace("\r\n", "\n"), encoding="utf-8", newline="\n")
|
||
|
||
|
||
def header(category: str, count: int, comment: str = "#", track: str = "") -> str:
|
||
source = f"data/{track}{category}.txt"
|
||
return (
|
||
f"{comment} discord-geodata — категория: {category}\n"
|
||
f"{comment} Правил: {count}. Файл генерируется автоматически, не редактируйте вручную.\n"
|
||
f"{comment} Источник: {source}\n"
|
||
)
|
||
|
||
|
||
def render_plain(rules: list[Rule]) -> str:
|
||
"""Только «голые» домены — для скриптов и ручной вставки."""
|
||
seen = {r.value for r in rules if r.type in (TYPE_DOMAIN, TYPE_FULL)}
|
||
return "".join(f"{value}\n" for value in sorted(seen))
|
||
|
||
|
||
def render_source(category: str, rules: list[Rule]) -> str:
|
||
return header(category, len(rules)) + "\n" + "".join(f"{r}\n" for r in rules)
|
||
|
||
|
||
def render_json(category: str, rules: list[Rule]) -> str:
|
||
payload = {
|
||
"category": category,
|
||
"count": len(rules),
|
||
"rules": [
|
||
{"type": r.type, "value": r.value, **({"attrs": list(r.attrs)} if r.attrs else {})}
|
||
for r in rules
|
||
],
|
||
}
|
||
return json.dumps(payload, ensure_ascii=False, indent=2) + "\n"
|
||
|
||
|
||
def render_singbox(rules: list[Rule]) -> str:
|
||
"""sing-box rule-set (source). domain: apex + domain_suffix: .apex == v2ray domain:."""
|
||
rule: dict[str, list[str]] = {}
|
||
for kind, values in (
|
||
("domain", sorted({r.value for r in rules if r.type in (TYPE_DOMAIN, TYPE_FULL)})),
|
||
("domain_suffix", sorted({f".{r.value}" for r in rules if r.type == TYPE_DOMAIN})),
|
||
("domain_keyword", sorted({r.value for r in rules if r.type == TYPE_KEYWORD})),
|
||
("domain_regex", sorted({r.value for r in rules if r.type == TYPE_REGEXP})),
|
||
):
|
||
if values:
|
||
rule[kind] = values
|
||
return json.dumps({"version": 1, "rules": [rule] if rule else []}, indent=2) + "\n"
|
||
|
||
|
||
def render_clash_yaml(category: str, rules: list[Rule]) -> str:
|
||
"""rule-provider с behavior: domain — keyword/regexp этим форматом не поддерживаются."""
|
||
lines = [header(category, len(rules)), "payload:\n"]
|
||
for rule in rules:
|
||
if rule.type == TYPE_DOMAIN:
|
||
lines.append(f" - '+.{rule.value}'\n")
|
||
elif rule.type == TYPE_FULL:
|
||
lines.append(f" - '{rule.value}'\n")
|
||
return "".join(lines)
|
||
|
||
|
||
def render_clash_list(category: str, rules: list[Rule]) -> str:
|
||
"""rule-provider с behavior: classical."""
|
||
mapping = {
|
||
TYPE_DOMAIN: "DOMAIN-SUFFIX",
|
||
TYPE_FULL: "DOMAIN",
|
||
TYPE_KEYWORD: "DOMAIN-KEYWORD",
|
||
TYPE_REGEXP: "DOMAIN-REGEX",
|
||
}
|
||
body = "".join(f"{mapping[r.type]},{r.value}\n" for r in rules)
|
||
return header(category, len(rules)) + body
|
||
|
||
|
||
def render_adguard(category: str, rules: list[Rule]) -> str:
|
||
lines = [header(category, len(rules), comment="!")]
|
||
for rule in rules:
|
||
if rule.type == TYPE_DOMAIN:
|
||
lines.append(f"||{rule.value}^\n")
|
||
elif rule.type == TYPE_FULL:
|
||
lines.append(f"||{rule.value}^$important\n")
|
||
return "".join(lines)
|
||
|
||
|
||
def render_ip_plain(nets: list[Net]) -> str:
|
||
return "".join(f"{net}\n" for net in nets)
|
||
|
||
|
||
def render_ip_source(category: str, nets: list[Net], options: Options) -> str:
|
||
lines = [header(category, len(nets), track=f"{IP_SUBDIR}/"), "\n"]
|
||
for network in options.networks:
|
||
lines.append(f"!network: {network}\n")
|
||
if options.ports:
|
||
lines.append(f"!port: {options.xray_ports}\n")
|
||
if options.restricted:
|
||
lines.append("\n")
|
||
lines += [f"{net}\n" for net in nets]
|
||
return "".join(lines)
|
||
|
||
|
||
def render_ip_json(category: str, nets: list[Net], options: Options) -> str:
|
||
payload: dict[str, object] = {"category": category, "count": len(nets)}
|
||
if options.networks:
|
||
payload["network"] = list(options.networks)
|
||
if options.ports:
|
||
payload["ports"] = [f"{a}-{b}" if a != b else str(a) for a, b in options.ports]
|
||
payload["cidr"] = [str(net) for net in nets]
|
||
return json.dumps(payload, ensure_ascii=False, indent=2) + "\n"
|
||
|
||
|
||
def render_ip_singbox(nets: list[Net], options: Options) -> str:
|
||
rule: dict[str, object] = {}
|
||
if options.networks:
|
||
rule["network"] = list(options.networks)
|
||
if nets:
|
||
rule["ip_cidr"] = [str(net) for net in nets]
|
||
ports = [a for a, b in options.ports if a == b]
|
||
ranges = [f"{a}:{b}" for a, b in options.ports if a != b]
|
||
if ports:
|
||
rule["port"] = ports
|
||
if ranges:
|
||
rule["port_range"] = ranges
|
||
return json.dumps({"version": 1, "rules": [rule] if nets else []}, indent=2) + "\n"
|
||
|
||
|
||
def render_ip_clash_yaml(category: str, nets: list[Net], options: Options) -> str:
|
||
lines = [header(category, len(nets), track=f"{IP_SUBDIR}/")]
|
||
if options.restricted:
|
||
lines.append(f"# Ограничения ({options}) в behavior: ipcidr не выразимы и здесь потеряны.\n")
|
||
lines.append("payload:\n")
|
||
lines += [f" - '{net}'\n" for net in nets]
|
||
return "".join(lines)
|
||
|
||
|
||
def render_ip_clash_list(category: str, nets: list[Net], options: Options) -> str:
|
||
lines = [header(category, len(nets), track=f"{IP_SUBDIR}/")]
|
||
if options.restricted:
|
||
lines.append(f"# Ограничения ({options}) в classical-правилах не выразимы и здесь потеряны.\n")
|
||
# no-resolve: правило по IP не должно заставлять резолвить домен заранее.
|
||
lines += [f"IP-CIDR{'6' if net.version == 6 else ''},{net},no-resolve\n" for net in nets]
|
||
return "".join(lines)
|
||
|
||
|
||
def essential(items: dict[str, list], collapse) -> list[str]:
|
||
"""Отбрасывает категории, целиком покрытые другой категорией.
|
||
|
||
В routing.json нет смысла перечислять и сводную категорию, и её
|
||
составляющие: одно правило Xray просто перекрывает другое.
|
||
"""
|
||
|
||
def covers(outer: str, inner: str) -> bool:
|
||
return collapse(items[outer] + items[inner]) == collapse(items[outer])
|
||
|
||
kept = []
|
||
for name in sorted(items):
|
||
if not items[name]:
|
||
continue
|
||
# При взаимном покрытии (списки совпали) оставляем первый по алфавиту —
|
||
# иначе категории вычеркнули бы друг друга.
|
||
if any(
|
||
other != name
|
||
and items[other]
|
||
and covers(other, name)
|
||
and (not covers(name, other) or other < name)
|
||
for other in items
|
||
):
|
||
continue
|
||
kept.append(name)
|
||
return kept
|
||
|
||
|
||
def render_xray_routing(
|
||
site: dict[str, list[Rule]],
|
||
ip: dict[str, tuple[list[Net], Options]],
|
||
site_file: str,
|
||
ip_file: str,
|
||
outbound: str,
|
||
) -> str:
|
||
"""Готовый фрагмент routing для 3x-ui: вставляется в конфиг как есть."""
|
||
rules: list[dict[str, object]] = []
|
||
for name in essential(site, collapse_rules):
|
||
rules.append(
|
||
{
|
||
"type": "field",
|
||
"domain": [f"ext:{site_file}:{name}"],
|
||
"outboundTag": outbound,
|
||
}
|
||
)
|
||
|
||
# Сравнивать на покрытие можно только внутри одинаковых ограничений:
|
||
# те же сети под «udp 50000-50099» — это совсем другое правило.
|
||
by_options: dict[Options, dict[str, list[Net]]] = {}
|
||
for name, (nets, options) in ip.items():
|
||
by_options.setdefault(options, {})[name] = nets
|
||
|
||
# Сначала безусловные списки, затем ограниченные по протоколу и портам:
|
||
# Xray берёт первое подошедшее правило, порядок здесь — это приоритет.
|
||
for restricted in (False, True):
|
||
for options in sorted(by_options, key=lambda o: (o.networks, o.ports)):
|
||
if options.restricted != restricted:
|
||
continue
|
||
for name in essential(by_options[options], collapse_nets):
|
||
rule: dict[str, object] = {"type": "field"}
|
||
if options.networks:
|
||
rule["network"] = ",".join(options.networks)
|
||
if options.ports:
|
||
rule["port"] = options.xray_ports
|
||
rule["ip"] = [f"ext:{ip_file}:{name}"]
|
||
rule["outboundTag"] = outbound
|
||
rules.append(rule)
|
||
|
||
payload = {
|
||
"_note": [
|
||
f"Фрагмент для Xray/3x-ui. Положите {site_file} и {ip_file} рядом с бинарём",
|
||
"(обычно /usr/local/x-ui/bin) и подставьте свой outboundTag.",
|
||
"Правила ниже перечислены от общих к ограниченным по портам.",
|
||
],
|
||
"routing": {"domainStrategy": "IPIfNonMatch", "rules": rules},
|
||
}
|
||
return json.dumps(payload, ensure_ascii=False, indent=2) + "\n"
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Компиляция бинарных rule-set (опционально)
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def compile_binaries(
|
||
out_dir: Path,
|
||
categories: list[str],
|
||
ip_categories: list[str],
|
||
singbox: str | None,
|
||
mihomo: str | None,
|
||
) -> list[str]:
|
||
produced = []
|
||
|
||
if singbox:
|
||
for track, names in (("site", categories), (IP_SUBDIR, ip_categories)):
|
||
for category in names:
|
||
src = out_dir / "sing-box" / track / f"{category}.json"
|
||
dst = src.with_suffix(".srs")
|
||
subprocess.run(
|
||
[singbox, "rule-set", "compile", "--output", str(dst), str(src)], check=True
|
||
)
|
||
produced.append(str(dst.relative_to(out_dir)))
|
||
else:
|
||
print(" sing-box не найден — .srs пропущены", file=sys.stderr)
|
||
|
||
if mihomo:
|
||
for track, names, behavior in (
|
||
("site", categories, "domain"),
|
||
(IP_SUBDIR, ip_categories, "ipcidr"),
|
||
):
|
||
for category in names:
|
||
src = out_dir / "clash" / track / f"{category}.yaml"
|
||
dst = out_dir / "mihomo" / track / f"{category}.mrs"
|
||
dst.parent.mkdir(parents=True, exist_ok=True)
|
||
subprocess.run(
|
||
[mihomo, "convert-ruleset", behavior, "yaml", str(src), str(dst)], check=True
|
||
)
|
||
produced.append(str(dst.relative_to(out_dir)))
|
||
else:
|
||
print(" mihomo не найден — .mrs пропущены", file=sys.stderr)
|
||
|
||
return produced
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Точка входа
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
SITE_DAT = "discord-geosite.dat"
|
||
IP_DAT = "discord-geoip.dat"
|
||
|
||
|
||
def build(out_dir: Path, singbox: str | None, mihomo: str | None, outbound: str) -> int:
|
||
categories = list_categories()
|
||
ip_categories = list_ip_categories()
|
||
if not categories and not ip_categories:
|
||
raise BuildError(f"в {DATA_DIR} нет ни одного .txt")
|
||
|
||
site = {name: resolve(name) for name in categories}
|
||
ip = {name: resolve_ip(name) for name in ip_categories}
|
||
|
||
if out_dir.exists():
|
||
shutil.rmtree(out_dir)
|
||
out_dir.mkdir(parents=True)
|
||
|
||
site_dat = encode_geosite_dat(site)
|
||
ip_dat = encode_geoip_dat({name: nets for name, (nets, _) in ip.items()})
|
||
|
||
# Штатные имена — чтобы файл можно было подсунуть вместо системного,
|
||
# и копии с префиксом — чтобы положить рядом и адресовать через ext:.
|
||
for name, blob in (
|
||
("geosite.dat", site_dat),
|
||
(SITE_DAT, site_dat),
|
||
("geoip.dat", ip_dat),
|
||
(IP_DAT, ip_dat),
|
||
):
|
||
(out_dir / name).write_bytes(blob)
|
||
digest = hashlib.sha256(blob).hexdigest()
|
||
write(out_dir / f"{name}.sha256sum", f"{digest} {name}\n")
|
||
|
||
# Самопроверка: файлы должны читаться обратно ровно в то, что мы записали.
|
||
if decode_geosite_dat(site_dat) != {n.upper(): r for n, r in site.items()}:
|
||
raise BuildError("geosite.dat не проходит round-trip проверку")
|
||
if decode_geoip_dat(ip_dat) != {n.upper(): nets for n, (nets, _) in ip.items()}:
|
||
raise BuildError("geoip.dat не проходит round-trip проверку")
|
||
|
||
for name, rules in site.items():
|
||
write(out_dir / "txt" / "site" / f"{name}.txt", render_plain(rules))
|
||
write(out_dir / "src" / "site" / f"{name}.txt", render_source(name, rules))
|
||
write(out_dir / "json" / "site" / f"{name}.json", render_json(name, rules))
|
||
write(out_dir / "sing-box" / "site" / f"{name}.json", render_singbox(rules))
|
||
write(out_dir / "clash" / "site" / f"{name}.yaml", render_clash_yaml(name, rules))
|
||
write(out_dir / "clash" / "site" / f"{name}.list", render_clash_list(name, rules))
|
||
write(out_dir / "adguard" / f"{name}.txt", render_adguard(name, rules))
|
||
|
||
for name, (nets, options) in ip.items():
|
||
write(out_dir / "txt" / IP_SUBDIR / f"{name}.txt", render_ip_plain(nets))
|
||
write(out_dir / "src" / IP_SUBDIR / f"{name}.txt", render_ip_source(name, nets, options))
|
||
write(out_dir / "json" / IP_SUBDIR / f"{name}.json", render_ip_json(name, nets, options))
|
||
write(out_dir / "sing-box" / IP_SUBDIR / f"{name}.json", render_ip_singbox(nets, options))
|
||
write(out_dir / "clash" / IP_SUBDIR / f"{name}.yaml", render_ip_clash_yaml(name, nets, options))
|
||
write(out_dir / "clash" / IP_SUBDIR / f"{name}.list", render_ip_clash_list(name, nets, options))
|
||
|
||
write(out_dir / "xray" / "routing.json", render_xray_routing(site, ip, SITE_DAT, IP_DAT, outbound))
|
||
|
||
compile_binaries(out_dir, categories, ip_categories, singbox, mihomo)
|
||
|
||
epoch = os.environ.get("SOURCE_DATE_EPOCH")
|
||
built_at = (
|
||
datetime.fromtimestamp(int(epoch), tz=timezone.utc)
|
||
if epoch
|
||
else datetime.now(tz=timezone.utc)
|
||
)
|
||
write(
|
||
out_dir / "metadata.json",
|
||
json.dumps(
|
||
{
|
||
"built_at": built_at.replace(microsecond=0).isoformat().replace("+00:00", "Z"),
|
||
"geosite_dat_sha256": hashlib.sha256(site_dat).hexdigest(),
|
||
"geosite_dat_size": len(site_dat),
|
||
"geoip_dat_sha256": hashlib.sha256(ip_dat).hexdigest(),
|
||
"geoip_dat_size": len(ip_dat),
|
||
"total_domains": sum(len(r) for r in site.values()),
|
||
"total_networks": sum(len(nets) for nets, _ in ip.values()),
|
||
"categories": {name: len(rules) for name, rules in site.items()},
|
||
"ip_categories": {
|
||
name: {
|
||
"count": len(nets),
|
||
**({"network": list(options.networks)} if options.networks else {}),
|
||
**({"ports": options.xray_ports} if options.ports else {}),
|
||
}
|
||
for name, (nets, options) in ip.items()
|
||
},
|
||
},
|
||
ensure_ascii=False,
|
||
indent=2,
|
||
)
|
||
+ "\n",
|
||
)
|
||
|
||
# Скрипт обновления едет вместе с артефактами: на сервере тогда нужен
|
||
# ровно один базовый URL — и для .dat, и для самого апдейтера.
|
||
updater = Path(__file__).with_name("update-geodata.sh")
|
||
if updater.is_file():
|
||
write(out_dir / updater.name, updater.read_text(encoding="utf-8"))
|
||
|
||
checksums = "".join(
|
||
f"{hashlib.sha256(path.read_bytes()).hexdigest()} {path.relative_to(out_dir).as_posix()}\n"
|
||
for path in sorted(out_dir.rglob("*"))
|
||
if path.is_file()
|
||
)
|
||
write(out_dir / "SHA256SUMS", checksums)
|
||
|
||
print(f"{SITE_DAT}: {len(site_dat)} байт, {len(site)} категорий")
|
||
for name in categories:
|
||
print(f" {name:<24} {len(site[name]):>5} правил")
|
||
print(f"{IP_DAT}: {len(ip_dat)} байт, {len(ip)} категорий")
|
||
for name in ip_categories:
|
||
nets, options = ip[name]
|
||
suffix = "" if not options.restricted else f" [{options}]"
|
||
print(f" {name:<24} {len(nets):>5} сетей{suffix}")
|
||
print(f"Готово: {out_dir}")
|
||
return 0
|
||
|
||
|
||
def check() -> int:
|
||
categories = list_categories()
|
||
ip_categories = list_ip_categories()
|
||
if not categories and not ip_categories:
|
||
raise BuildError(f"в {DATA_DIR} нет ни одного .txt")
|
||
|
||
problems = []
|
||
for name in categories:
|
||
if not CATEGORY_RE.match(name):
|
||
problems.append(f"{name}: имя категории должно быть [a-z0-9-]")
|
||
seen: dict[Rule, None] = {}
|
||
for kind, payload in parse_category(name):
|
||
if kind != "rule":
|
||
continue
|
||
rule: Rule = payload # type: ignore[assignment]
|
||
if rule in seen:
|
||
problems.append(f"{name}.txt: дубликат '{rule}'")
|
||
seen[rule] = None
|
||
suffixes = {r.value for r in seen if r.type == TYPE_DOMAIN}
|
||
for rule in seen:
|
||
if rule.type != TYPE_DOMAIN:
|
||
continue
|
||
parent = rule.value.split(".", 1)[1] if "." in rule.value else ""
|
||
while parent:
|
||
if parent in suffixes:
|
||
problems.append(f"{name}.txt: '{rule.value}' уже покрыт '{parent}'")
|
||
break
|
||
parent = parent.split(".", 1)[1] if "." in parent else ""
|
||
resolve(name) # ловит циклы и битые include
|
||
|
||
for name in ip_categories:
|
||
if not CATEGORY_RE.match(name):
|
||
problems.append(f"{IP_SUBDIR}/{name}: имя категории должно быть [a-z0-9-]")
|
||
entries, _ = parse_ip_category(name)
|
||
seen_nets: dict[Net, None] = {}
|
||
for kind, payload in entries:
|
||
if kind != "net":
|
||
continue
|
||
net: Net = payload # type: ignore[assignment]
|
||
if net in seen_nets:
|
||
problems.append(f"{IP_SUBDIR}/{name}.txt: дубликат '{net}'")
|
||
seen_nets[net] = None
|
||
by_version: dict[int, list[Net]] = {}
|
||
for net in seen_nets:
|
||
by_version.setdefault(net.version, []).append(net)
|
||
for nets in by_version.values():
|
||
# Сети отсортированы по префиксу: каждую сравниваем только с более
|
||
# широкими, поэтому пара «покрывающая — покрытая» находится один раз.
|
||
ordered = sorted(nets, key=lambda n: n.prefix)
|
||
for index, net in enumerate(ordered):
|
||
for wider in ordered[:index]:
|
||
if net.network.subnet_of(wider.network): # type: ignore[arg-type]
|
||
problems.append(f"{IP_SUBDIR}/{name}.txt: '{net}' уже покрыт '{wider}'")
|
||
break
|
||
resolve_ip(name) # ловит циклы, битые include и конфликт ограничений
|
||
|
||
for problem in problems:
|
||
print(f"error: {problem}", file=sys.stderr)
|
||
if problems:
|
||
return 1
|
||
print(f"ok: {len(categories)} доменных и {len(ip_categories)} IP-категорий, замечаний нет")
|
||
return 0
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
for stream in (sys.stdout, sys.stderr):
|
||
stream.reconfigure(encoding="utf-8", errors="replace") # type: ignore[union-attr]
|
||
|
||
parser = argparse.ArgumentParser(
|
||
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
|
||
)
|
||
parser.add_argument("--out", type=Path, default=ROOT / "dist", help="каталог сборки (по умолчанию dist/)")
|
||
parser.add_argument("--check", action="store_true", help="только проверить исходники, ничего не собирать")
|
||
parser.add_argument("--sing-box", dest="singbox", help="путь к бинарю sing-box для сборки .srs")
|
||
parser.add_argument("--mihomo", help="путь к бинарю mihomo для сборки .mrs")
|
||
parser.add_argument("--outbound", default="proxy", help="outboundTag в xray/routing.json")
|
||
args = parser.parse_args(argv)
|
||
|
||
try:
|
||
if args.check:
|
||
return check()
|
||
return build(
|
||
args.out.resolve(),
|
||
args.singbox or shutil.which("sing-box"),
|
||
args.mihomo or shutil.which("mihomo"),
|
||
args.outbound,
|
||
)
|
||
except BuildError as exc:
|
||
print(f"error: {exc}", file=sys.stderr)
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|