140 lines
4.3 KiB
Python
140 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
|
|
@dataclass
|
|
class GlobalConfig:
|
|
cache_dir: Path = Path("/var/lib/xray-lists/cache")
|
|
output_dir: Path = Path("/var/lib/xray-lists/generated")
|
|
dnsmasq_output: Path = Path("/var/lib/xray-lists/generated/nftsets.conf")
|
|
proxy: str | None = None
|
|
proxy_user: str | None = None
|
|
proxy_pass: str | None = None
|
|
http_timeout: int = 20
|
|
|
|
|
|
@dataclass
|
|
class ListEntry:
|
|
id: str
|
|
kind: str
|
|
static: list[str] = field(default_factory=list)
|
|
urls: list[str] = field(default_factory=list)
|
|
output: Path | None = None
|
|
dnsmasq_target: str | None = None
|
|
|
|
def source_label(self) -> str:
|
|
parts = []
|
|
if self.static:
|
|
parts.append(f"{len(self.static)} static")
|
|
if self.urls:
|
|
parts.append(f"{len(self.urls)} url(s)")
|
|
return ", ".join(parts) or "empty"
|
|
|
|
|
|
@dataclass
|
|
class Config:
|
|
global_cfg: GlobalConfig
|
|
ip_sets: list[ListEntry]
|
|
domain_sets: list[ListEntry]
|
|
|
|
def all_entries(self) -> list[ListEntry]:
|
|
return self.ip_sets + self.domain_sets
|
|
|
|
|
|
def _die(msg: str) -> None:
|
|
print(f"config error: {msg}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def load_config(path: str | Path) -> Config:
|
|
path = Path(path)
|
|
if not path.is_file():
|
|
_die(f"config file not found: {path}")
|
|
|
|
with path.open("r", encoding="utf-8") as fh:
|
|
raw = yaml.safe_load(fh) or {}
|
|
|
|
g_raw = raw.get("global", {}) or {}
|
|
gcfg = GlobalConfig(
|
|
cache_dir=Path(g_raw.get("cache_dir", GlobalConfig.cache_dir)),
|
|
output_dir=Path(g_raw.get("output_dir", GlobalConfig.output_dir)),
|
|
dnsmasq_output=Path(g_raw.get("dnsmasq_output", GlobalConfig.dnsmasq_output)),
|
|
proxy=g_raw.get("proxy"),
|
|
proxy_user=g_raw.get("proxy_user"),
|
|
proxy_pass=g_raw.get("proxy_pass"),
|
|
http_timeout=int(g_raw.get("http_timeout", GlobalConfig.http_timeout)),
|
|
)
|
|
|
|
seen_ids: set[str] = set()
|
|
|
|
def parse_ip_entries(raw_list) -> list[ListEntry]:
|
|
out = []
|
|
for item in raw_list or []:
|
|
eid = item.get("id")
|
|
if not eid:
|
|
_die(f"ip entry missing required 'id': {item}")
|
|
if eid in seen_ids:
|
|
_die(f"duplicate entry id '{eid}'")
|
|
seen_ids.add(eid)
|
|
|
|
static = item.get("static") or []
|
|
urls = item.get("urls") or []
|
|
if not static and not urls:
|
|
_die(f"entry '{eid}' has neither 'static' nor 'urls' - nothing to do")
|
|
|
|
output = item.get("output")
|
|
if not output:
|
|
_die(f"ip entry '{eid}' missing required 'output' (filename or path)")
|
|
output_path = Path(output)
|
|
if not output_path.is_absolute():
|
|
output_path = gcfg.output_dir / output_path
|
|
|
|
out.append(ListEntry(
|
|
id=eid, kind="ip",
|
|
static=[str(x) for x in static],
|
|
urls=[str(x) for x in urls],
|
|
output=output_path,
|
|
))
|
|
return out
|
|
|
|
def parse_domain_entries(raw_list) -> list[ListEntry]:
|
|
out = []
|
|
for item in raw_list or []:
|
|
eid = item.get("id")
|
|
if not eid:
|
|
_die(f"domain entry missing required 'id': {item}")
|
|
if eid in seen_ids:
|
|
_die(f"duplicate entry id '{eid}'")
|
|
seen_ids.add(eid)
|
|
|
|
static = item.get("static") or []
|
|
urls = item.get("urls") or []
|
|
if not static and not urls:
|
|
_die(f"entry '{eid}' has neither 'static' nor 'urls' - nothing to do")
|
|
|
|
target = item.get("dnsmasq_target")
|
|
if not target:
|
|
_die(f"domain entry '{eid}' missing required 'dnsmasq_target' "
|
|
f"(e.g. '4#ip#xray#v2ray_dom')")
|
|
|
|
out.append(ListEntry(
|
|
id=eid, kind="domain",
|
|
static=[str(x) for x in static],
|
|
urls=[str(x) for x in urls],
|
|
dnsmasq_target=str(target),
|
|
))
|
|
return out
|
|
|
|
ip_sets = parse_ip_entries(raw.get("ip_sets"))
|
|
domain_sets = parse_domain_entries(raw.get("domain_sets"))
|
|
|
|
if not ip_sets and not domain_sets:
|
|
_die("config has no ip_sets/domain_sets defined")
|
|
|
|
return Config(global_cfg=gcfg, ip_sets=ip_sets, domain_sets=domain_sets)
|