utils/proxy_manager.py

import time
from dataclasses import dataclass
from typing import Optional

import requests


@dataclass
class ProxyConfig:
    enabled: bool = False
    proxy_url: str = ""
    health_check_url: str = "https://www.nicovideo.jp/"
    check_interval_seconds: int = 300
    success_ttl_seconds: int = 600
    request_timeout: int = 5


class ProxyManager:
    """Manage availability of an upstream proxy server with simple health checks."""

    def __init__(self, config: ProxyConfig, tunnel_controller=None):
        self.config = config
        self.tunnel_controller = tunnel_controller
        self._last_checked: float = 0.0
        self._last_success: float = 0.0
        self._available: bool = False

    @property
    def enabled(self) -> bool:
        if not self.config.enabled:
            return False
        if self.config.proxy_url:
            return True
        if self.tunnel_controller:
            return True
        return False

    def get_active_proxy(self) -> Optional[str]:
        proxy_url = self._get_proxy_url()
        if not self.config.enabled or not proxy_url:
            return None
        now = time.time()
        if self._available and now - self._last_success < max(30, self.config.success_ttl_seconds):
            return proxy_url
        if now - self._last_checked < max(10, self.config.check_interval_seconds):
            return proxy_url if self._available else None
        self._last_checked = now
        self._available = self._check_proxy()
        if self._available:
            self._last_success = time.time()
            return proxy_url
        return None

    def _check_proxy(self) -> bool:
        if not self.config.health_check_url:
            print("[WARN] Proxy health check URL is not configured.")
            return False
        proxy_val = self._get_proxy_url()
        if not proxy_val:
            return False
        proxies = {'http': proxy_val, 'https': proxy_val}
        try:
            response = requests.get(
                self.config.health_check_url,
                proxies=proxies,
                timeout=max(1, self.config.request_timeout)
            )
            if response.ok:
                print(f"[INFO] Upstream proxy reachable at {self.config.health_check_url}.")
                return True
            print(f"[WARN] Proxy health check returned status {response.status_code}.")
        except requests.RequestException as exc:
            print(f"[WARN] Proxy health check failed: {exc}")
        return False

    def _get_proxy_url(self) -> Optional[str]:
        if self.config.proxy_url:
            return self.config.proxy_url
        if self.tunnel_controller:
            return self.tunnel_controller.ensure_tunnel()
        return None