from __future__ import annotations

import concurrent.futures
import socket
import time
from typing import Callable, Dict, List, Sequence, Tuple

from common.shared import (
    SOCKS_PASSWORD,
    SOCKS_PORT,
    SOCKS_USER,
    SOCKS_VERIFY_INTERVAL,
    SSH_PORT_POLL_SECONDS,
    SSH_PORT_PROBE_CONNECT_TIMEOUT,
    SSH_PORT_PROBE_INTERVAL,
    SSH_PORT_WAIT_TIMEOUT,
)
from common.ip_utils import is_public_ip


def check_socks5_auth(
    ip: str,
    port: int,
    username: str,
    password: str,
    *,
    timeout: int = 4,
) -> Tuple[bool, str]:
    """检查 SOCKS5 服务是否能完成账号密码认证。"""
    try:
        with socket.create_connection((ip, port), timeout=timeout) as sock:
            sock.settimeout(timeout)
            sock.sendall(b"\x05\x01\x02")
            method_reply = sock.recv(2)
            if method_reply != b"\x05\x02":
                return False, f"SOCKS握手失败: {method_reply!r}"
            user_bytes = username.encode("utf-8")
            pass_bytes = password.encode("utf-8")
            if len(user_bytes) > 255 or len(pass_bytes) > 255:
                return False, "SOCKS账号或密码过长"
            sock.sendall(b"\x01" + bytes([len(user_bytes)]) + user_bytes + bytes([len(pass_bytes)]) + pass_bytes)
            auth_reply = sock.recv(2)
            if len(auth_reply) < 2:
                return False, f"SOCKS认证无返回: {auth_reply!r}"
            if auth_reply[1] != 0:
                return False, f"SOCKS认证失败: {auth_reply!r}"
            return True, "SOCKS认证成功"
    except OSError as exc:
        return False, str(exc)


def wait_for_public_ports(
    ips: Sequence[str],
    port: int,
    emit: Callable[[str], None],
    *,
    timeout: int = 360,
    interval: int = 8,
) -> Tuple[List[str], List[str]]:
    """等待一组公网 IP 的 SOCKS5 端口完成认证检查。"""
    unique_ips: List[str] = []
    for ip in ips:
        if is_public_ip(ip) and ip not in unique_ips:
            unique_ips.append(ip)
    deadline = time.time() + timeout
    available: List[str] = []
    unavailable = list(unique_ips)
    while time.time() < deadline:
        available = []
        unavailable = []
        reasons: Dict[str, str] = {}
        for ip in unique_ips:
            ok, reason = check_socks5_auth(ip, port, SOCKS_USER, SOCKS_PASSWORD)
            if ok:
                available.append(ip)
            else:
                unavailable.append(ip)
                reasons[ip] = reason
        if not unavailable:
            return available, unavailable
        reason_preview = "; ".join(f"{ip}={reasons.get(ip, '')}" for ip in unavailable[:5])
        emit(f"SOCKS5认证检查: 可用 {len(available)}/{len(unique_ips)}，等待不可用IP: {', '.join(unavailable)}；{reason_preview}")
        time.sleep(interval)
    return available, unavailable


def should_repair_socks(reasons: Dict[str, str]) -> bool:
    """根据 SOCKS5 检测失败原因判断是否需要尝试修复远端服务。"""
    repair_tokens = (
        "握手失败",
        "认证无返回",
        "认证失败",
        "connection reset",
        "connection refused",
        "timed out",
        "timeout",
        "eof",
        "b''",
    )
    for reason in reasons.values():
        lowered = str(reason).lower()
        if any(token in lowered for token in repair_tokens):
            return True
    return False


def ssh_setup_check_passed(output: str) -> bool:
    """判断远端安装脚本输出里是否出现 SOCKS5 或 L2TP 成功痕迹。"""
    lowered = output.lower()
    has_socks = "danted" in lowered or ":23456" in lowered or "0.0.0.0:23456" in lowered
    has_l2tp = "xl2tpd" in lowered or "ipsec" in lowered or "pluto" in lowered
    return has_socks or has_l2tp


def ssh_route_check_failed(output: str) -> bool:
    """识别 SSH 脚本中路由缺失或网络不可达的失败输出。"""
    lowered = output.lower()
    if "route-missing" in lowered or "network is unreachable" in lowered:
        return True
    return False


def probe_socks5_status(
    ips: Sequence[str],
    port: int,
) -> Tuple[List[str], List[str], Dict[str, str]]:
    """并发探测一组公网 IP 的 SOCKS5 认证状态。"""
    unique_ips: List[str] = []
    for ip in ips:
        if is_public_ip(ip) and ip not in unique_ips:
            unique_ips.append(ip)
    available: List[str] = []
    unavailable: List[str] = []
    reasons: Dict[str, str] = {}
    with concurrent.futures.ThreadPoolExecutor(max_workers=min(len(unique_ips), 12) or 1) as executor:
        futures = {
            executor.submit(check_socks5_auth, ip, port, SOCKS_USER, SOCKS_PASSWORD): ip
            for ip in unique_ips
        }
        results: Dict[str, Tuple[bool, str]] = {}
        for future in concurrent.futures.as_completed(futures):
            ip = futures[future]
            try:
                results[ip] = future.result()
            except Exception as exc:
                results[ip] = (False, str(exc))
    for ip in unique_ips:
        ok, reason = results.get(ip, (False, "未返回检测结果"))
        if ok:
            available.append(ip)
        else:
            unavailable.append(ip)
            reasons[ip] = reason
    return available, unavailable, reasons


def wait_for_socks5_status(
    ips: Sequence[str],
    port: int,
    emit: Callable[[str], None],
    *,
    timeout: int,
    interval: int = SOCKS_VERIFY_INTERVAL,
) -> Tuple[List[str], List[str], Dict[str, str]]:
    """循环等待 SOCKS5 认证状态，直到全部可用或超时。"""
    deadline = time.time() + timeout
    available: List[str] = []
    unavailable: List[str] = []
    reasons: Dict[str, str] = {}
    while time.time() < deadline:
        available, unavailable, reasons = probe_socks5_status(ips, port)
        if not unavailable:
            return available, unavailable, reasons
        reason_preview = "; ".join(f"{ip}={reasons.get(ip, '')}" for ip in unavailable[:5])
        emit(f"SOCKS5认证检查: 可用 {len(available)}/{len(available) + len(unavailable)}，不可用: {', '.join(unavailable)}；{reason_preview}")
        time.sleep(interval)
    return available, unavailable, reasons


def wait_for_port(host: str, port: int, timeout: int, emit: Callable[[str], None]) -> None:
    """等待单个 TCP 端口开放，主要用于 SSH 端口轮询。"""
    deadline = time.time() + timeout
    last_error = ""
    while time.time() < deadline:
        try:
            with socket.create_connection((host, port), timeout=3):
                return
        except OSError as exc:
            last_error = str(exc)
            emit("等待 SSH 端口开放...")
            time.sleep(SSH_PORT_POLL_SECONDS)
    raise TimeoutError(f"SSH端口等待超时: {last_error}")


def probe_ssh_banner(host: str, port: int, timeout: int) -> Tuple[bool, str]:
    """探测 SSH 端口是否开放，并尽量读取 SSH banner。"""
    try:
        with socket.create_connection((host, port), timeout=timeout) as sock:
            sock.settimeout(timeout)
            try:
                banner = sock.recv(64)
            except socket.timeout:
                return True, "端口已通，直接尝试SSH登录"
            if banner.startswith(b"SSH-"):
                return True, banner.decode("ascii", errors="replace").strip()
            return False, f"端口已通但不是SSH banner: {banner[:24]!r}"
    except OSError as exc:
        return False, str(exc)


def unique_public_hosts(hosts: Sequence[str]) -> List[str]:
    """过滤并保留不重复的公网 IP。"""
    unique_hosts: List[str] = []
    for host in hosts:
        candidate = str(host or "").strip()
        if is_public_ip(candidate) and candidate not in unique_hosts:
            unique_hosts.append(candidate)
    return unique_hosts


def wait_for_ssh_hosts(
    hosts: Sequence[str],
    port: int,
    emit: Callable[[str], None],
    *,
    timeout: int = SSH_PORT_WAIT_TIMEOUT,
) -> List[str]:
    """并发等待多个公网 IP 的 SSH 端口，返回已就绪入口列表。"""
    unique_hosts = unique_public_hosts(hosts)
    if not unique_hosts:
        raise TimeoutError("没有可用于SSH探测的公网IP。")

    deadline = time.time() + timeout
    last_errors: Dict[str, str] = {}
    emit(f"SSH入口探测: {', '.join(unique_hosts)}")
    while time.time() < deadline:
        ready: Dict[str, str] = {}
        with concurrent.futures.ThreadPoolExecutor(max_workers=min(len(unique_hosts), 10)) as executor:
            futures = {
                executor.submit(probe_ssh_banner, host, port, SSH_PORT_PROBE_CONNECT_TIMEOUT): host
                for host in unique_hosts
            }
            for future in concurrent.futures.as_completed(futures):
                host = futures[future]
                ok, reason = future.result()
                if ok:
                    ready[host] = reason
                    continue
                last_errors[host] = reason
        if ready:
            ordered_ready = [host for host in unique_hosts if host in ready]
            for host in ordered_ready:
                emit(f"SSH入口确认: {host}:{port} {ready[host]}")
            return ordered_ready
        preview = "；".join(f"{host}={last_errors.get(host, '')}" for host in unique_hosts[:3])
        emit(f"SSH端口暂未就绪，{SSH_PORT_PROBE_INTERVAL}秒后重试；{preview}")
        time.sleep(SSH_PORT_PROBE_INTERVAL)
    detail = "；".join(f"{host}={last_errors.get(host, '')}" for host in unique_hosts)
    raise TimeoutError(f"SSH端口等待超时({timeout}秒): {detail}")


def wait_for_any_ssh_port(
    hosts: Sequence[str],
    port: int,
    emit: Callable[[str], None],
    *,
    timeout: int = SSH_PORT_WAIT_TIMEOUT,
) -> str:
    """等待任意一个 SSH 入口可用，并返回第一个已就绪 IP。"""
    return wait_for_ssh_hosts(hosts, port, emit, timeout=timeout)[0]


def is_network_transient_error(exc: BaseException) -> bool:
    """判断异常是否像临时网络抖动，供华为云和火山云重试逻辑共用。"""
    text = str(exc).lower()
    return any(
        token in text
        for token in (
            "网络连接失败",
            "winerror 10060",
            "timed out",
            "timeout",
            "connection reset",
            "connection aborted",
            "10053",
            "10054",
            "remote end closed",
            "temporarily unavailable",
            "urlopen error",
        )
    )
