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)