siloqy/prod/bingx/dns_cache.py

from __future__ import annotations

import socket
import time
from dataclasses import dataclass, field
from typing import Mapping


_STATIC_ENDPOINT_IPS: dict[str, tuple[str, ...]] = {
    "open-api.bingx.com": (
        "18.66.122.103",
        "18.66.122.71",
        "18.66.122.3",
        "18.66.122.36",
        "2600:9000:2250:4800:1a:4f2d:b440:93a1",
        "2600:9000:2250:c400:1a:4f2d:b440:93a1",
        "2600:9000:2250:6c00:1a:4f2d:b440:93a1",
        "2600:9000:2250:7200:1a:4f2d:b440:93a1",
        "2600:9000:2250:4600:1a:4f2d:b440:93a1",
        "2600:9000:2250:ee00:1a:4f2d:b440:93a1",
        "2600:9000:2250:fc00:1a:4f2d:b440:93a1",
        "2600:9000:2250:6800:1a:4f2d:b440:93a1",
    ),
    "open-api-vst.bingx.com": (
        "18.165.122.31",
        "18.165.122.22",
        "18.165.122.47",
        "18.165.122.63",
        "2600:9000:2375:a200:14:7788:7980:93a1",
        "2600:9000:2375:9000:14:7788:7980:93a1",
        "2600:9000:2375:2c00:14:7788:7980:93a1",
        "2600:9000:2375:7800:14:7788:7980:93a1",
        "2600:9000:2375:f600:14:7788:7980:93a1",
        "2600:9000:2375:ce00:14:7788:7980:93a1",
        "2600:9000:2375:e800:14:7788:7980:93a1",
        "2600:9000:2375:e200:14:7788:7980:93a1",
    ),
    "open-api-swap.bingx.com": (
        "3.164.68.61",
        "3.164.68.42",
        "3.164.68.75",
        "3.164.68.125",
        "2600:9000:278c:7400:1f:3dec:7600:93a1",
        "2600:9000:278c:3e00:1f:3dec:7600:93a1",
        "2600:9000:278c:5800:1f:3dec:7600:93a1",
        "2600:9000:278c:3400:1f:3dec:7600:93a1",
        "2600:9000:278c:ce00:1f:3dec:7600:93a1",
        "2600:9000:278c:9a00:1f:3dec:7600:93a1",
        "2600:9000:278c:9000:1f:3dec:7600:93a1",
        "2600:9000:278c:c800:1f:3dec:7600:93a1",
    ),
    "open-api.bingx.pro": (
        "2606:4700:4403::ac40:9313",
        "2a06:98c1:310d::6812:28ed",
    ),
    "open-api-vst.bingx.pro": (
        "2a06:98c1:310d::6812:28ed",
        "2606:4700:4403::ac40:9313",
    ),
    "open-api-swap.bingx.pro": (
        "2a06:98c1:310d::6812:28ed",
        "2606:4700:4403::ac40:9313",
    ),
}


@dataclass(slots=True)
class BingxDnsRecord:
    hostname: str
    ips: tuple[str, ...]
    source: str
    updated_at_ns: int


@dataclass(slots=True)
class BingxDnsFallbackCache:
    default_ips: Mapping[str, tuple[str, ...]] = field(default_factory=lambda: dict(_STATIC_ENDPOINT_IPS))
    _records: dict[str, BingxDnsRecord] = field(default_factory=dict, init=False, repr=False)

    def resolve(self, hostname: str) -> tuple[str, ...]:
        record = self._records.get(hostname)
        if record is not None:
            return record.ips
        return tuple(self.default_ips.get(hostname, ()))

    def record_static(self, hostname: str) -> tuple[str, ...]:
        ips = self.resolve(hostname)
        if ips:
            self._records[hostname] = BingxDnsRecord(
                hostname=hostname,
                ips=ips,
                source="static",
                updated_at_ns=time.monotonic_ns(),
            )
        return ips

    def refresh_from_dns(self, hostname: str) -> tuple[str, ...]:
        seen: list[str] = []
        for family in (socket.AF_UNSPEC,):
            infos = socket.getaddrinfo(hostname, None, family=family, type=socket.SOCK_STREAM)
            for info in infos:
                address = info[4][0]
                if address not in seen:
                    seen.append(address)
        ips = tuple(seen)
        if ips:
            self._records[hostname] = BingxDnsRecord(
                hostname=hostname,
                ips=ips,
                source="dns",
                updated_at_ns=time.monotonic_ns(),
            )
        return ips

    def maybe_refresh_from_dns(self, hostname: str, *, min_interval_secs: int = 300) -> tuple[str, ...] | None:
        record = self._records.get(hostname)
        if record is not None:
            age_secs = (time.monotonic_ns() - record.updated_at_ns) / 1_000_000_000
            if age_secs < min_interval_secs:
                return record.ips
        try:
            return self.refresh_from_dns(hostname)
        except Exception:
            return None


def is_bingx_hostname(hostname: str | None) -> bool:
    return hostname in _STATIC_ENDPOINT_IPS