"""Proxy and SSH tunnel utilities for bypassing network restrictions.

Provides automatic proxy detection, manual proxy configuration,
and SSH tunnel management for use behind the GFW.

Usage:
    from src.utils.proxy_tunnel import get_proxy_url, ProxyConfig, SSHTunnel

    # Auto-detect or load configured proxy
    proxy = get_proxy_url()  # e.g. "socks5://127.0.0.1:1080"

    # SSH tunnel to forward local port → remote host
    tunnel = SSHTunnel("user@vps.example.com", local_port=11080, remote_host="api.openai.com", remote_port=443)
    tunnel.start()
    ...
    tunnel.stop()
"""

from __future__ import annotations

import logging
import os
import re
import socket
import subprocess
import sys
import threading
import time
from dataclasses import dataclass, field

log = logging.getLogger(__name__)

# Well-known local proxy ports used by popular clients in China
_COMMON_PROXY_PORTS: list[tuple[str, int]] = [
    ("socks5", 1080),   # Shadowsocks / V2Ray default SOCKS5
    ("socks5", 10808),  # V2RayN SOCKS5
    ("http", 1087),     # ClashX HTTP
    ("http", 7890),     # Clash HTTP
    ("socks5", 7891),   # Clash SOCKS5
    ("http", 8118),     # Privoxy
    ("http", 8080),     # Generic HTTP proxy
    ("http", 8889),     # Surge HTTP
    ("socks5", 1081),   # ShadowsocksR
]


@dataclass
class ProxyConfig:
    """Resolved proxy configuration."""

    scheme: str = ""        # "http", "https", "socks5"
    host: str = ""
    port: int = 0
    username: str = ""
    password: str = ""
    source: str = ""        # how this proxy was discovered

    @property
    def url(self) -> str:
        """Return proxy URL string like ``socks5://host:port``."""
        if not self.host:
            return ""
        auth = ""
        if self.username:
            auth = self.username
            if self.password:
                auth += f":{self.password}"
            auth += "@"
        return f"{self.scheme}://{auth}{self.host}:{self.port}"

    @property
    def is_valid(self) -> bool:
        return bool(self.scheme and self.host and self.port)

    def as_openai_proxies(self) -> dict[str, str]:
        """Return dict suitable for ``httpx.Client(proxy=...)`` or env vars."""
        if not self.is_valid:
            return {}
        u = self.url
        return {"http://": u, "https://": u}


# ------------------------------------------------------------------ #
# Detection: environment variables
# ------------------------------------------------------------------ #

def _proxy_from_env() -> ProxyConfig | None:
    """Read proxy from standard environment variables."""
    for var in ("https_proxy", "HTTPS_PROXY", "http_proxy", "HTTP_PROXY",
                "all_proxy", "ALL_PROXY"):
        raw = os.environ.get(var, "").strip()
        if raw:
            parsed = _parse_proxy_url(raw)
            if parsed and parsed.is_valid:
                parsed.source = f"env:{var}"
                return parsed
    return None


def _parse_proxy_url(url: str) -> ProxyConfig | None:
    """Parse ``scheme://[user:pass@]host:port`` into ProxyConfig."""
    m = re.match(
        r"^(?P<scheme>https?|socks[45]?)://"
        r"(?:(?P<user>[^:@]+)(?::(?P<pass>[^@]+))?@)?"
        r"(?P<host>[^:/]+)"
        r"(?::(?P<port>\d+))?",
        url,
    )
    if not m:
        return None
    scheme = m.group("scheme")
    host = m.group("host")
    port = int(m.group("port") or ("1080" if "socks" in scheme else "8080"))
    return ProxyConfig(
        scheme=scheme,
        host=host,
        port=port,
        username=m.group("user") or "",
        password=m.group("pass") or "",
    )


# ------------------------------------------------------------------ #
# Detection: macOS system proxy (scutil)
# ------------------------------------------------------------------ #

def _proxy_from_macos_system() -> ProxyConfig | None:
    """Read proxy from macOS System Preferences via scutil."""
    if sys.platform != "darwin":
        return None
    try:
        out = subprocess.check_output(
            ["scutil", "--proxy"], timeout=3, text=True, stderr=subprocess.DEVNULL,
        )
    except Exception:
        return None

    # SOCKS proxy
    if re.search(r"SOCKSEnable\s*:\s*1", out):
        host = _scutil_value(out, "SOCKSProxy")
        port = _scutil_value(out, "SOCKSPort")
        if host and port:
            return ProxyConfig(
                scheme="socks5", host=host, port=int(port), source="macos:socks",
            )

    # HTTPS proxy
    if re.search(r"HTTPSEnable\s*:\s*1", out):
        host = _scutil_value(out, "HTTPSProxy")
        port = _scutil_value(out, "HTTPSPort")
        if host and port:
            return ProxyConfig(
                scheme="http", host=host, port=int(port), source="macos:https",
            )

    # HTTP proxy
    if re.search(r"HTTPEnable\s*:\s*1", out):
        host = _scutil_value(out, "HTTPProxy")
        port = _scutil_value(out, "HTTPPort")
        if host and port:
            return ProxyConfig(
                scheme="http", host=host, port=int(port), source="macos:http",
            )

    return None


def _scutil_value(text: str, key: str) -> str | None:
    m = re.search(rf"{key}\s*:\s*(\S+)", text)
    return m.group(1) if m else None


# ------------------------------------------------------------------ #
# Detection: Windows system proxy (registry)
# ------------------------------------------------------------------ #

def _proxy_from_windows_system() -> ProxyConfig | None:
    """Read proxy from Windows Internet Settings registry."""
    if sys.platform != "win32":
        return None
    try:
        import winreg
        key = winreg.OpenKey(
            winreg.HKEY_CURRENT_USER,
            r"Software\Microsoft\Windows\CurrentVersion\Internet Settings",
        )
        enabled, _ = winreg.QueryValueEx(key, "ProxyEnable")
        if not enabled:
            return None
        server, _ = winreg.QueryValueEx(key, "ProxyServer")
        winreg.CloseKey(key)
    except Exception:
        return None

    if not server:
        return None

    # Format: "host:port" or "socks=host:port" or "http=host:port;https=host:port"
    # Try socks first
    m = re.search(r"socks=([^;:]+):(\d+)", server)
    if m:
        return ProxyConfig(
            scheme="socks5", host=m.group(1), port=int(m.group(2)),
            source="windows:registry:socks",
        )
    # Try https
    m = re.search(r"https?=([^;:]+):(\d+)", server)
    if m:
        return ProxyConfig(
            scheme="http", host=m.group(1), port=int(m.group(2)),
            source="windows:registry:http",
        )
    # Plain host:port
    m = re.match(r"([^:]+):(\d+)$", server.strip())
    if m:
        return ProxyConfig(
            scheme="http", host=m.group(1), port=int(m.group(2)),
            source="windows:registry",
        )
    return None


# ------------------------------------------------------------------ #
# Detection: scan common local proxy ports
# ------------------------------------------------------------------ #

def _probe_local_proxy() -> ProxyConfig | None:
    """Try connecting to well-known local proxy ports."""
    for scheme, port in _COMMON_PROXY_PORTS:
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(0.3)
            sock.connect(("127.0.0.1", port))
            sock.close()
            log.debug("Local proxy detected at 127.0.0.1:%d (%s)", port, scheme)
            return ProxyConfig(
                scheme=scheme, host="127.0.0.1", port=port,
                source=f"probe:{scheme}:{port}",
            )
        except (OSError, socket.timeout):
            continue
    return None


# ------------------------------------------------------------------ #
# Main API: get_proxy_config / get_proxy_url
# ------------------------------------------------------------------ #

_cached_proxy: ProxyConfig | None = None
_cache_lock = threading.Lock()
_cache_ts: float = 0.0
_CACHE_TTL = 120.0  # 2 minutes


def get_proxy_config(
    *,
    force: bool = False,
    manual_url: str = "",
) -> ProxyConfig | None:
    """Detect or return a usable proxy configuration.

    Resolution order:
      1. ``manual_url`` (if provided) — user-configured proxy
      2. Environment variables (https_proxy, http_proxy, all_proxy)
      3. System proxy (macOS scutil / Windows registry)
      4. Local port scan (Shadowsocks, Clash, V2Ray, etc.)

    Results are cached for 2 minutes unless *force=True*.

    Returns:
        ProxyConfig if a proxy was found, None otherwise.
    """
    global _cached_proxy, _cache_ts

    # Manual URL takes highest priority and is not cached
    if manual_url:
        parsed = _parse_proxy_url(manual_url)
        if parsed and parsed.is_valid:
            parsed.source = "manual"
            log.info("Using manual proxy: %s", parsed.url)
            return parsed
        log.warning("Invalid manual proxy URL: %s", manual_url[:80])
        return None

    now = time.monotonic()
    with _cache_lock:
        if not force and _cached_proxy is not None and (now - _cache_ts) < _CACHE_TTL:
            return _cached_proxy

    # Try detection strategies in order
    proxy = (
        _proxy_from_env()
        or (_proxy_from_macos_system() if sys.platform == "darwin" else None)
        or (_proxy_from_windows_system() if sys.platform == "win32" else None)
        or _probe_local_proxy()
    )

    with _cache_lock:
        _cached_proxy = proxy
        _cache_ts = now

    if proxy:
        log.info("Proxy detected [%s]: %s", proxy.source, proxy.url)
    else:
        log.debug("No proxy detected")

    return proxy


def get_proxy_url(*, force: bool = False, manual_url: str = "") -> str:
    """Convenience wrapper — returns proxy URL string or empty string."""
    cfg = get_proxy_config(force=force, manual_url=manual_url)
    return cfg.url if cfg else ""


def clear_proxy_cache() -> None:
    """Force next call to re-detect proxy."""
    global _cached_proxy, _cache_ts
    with _cache_lock:
        _cached_proxy = None
        _cache_ts = 0.0


# ------------------------------------------------------------------ #
# SSH Tunnel
# ------------------------------------------------------------------ #

@dataclass
class SSHTunnel:
    """Manage an SSH local-port-forward tunnel as a subprocess.

    Example::

        tunnel = SSHTunnel(
            ssh_target="user@my-vps.com",
            local_port=11080,
            remote_host="api.deepseek.com",
            remote_port=443,
        )
        tunnel.start()
        # ... use localhost:11080 as proxy to api.deepseek.com:443 ...
        tunnel.stop()
    """

    ssh_target: str                     # user@host or host (uses ~/.ssh/config)
    local_port: int = 11080
    remote_host: str = "127.0.0.1"
    remote_port: int = 443
    ssh_port: int = 22
    ssh_key: str = ""                   # path to private key (optional)
    connect_timeout: int = 10

    _process: subprocess.Popen | None = field(default=None, repr=False, init=False)
    _lock: threading.Lock = field(default_factory=threading.Lock, repr=False, init=False)

    @property
    def is_running(self) -> bool:
        with self._lock:
            return self._process is not None and self._process.poll() is None

    @property
    def local_url(self) -> str:
        """URL to use as proxy target (points to local forwarded port)."""
        return f"http://127.0.0.1:{self.local_port}"

    def start(self) -> bool:
        """Start the SSH tunnel. Returns True on success."""
        with self._lock:
            if self._process is not None and self._process.poll() is None:
                log.debug("SSH tunnel already running (pid=%d)", self._process.pid)
                return True

            cmd = [
                "ssh", "-N",                           # no remote command
                "-L", f"{self.local_port}:{self.remote_host}:{self.remote_port}",
                "-p", str(self.ssh_port),
                "-o", f"ConnectTimeout={self.connect_timeout}",
                "-o", "StrictHostKeyChecking=accept-new",
                "-o", "ServerAliveInterval=30",
                "-o", "ServerAliveCountMax=3",
                "-o", "ExitOnForwardFailure=yes",
            ]
            if self.ssh_key:
                cmd.extend(["-i", self.ssh_key])
            cmd.append(self.ssh_target)

            try:
                self._process = subprocess.Popen(
                    cmd,
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.PIPE,
                    stdin=subprocess.DEVNULL,
                )
            except FileNotFoundError:
                log.error("ssh binary not found — cannot create tunnel")
                return False
            except Exception as exc:
                log.error("Failed to start SSH tunnel: %s", exc)
                return False

        # Wait briefly for the tunnel to establish
        if not self._wait_for_port(timeout=self.connect_timeout + 2):
            stderr = ""
            with self._lock:
                if self._process and self._process.poll() is not None:
                    stderr = (self._process.stderr.read() or b"").decode(errors="replace")
            log.error("SSH tunnel failed to bind local port %d: %s", self.local_port, stderr[:300])
            self.stop()
            return False

        log.info(
            "SSH tunnel started: localhost:%d → %s:%d via %s (pid=%d)",
            self.local_port, self.remote_host, self.remote_port,
            self.ssh_target, self._process.pid,
        )
        return True

    def stop(self) -> None:
        """Stop the SSH tunnel subprocess."""
        with self._lock:
            if self._process is None:
                return
            proc = self._process
            self._process = None

        try:
            proc.terminate()
            proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.wait(timeout=2)
        except Exception:
            pass
        log.info("SSH tunnel stopped")

    def _wait_for_port(self, timeout: float = 12) -> bool:
        """Wait until the local port is accepting connections."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            # Check if process died
            with self._lock:
                if self._process and self._process.poll() is not None:
                    return False
            try:
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.settimeout(0.5)
                sock.connect(("127.0.0.1", self.local_port))
                sock.close()
                return True
            except (OSError, socket.timeout):
                time.sleep(0.3)
        return False

    def __del__(self) -> None:
        self.stop()


# ------------------------------------------------------------------ #
# High-level: auto-detect proxy when behind GFW
# ------------------------------------------------------------------ #

def get_proxy_for_gfw(manual_url: str = "") -> ProxyConfig | None:
    """Return proxy config only if behind the GFW and a proxy is available.

    Combines GFW detection with proxy detection for one-call usage::

        proxy = get_proxy_for_gfw()
        if proxy:
            # route requests through proxy
            httpx.Client(proxy=proxy.url)
    """
    from .gfw_detect import is_behind_gfw

    if not is_behind_gfw():
        return None

    return get_proxy_config(manual_url=manual_url)
