from __future__ import annotations import json from dataclasses import dataclass from pathlib import Path from typing import Literal Strategy = Literal["broadcast", "top2", "top3", "top4", "backup"] KernelMode = Literal["auto", "20", "24"] @dataclass class RelayNode: name: str host: str port: int token: str weight: int = 100 @dataclass class Config: relays: list[RelayNode] strategy: Strategy = "top3" kernel_mode: KernelMode = "auto" redundancy: int = 3 tcp_warmup_bytes: int = 1048576 probe_interval: float = 15.0 tcp_loser_grace_ms: int = 1500 direct_open_timeout: float = 10.0 relay_open_timeout: float = 10.0 tcp_connect_happy_eyeballs_delay: float | None = None relay_reconnect_delay: float = 3.0 relay_tcp_nodelay: bool = True @classmethod def load(cls, path: str) -> "Config": raw = json.loads(Path(path).read_text()) relays = [RelayNode(**item) for item in raw.get("relays", [])] return cls( relays=relays, strategy=raw.get("strategy", "top3"), kernel_mode=raw.get("kernel_mode", "auto"), redundancy=raw.get("redundancy", 3), tcp_warmup_bytes=raw.get("tcp_warmup_bytes", 1048576), probe_interval=raw.get("probe_interval", 15.0), tcp_loser_grace_ms=raw.get("tcp_loser_grace_ms", 1500), direct_open_timeout=raw.get("direct_open_timeout", 10.0), relay_open_timeout=raw.get("relay_open_timeout", 10.0), tcp_connect_happy_eyeballs_delay=raw.get("tcp_connect_happy_eyeballs_delay"), relay_reconnect_delay=raw.get("relay_reconnect_delay", 3.0), relay_tcp_nodelay=raw.get("relay_tcp_nodelay", True), )