multinetspeeder/config.py

from __future__ import annotations

import json
import os
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
    direct_ipv6_enabled: bool = True
    tcp_failover_idle_ms: int = 1200
    relay_reconnect_delay: float = 3.0
    relay_reconnect_attempts: int = 5
    relay_reconnect_max_delay: float = 30.0
    relay_ping_interval: float = 10.0
    relay_ping_timeout: float = 25.0
    relay_tcp_nodelay: bool = True
    direct_redundancy: int = 2
    direct_redundancy_v4: int | None = None
    direct_redundancy_v6: int | None = None
    direct_max_redundancy: int = 3
    udp_redundancy: int = 2
    udp_direct_redundancy: int = 3
    udp_direct_redundancy_v4: int | None = None
    udp_direct_redundancy_v6: int | None = None
    udp_always_broadcast: bool = True
    udp_copy_interval_ms: int = 2
    udp_failover_idle_ms: int = 700
    socks_host: str = "127.0.0.1"
    socks_port: int = 0

    @classmethod
    def from_dict(cls, raw: dict) -> "Config":
        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"),
            direct_ipv6_enabled=raw.get("direct_ipv6_enabled", True),
            tcp_failover_idle_ms=max(100, raw.get("tcp_failover_idle_ms", 1200)),
            relay_reconnect_delay=raw.get("relay_reconnect_delay", 3.0),
            relay_reconnect_attempts=max(1, raw.get("relay_reconnect_attempts", 5)),
            relay_reconnect_max_delay=max(raw.get("relay_reconnect_delay", 3.0), raw.get("relay_reconnect_max_delay", 30.0)),
            relay_ping_interval=max(1.0, raw.get("relay_ping_interval", 10.0)),
            relay_ping_timeout=max(1.0, raw.get("relay_ping_timeout", 25.0)),
            relay_tcp_nodelay=raw.get("relay_tcp_nodelay", True),
            direct_redundancy=max(1, raw.get("direct_redundancy", 2)),
            direct_redundancy_v4=raw.get("direct_redundancy_v4"),
            direct_redundancy_v6=raw.get("direct_redundancy_v6"),
            direct_max_redundancy=max(1, raw.get("direct_max_redundancy", 3)),
            udp_redundancy=max(0, raw.get("udp_redundancy", 2)),
            udp_direct_redundancy=max(1, raw.get("udp_direct_redundancy", 3)),
            udp_direct_redundancy_v4=raw.get("udp_direct_redundancy_v4"),
            udp_direct_redundancy_v6=raw.get("udp_direct_redundancy_v6"),
            udp_always_broadcast=raw.get("udp_always_broadcast", True),
            udp_copy_interval_ms=max(0, raw.get("udp_copy_interval_ms", 2)),
            udp_failover_idle_ms=max(100, raw.get("udp_failover_idle_ms", 700)),
            socks_host=raw.get("socks_host", "127.0.0.1"),
            socks_port=max(0, raw.get("socks_port", 0)),
        )

    @classmethod
    def load(cls, path: str) -> "Config":
        inline_raw = os.environ.get("MYNETSPEEDER_CONFIG_JSON")
        if inline_raw:
            return cls.from_dict(json.loads(inline_raw))
        return cls.from_dict(json.loads(Path(path).read_text()))