"""YouTube push proxy resolution (from d2ypp2 douyin_youtube_ffplay, simplified)."""

from __future__ import annotations

import os
import re
import shutil
import socket
from pathlib import Path
from urllib.parse import urlparse

_PROXY_ENV_KEYS = (
    "http_proxy",
    "https_proxy",
    "HTTP_PROXY",
    "HTTPS_PROXY",
    "ALL_PROXY",
    "all_proxy",
)


def _is_disabled_value(value: str) -> bool:
    return (value or "").strip().lower() in {"0", "false", "no", "off", "none", "direct"}


def _is_local_tcp_open(host: str, port: int, timeout: float = 0.35) -> bool:
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False


def normalize_push_proxy_mode(configured: str | None = None) -> str:
    value = configured or os.environ.get("YOUTUBE_PUSH_PROXY_MODE") or os.environ.get("LIVE_YOUTUBE_PUSH_PROXY_MODE")
    if not value:
        value = "auto"
    low = (value or "").strip().lower()
    if low in {"0", "false", "no", "off", "none", "direct"}:
        return "off"
    if low in {"env", "environment", "http_proxy", "https_proxy"}:
        return "env"
    if low in {"proxychains", "proxy", "chain", "proxychains4"}:
        return "proxychains"
    return "auto"


def resolve_youtube_push_proxy(configured: str | None = None) -> str:
    for value in (
        configured,
        os.environ.get("YOUTUBE_PUSH_PROXY"),
        os.environ.get("LIVE_YOUTUBE_PUSH_PROXY"),
    ):
        value = (value or "").strip()
        if _is_disabled_value(value):
            return ""
        if value:
            return value

    auto = (os.environ.get("YOUTUBE_PUSH_PROXY_AUTO") or "1").strip().lower()
    port_raw = os.environ.get("YOUTUBE_PUSH_PROXY_PORT") or os.environ.get("LIVE_YOUTUBE_PUSH_PROXY_PORT") or "7890"
    try:
        proxy_port = int(str(port_raw).strip())
    except (TypeError, ValueError):
        proxy_port = 7890
    if auto not in {"0", "false", "no", "off"} and _is_local_tcp_open("127.0.0.1", proxy_port):
        return f"socks5://127.0.0.1:{proxy_port}"

    for key in ("https_proxy", "http_proxy", "HTTPS_PROXY", "HTTP_PROXY", "ALL_PROXY", "all_proxy"):
        value = (os.environ.get(key) or "").strip()
        if _is_disabled_value(value):
            return ""
        if value:
            return value
    return ""


def redact_proxy_url(value: str) -> str:
    return re.sub(r"(://[^:/@\s]+:)[^@/\s]+(@)", r"\1<redacted>\2", value or "")


def should_use_proxychains(push_proxy: str, proxy_mode: str = "auto") -> bool:
    if not push_proxy:
        return False
    mode = normalize_push_proxy_mode(proxy_mode)
    if mode in {"off", "env"}:
        return False
    if mode == "proxychains":
        return bool(shutil.which("proxychains4"))
    scheme = (urlparse(push_proxy if "://" in push_proxy else f"socks5://{push_proxy}").scheme or "socks5").lower()
    if scheme.startswith("socks"):
        return bool(shutil.which("proxychains4"))
    return False


def proxychains_proxy_line(push_proxy: str) -> str | None:
    value = (push_proxy or "").strip()
    if not value:
        return None
    parsed = urlparse(value if "://" in value else f"socks5://{value}")
    host = parsed.hostname
    port = parsed.port
    if not host or not port:
        return None
    scheme = (parsed.scheme or "socks5").lower()
    if scheme in {"socks", "socks5", "socks5h"}:
        proxy_type = "socks5"
    elif scheme == "socks4":
        proxy_type = "socks4"
    elif scheme in {"http", "https"}:
        proxy_type = "http"
    else:
        return None
    parts = [proxy_type, host, str(port)]
    if parsed.username:
        parts.append(parsed.username)
    if parsed.password:
        parts.append(parsed.password)
    return " ".join(parts)


def proxychains_config_for_push(push_proxy: str, *, base_dir: str | Path | None = None) -> str | None:
    explicit = (
        os.environ.get("YOUTUBE_PROXYCHAINS_CONF")
        or os.environ.get("LIVE_YOUTUBE_PROXYCHAINS_CONF")
        or ""
    ).strip()
    if explicit and os.path.isfile(explicit):
        return explicit

    proxy_line = proxychains_proxy_line(push_proxy)
    root = Path(base_dir or os.path.dirname(os.path.abspath(__file__)))
    if proxy_line:
        runtime_dir = root / ".process_runner"
        try:
            runtime_dir.mkdir(parents=True, exist_ok=True)
            conf = runtime_dir / "proxychains-youtube.conf"
            conf.write_text(
                "strict_chain\n"
                "proxy_dns\n"
                "tcp_read_time_out 30000\n"
                "tcp_connect_time_out 8000\n\n"
                "[ProxyList]\n"
                f"{proxy_line}\n",
                encoding="utf-8",
            )
            try:
                os.chmod(conf, 0o600)
            except OSError:
                pass
            return str(conf)
        except OSError:
            pass

    bundled = root / "config" / "proxychains-youtube.conf"
    if bundled.is_file():
        return str(bundled)
    return None


def ffmpeg_env_with_push_proxy(
    push_proxy: str,
    proxy_mode: str = "auto",
    *,
    use_proxychains: bool = False,
    base_env: dict[str, str] | None = None,
) -> dict[str, str]:
    env = dict(base_env or os.environ)
    for key in _PROXY_ENV_KEYS:
        env.pop(key, None)
    push_proxy = (push_proxy or "").strip()
    if not push_proxy:
        return env
    mode = normalize_push_proxy_mode(proxy_mode)
    if mode == "off" or use_proxychains:
        return env
    for key in _PROXY_ENV_KEYS:
        env[key] = push_proxy
    return env


def proxychains_command(ffmpeg_command: list[str], push_proxy: str, proxy_mode: str = "auto", *, base_dir: str | Path | None = None) -> list[str] | None:
    if not should_use_proxychains(push_proxy, proxy_mode):
        return None
    conf = proxychains_config_for_push(push_proxy, base_dir=base_dir)
    if not conf:
        return None
    return ["proxychains4", "-q", "-f", conf, *ffmpeg_command]


def apply_push_proxy_to_ffmpeg_cmd(
    ffmpeg_command: list[str],
    push_proxy: str,
    proxy_mode: str = "auto",
    *,
    base_dir: str | Path | None = None,
) -> tuple[list[str], bool]:
    """Return (command, used_proxychains). Falls back to -http_proxy when proxychains unavailable."""
    wrapped = proxychains_command(ffmpeg_command, push_proxy, proxy_mode, base_dir=base_dir)
    if wrapped:
        return wrapped, True
    mode = normalize_push_proxy_mode(proxy_mode)
    if push_proxy and mode != "off":
        cmd = list(ffmpeg_command)
        cmd.insert(1, "-http_proxy")
        cmd.insert(2, push_proxy)
        return cmd, False
    return ffmpeg_command, False