scanner.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Clean IP Scanner — Xray Core (Windows-friendly)

Wizard mode:
    py -3.12 .\\scanner.py

CLI mode:
    py -3.12 .\\scanner.py -cf config.txt -rf .\\ranges\\cloudflare_ipv4.txt -n 200 -w 12
"""
import os, sys, json, time, random, socket, signal, zipfile, platform, argparse, tempfile, threading, subprocess, ipaddress, shutil, urllib.request, hashlib, csv, re, ssl, math, statistics, atexit
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from typing import List, Tuple
from datetime import datetime


try:
    import requests
    HAS_REQUESTS = True
except ImportError:
    HAS_REQUESTS = False
    requests = None

try:
    import socks  # noqa
    HAS_SOCKS = True
except ImportError:
    HAS_SOCKS = False

def _check_deps():
    if not HAS_REQUESTS:
        print("[!] requests library is required. Please install it.")
        sys.exit(1)
    if not HAS_SOCKS:
        print("[!] PySocks library is required for SOCKS support. Please install it.")
        sys.exit(1)



try:
    import rich  # noqa: F401
    from rich.console import Console
    from rich.table import Table
    from rich.live import Live
    from rich import box
    from rich.panel import Panel
    from rich.align import Align
    from rich.text import Text
    from rich.layout import Layout
    from rich.progress import Progress, SpinnerColumn, BarColumn, TimeRemainingColumn
    from rich.spinner import Spinner
    HAS_RICH = True
except ImportError:
    HAS_RICH = False
    Console = None
    Table = None
    Live = None
    box = None
    Panel = None
    Align = None
    Text = None
    Layout = None
except Exception:
    # Rich is installed but something went wrong at import-time.
    # Fall back gracefully instead of crashing.
    HAS_RICH = False
    Console = None
    Table = None
    Live = None
    box = None
    Panel = None
    Align = None
    Text = None
    Layout = None


DEFAULT_PORTS = [443]
CONCURRENCY = 8
TIMEOUT = 5
MAX_LATENCY = 500
SCAN_COUNT = 100
DOWNLOAD_TEST_SIZE = 100  # KB
UPLOAD_TEST_SIZE = 50     # KB
SPEED_TEST_SECONDS = 5.0  # realistic throughput window
SPEED_TEST_CHUNK_KB = 1024
DEFAULT_REAL_DELAY_HOST = "1.1.1.1"
DEFAULT_REAL_DELAY_PORT = 443
DEFAULT_XRAY_WORKERS = 24

DOWNLOAD_TEST_URLS = [
    "https://speed.cloudflare.com/__down?bytes={}",
    "https://speed.hetzner.de/1MB.bin",
    "https://cachefly.cachefly.net/1mb.test",
]
UPLOAD_TEST_URLS = [
    "https://speed.cloudflare.com/__up",
    "https://httpbin.org/post",
]

# Default multi-endpoint latency URLs (can be overridden via CLI)
DEFAULT_LATENCY_URLS = [
    "https://cp.cloudflare.com/generate_204",
    "https://www.google.com/generate_204",
    "https://www.cloudflare.com/cdn-cgi/trace",
]

RANGES_DIR = "ranges"
CACHE_FILE = "cache.json"
LOG_DIR = "logs"

class Colors:
    RESET="\033[0m"; BOLD="\033[1m"; DIM="\033[2m"
    RED="\033[91m"; GREEN="\033[92m"; YELLOW="\033[93m"; BLUE="\033[94m"; MAGENTA="\033[95m"; CYAN="\033[96m"; WHITE="\033[97m"
    @staticmethod
    def disable():
        for a in dir(Colors):
            if a.isupper() and not a.startswith("_"):
                setattr(Colors, a, "")
if sys.platform == "win32":
    os.system("color")


class PortManager:
    """Thread-safe port allocation for Xray instances (range 10000-60000)."""
    _PORT_START = 10000
    _PORT_END = 60000
    _lock = threading.Lock()
    _used_ports = set()
    
    @classmethod
    def acquire_port(cls) -> int:
        """Allocate an unused port in the range [10000, 60000)."""
        with cls._lock:
            # Start from random offset instead of always port 10000
            # This avoids "stuck" ports if previous session left zombie processes
            start = random.randint(cls._PORT_START, cls._PORT_END - 2000)
            # Try from random start, then wrap around if needed
            for offset in range(cls._PORT_END - cls._PORT_START):
                port = start + offset
                if port >= cls._PORT_END:
                    port = cls._PORT_START + (port - cls._PORT_END)
                if port not in cls._used_ports:
                    cls._used_ports.add(port)
                    return port
            raise RuntimeError("No ports available in range 10000-60000 (all exhausted)")
    
    @classmethod
    def release_port(cls, port: int) -> None:
        """Release a port back to the pool."""
        with cls._lock:
            cls._used_ports.discard(port)
    
    @classmethod
    def reset(cls) -> None:
        """Reset all port allocations (for cleanup)."""
        with cls._lock:
            cls._used_ports.clear()


# Global process tracking and cleanup
_global_active_procs = set()
_global_proc_lock = threading.Lock()

def _register_global_proc(proc):
    """Register a process for automatic cleanup."""
    if proc:
        with _global_proc_lock:
            _global_active_procs.add(proc)

def _cleanup_all_processes():
    """Kill all registered processes immediately upon exit."""
    with _global_proc_lock:
        procs = list(_global_active_procs)
    for p in procs:
        try:
            if p and p.poll() is None:
                p.terminate()
        except Exception:
            pass
    for p in procs:
        try:
            if p and p.poll() is None:
                p.kill()
        except Exception:
            pass
    with _global_proc_lock:
        _global_active_procs.clear()

# Register cleanup to run at exit
atexit.register(_cleanup_all_processes)


class LiveTUI:
    """
    Simple htop-like live table. Cross-platform best-effort.
    Keys:
      s = sort by score
      l = sort by latency median
      j = sort by jitter
      o = sort by loss
      t = sort by TLS
      d = sort by DL
      u = sort by UL
      q = quit TUI (scan continues, but stops drawing)
    """
    def __init__(self, title:str="Clean IP Scanner", sort_key:str="score", top:int=20, scan_order:str="random"):
        self.title = title
        self.sort_key = sort_key
        self.scan_order = scan_order or "random"
        self.top = top
        self._lock = threading.Lock()
        self._items = []
        self._stop = False
        self._quit_draw = False
        self._status_line = ""
        self._thread = None

    def start(self):
        self._thread=threading.Thread(target=self._loop, daemon=True)
        self._thread.start()

    def stop(self):
        self._stop=True
        if self._thread:
            try: self._thread.join(timeout=1.0)
            except Exception: pass

    def update(self, items:list, status_line:str=""):
        with self._lock:
            self._items=list(items)
            self._status_line=status_line

    def _get_key(self):
        # Windows non-blocking
        try:
            if os.name=="nt":
                import msvcrt
                if msvcrt.kbhit():
                    ch=msvcrt.getch()
                    try: return ch.decode("utf-8", errors="ignore")
                    except Exception: return ""
            else:
                import sys, select
                dr,_,_=select.select([sys.stdin],[],[],0)
                if dr:
                    return sys.stdin.read(1)
        except Exception:
            return ""
        return ""

    def _sort_items(self, arr):
        k=self.sort_key
        def val(r):
            if k=="lat":
                return getattr(r,"latency_median",999999)
            if k=="jitter":
                return getattr(r,"jitter_stddev",999999)
            if k=="loss":
                return getattr(r,"loss",999999)
            if k=="tls":
                return getattr(r,"tls_time",999999)
            if k=="dl":
                return -getattr(r,"download_speed",0.0)
            if k=="ul":
                return -getattr(r,"upload_speed",0.0)
            return getattr(r,"score",999999)
        return sorted(arr, key=val)

    def _render(self):
        with self._lock:
            items = list(self._items)
            status = self._status_line
        items = self._sort_items([r for r in items if str(getattr(r, "status", "")).lower() == "clean"])[:self.top]
        clear_console()
        # Boxed header
        box_width = 86
        sys.stdout.write("┌" + "─" * (box_width - 2) + "┐\n")
        sys.stdout.write(f"│ 🔎 Clean IP Scanner Demo | Runtime: {int(time.time() - getattr(self, '_start_time', time.time()))}s".ljust(box_width - 1) + "│\n")
        sys.stdout.write(f"│ ✅ {len(items)} ❌ 0 ⏱ 0 ⚠ 0 │ Total: {len(items)}/{len(items)}".ljust(box_width - 1) + "│\n")
        sys.stdout.write("├" + "─" * (box_width - 2) + "┤\n")
        # Table header
        sys.stdout.write(f"│ {'IP':<18} {'Delay':>6} {'JIT':>6} {'LOSS':>6} {'TLS':>6} {'DL':>6} {'UL':>6} {'SCORE':>7} {'CDN':<8} {'H2':>2} {'H3':>2} │\n")
        sys.stdout.write("├" + "─" * (box_width - 2) + "┤\n")
        # Table rows
        # Star scoring system
        scores = [getattr(r, 'score', 999999) for r in items]
        if scores:
            min_score = min(scores)
            max_score = max(scores)
            if max_score == min_score:
                max_score = min_score + 1
        else:
            min_score = 0
            max_score = 1

        def stars_for(score):
            # 6 stars for best, 1 for worst
            if max_score == min_score:
                return '⭐️' * 6
            norm = (score - min_score) / (max_score - min_score)
            n_stars = 6 - int(norm * 5 + 0.5)
            n_stars = max(1, min(6, n_stars))
            return '⭐️' * n_stars

        # Compose all IP rows and pretty info into the box
        box_rows = []
        for r in items:
            ip = f"{r.ip}:{r.port}"
            cdn = (getattr(r, "cdn_vendor", "") or "")
            if getattr(r, "cdn_ok", True) is False:
                cdn = "FAKE"
            row = f"│ {ip:<18} {getattr(r,'real_delay_ms',0.0):6.1f} {r.jitter_stddev:6.1f} {r.loss:6.1f} {r.tls_time:6.1f} {r.download_speed:6.2f} {r.upload_speed:6.2f} {r.score:7.1f} {cdn:<8} {('Y' if getattr(r,'h2',False) else '-'):>2} {('Y' if getattr(r,'h3',False) else '-'):>2} │ {stars_for(getattr(r, 'score', 999999))}"
            box_rows.append(row)
            # Add pretty info as extra lines inside the box, indented
            box_rows.append(f"│   {ip}  {stars_for(getattr(r, 'score', 999999))}".ljust(box_width - 1) + "│")
            box_rows.append(f"│     Delay: {getattr(r,'real_delay_ms',0.0):.1f} ms".ljust(box_width - 1) + "│")
            box_rows.append(f"│     JIT: {r.jitter_stddev:.1f} ms".ljust(box_width - 1) + "│")
            box_rows.append(f"│     LOSS: {r.loss:.1f} %".ljust(box_width - 1) + "│")
            box_rows.append(f"│     TLS: {r.tls_time:.1f} ms".ljust(box_width - 1) + "│")
            box_rows.append(f"│     DL: {r.download_speed:.2f} Mbps".ljust(box_width - 1) + "│")
            box_rows.append(f"│     UL: {r.upload_speed:.2f} Mbps".ljust(box_width - 1) + "│")
            box_rows.append(f"│     SCORE: {r.score:.1f}".ljust(box_width - 1) + "│")
            box_rows.append(f"│     CDN: {getattr(r, 'cdn_vendor', '')}".ljust(box_width - 1) + "│")
            box_rows.append(f"│     H2: {'Y' if getattr(r,'h2',False) else '-'}".ljust(box_width - 1) + "│")
            box_rows.append(f"│     H3: {'Y' if getattr(r,'h3',False) else '-'}".ljust(box_width - 1) + "│")
            box_rows.append(f"│".ljust(box_width - 1) + "│")
        for row in box_rows:
            sys.stdout.write(row + "\n")
        sys.stdout.write("└" + "─" * (box_width - 2) + "┘\n")
        sys.stdout.write(status + "\n")
        sys.stdout.flush()

    def _loop(self):
        # Ensure VT sequences on Windows terminals
        try:
            if os.name=="nt":
                os.system("")
        except Exception:
            pass
        while not self._stop:
            ch=self._get_key()
            if ch:
                ch=ch.lower()
                if ch=="q":
                    self._quit_draw=True
                elif ch=="s":
                    self.sort_key="score"
                elif ch=="l":
                    self.sort_key="lat"
                elif ch=="j":
                    self.sort_key="jitter"
                elif ch=="o":
                    self.sort_key="loss"
                elif ch=="t":
                    self.sort_key="tls"
                elif ch=="d":
                    self.sort_key="dl"
                elif ch=="u":
                    self.sort_key="ul"
            if not self._quit_draw:
                self._render()
            time.sleep(0.4)

def clear_console():
    """Clear terminal screen (best-effort)."""
    try:
        # ANSI clear + cursor home (works on most modern terminals, including Windows Terminal/PowerShell)
        print("\x1b[2J\x1b[H", end="")
    except Exception:
        pass
    try:
        import os
        if os.name == "nt":
            os.system("cls")
        else:
            os.system("clear")
    except Exception:
        pass

def banner():
    # Big colorful 'exelite' splash
    logo_lines = [
        r"███████╗██╗  ██╗███████╗██╗     ██╗████████╗███████╗",
        r"██╔════╝╚██╗██╔╝██╔════╝██║     ██║╚══██╔══╝██╔════╝",
        r"█████╗   ╚███╔╝ █████╗  ██║     ██║   ██║   █████╗  ",
        r"██╔══╝   ██╔██╗ ██╔══╝  ██║     ██║   ██║   ██╔══╝  ",
        r"███████╗██╔╝ ██╗███████╗███████╗██║   ██║   ███████╗",
        r"╚══════╝╚═╝  ╚═╝╚══════╝╚══════╝╚═╝   ╚═╝   ╚══════╝",
    ]
    splash_colors = [Colors.MAGENTA, Colors.CYAN, Colors.BLUE, Colors.GREEN, Colors.YELLOW, Colors.RED]
    for i, ln in enumerate(logo_lines):
        print(splash_colors[i % len(splash_colors)] + Colors.BOLD + ln + Colors.RESET)
    print()
    print(f"""{Colors.CYAN}{Colors.BOLD}
╔══════════════════════════════════════════════════════════════╗
║                 Clean IP Scanner — Xray Core                ║
║                 Wizard + CLI (Windows Ready)                ║
╠══════════════════════════════════════════════════════════════╣
║  Tests:  Latency  |  TLS  |  Download  |  Upload            ║
╚══════════════════════════════════════════════════════════════╝
{Colors.RESET}""")

def info(m): print(f"  {Colors.CYAN}▶{Colors.RESET} {m}")
def ok(m):   print(f"  {Colors.GREEN}✓{Colors.RESET} {m}")
def warn(m): print(f"  {Colors.YELLOW}⚠{Colors.RESET} {m}")
def err(m):  print(f"  {Colors.RED}✗{Colors.RESET} {m}")
def section(t):
    """Print a section header (skip if Rich TUI is active to avoid layout disruption)."""
    # Only print if we're not running with Rich TUI (to avoid console clutter)
    w=62
    print(f"\n  {Colors.BLUE}{'─'*w}{Colors.RESET}")
    print(f"  {Colors.BOLD}{Colors.BLUE}  {t}{Colors.RESET}")
    print(f"  {Colors.BLUE}{'─'*w}{Colors.RESET}")

def finished():
    """Display a large 'Finished' banner (no colors)."""
    print("\n")
    print("╔══════════════════════════════════════════════════════════════════════════╗")
    print("║                                                                          ║")
    print("║             ███████╗██╗███╗   ██╗██╗███████╗██╗  ██╗███████╗             ║")
    print("║             ██╔════╝██║████╗  ██║██║██╔════╝██║  ██║██╔════╝             ║")
    print("║             █████╗  ██║██╔██╗ ██║██║███████╗███████║█████╗               ║")
    print("║             ██╔══╝  ██║██║╚██╗██║██║╚════██║██╔══██║██╔══╝               ║")
    print("║             ██║     ██║██║ ╚████║██║███████║██║  ██║███████╗             ║")
    print("║             ╚═╝     ╚═╝╚═╝  ╚═══╝╚═╝╚══════╝╚═╝  ╚═╝╚══════╝             ║")
    print("║                                                                          ║")
    print("╚══════════════════════════════════════════════════════════════════════════╝")
    print("\n")

@dataclass
class ScanResult:
    ip:str; port:int
    # Aggregated latency stats across endpoints (ms)
    latency_min:float=0.0
    latency_mean:float=0.0
    latency_median:float=0.0
    jitter_stddev:float=0.0
    # Back-compat quick fields
    latency:float=0.0
    download_speed:float=0.0
    upload_speed:float=0.0
    download_stddev:float=0.0
    upload_stddev:float=0.0
    speed_rounds:int=1
    loss:float=0.0
    tls_time:float=0.0
    samples:int=0
    endpoints_ok:int=0
    endpoints_total:int=0
    endpoint_breakdown:dict=None
    # Extra checks
    icmp_loss:float=0.0
    tcp_loss:float=0.0
    cdn_ok:bool=True
    cdn_vendor:str=""
    h2:bool=False
    h3:bool=False
    real_delay_ms:float=0.0
    stability_uptime:float=0.0
    stability_variance_ms:float=0.0
    stability_spikes:int=0
    stability_traffic_mbps:float=0.0
    geo:dict=None
    status:str="pending"; region:str=""; error_msg:str=""
    _phase:str="pending"  # Track current scan phase: latency, tls, dl, ul, finished
    @property
    def score(self)->float:
        """Lower is better. Combines latency/jitter/loss/TLS/stability/speed and speed consistency."""
        if self.status != "clean":
            return 99999.0

        lat = float(self.latency_median or self.latency or 99999.0)
        jit = float(self.jitter_stddev or 0.0)
        loss = float(self.loss or 0.0)
        tls = float(self.tls_time or 0.0)
        rtt = float(self.real_delay_ms or 0.0)

        dl = float(self.download_speed or 0.0)
        ul = float(self.upload_speed or 0.0)
        dl_std = float(self.download_stddev or 0.0)
        ul_std = float(self.upload_stddev or 0.0)

        # Stability: higher uptime is better; variance/spikes are worse
        uptime = float(self.stability_uptime or 0.0)  # percent
        var_ms = float(self.stability_variance_ms or 0.0)
        spikes = float(self.stability_spikes or 0.0)

        # Normalize / weights
        # Base penalties (ms-scale)
        score = 0.0
        score += lat * 2.2
        score += jit * 1.4
        score += loss * 35.0
        score += tls * 0.8
        if rtt > 0:
            score += rtt * 1.1

        # Speed rewards (Mbps) -> subtract; cap so it doesn't dominate
        score -= min(dl, 50.0) * 2.0
        score -= min(ul, 20.0) * 1.2

        # **CONSISTENCY PENALTIES** (higher stddev worse) - EMPHASIZE THIS
        # Penalize speed variance heavily - we want consistent IPs
        score += dl_std * 5.0  # Increased from 3.0
        score += ul_std * 4.0  # Increased from 2.0
        
        # Variance ratio penalty: if stddev is > 50% of mean, heavily penalize
        if dl > 0 and dl_std > dl * 0.5:
            score += (dl_std / dl) * 10.0
        if ul > 0 and ul_std > ul * 0.5:
            score += (ul_std / ul) * 8.0

        # Stability (if ran): reward uptime, penalize variance/spikes
        if uptime > 0:
            score -= uptime * 0.8  # Increased reward for good uptime
            score += var_ms * 1.2  # Increased penalty for latency variance
            score += spikes * 10.0  # Increased penalty for spikes
        if self.stability_traffic_mbps > 0:
            score -= min(float(self.stability_traffic_mbps), 40.0) * 0.7

        # Keep score in a sensible range
        if score < 0:
            score = 0.0
        return float(round(score, 2))

class XrayManager:
    XRAY_VERSION="1.8.24"
    XRAY_RELEASES_URL="https://github.com/XTLS/Xray-core/releases/download"
    def __init__(self, work_dir=None):
        self.work_dir = work_dir or os.path.join(tempfile.gettempdir(),"clean_ip_scanner")
        os.makedirs(self.work_dir, exist_ok=True)
        self.xray_path = os.path.join(self.work_dir, "xray.exe" if platform.system().lower()=="windows" else "xray")
    def is_installed(self)->bool:
        return os.path.isfile(self.xray_path) and os.access(self.xray_path, os.X_OK)
    def _platform_zip(self)->str:
        sysn=platform.system().lower(); mach=platform.machine().lower()
        if sysn=="windows":
            return "Xray-windows-64.zip" if mach in ("amd64","x86_64") else "Xray-windows-32.zip"
        if sysn=="linux": return "Xray-linux-64.zip"
        if sysn=="darwin": return "Xray-macos-64.zip"
        return "Xray-linux-64.zip"
    def download_xray(self)->bool:
        fn=self._platform_zip()
        url=f"{self.XRAY_RELEASES_URL}/v{self.XRAY_VERSION}/{fn}"
        zpath=os.path.join(self.work_dir, fn)
        section("Downloading Xray Core")
        info(f"Version: {self.XRAY_VERSION}")
        info(f"URL: {url}")
        try:
            req=urllib.request.Request(url, headers={"User-Agent":"Mozilla/5.0"})
            with urllib.request.urlopen(req, timeout=60) as r:
                total=int(r.headers.get("Content-Length",0))
                got=0
                with open(zpath,"wb") as f:
                    while True:
                        ch=r.read(8192)
                        if not ch: break
                        f.write(ch); got+=len(ch)
                        if total>0:
                            pct=got/total*100
                            bar_len=30
                            filled=int(bar_len*got/total)
                            bar="█"*filled+"░"*(bar_len-filled)
                            sys.stdout.write(f"\r  ⏳ [{bar}] {pct:5.1f}%")
                            sys.stdout.flush()
            print()
            info("Extracting...")
            with zipfile.ZipFile(zpath,"r") as zf:
                zf.extractall(self.work_dir)
            if platform.system().lower()!="windows":
                os.chmod(self.xray_path,0o755)
            os.remove(zpath)
            ok(f"Xray installed: {self.xray_path}")
            return True
        except Exception as e:
            err(f"Failed to download xray-core: {e}")
            return False
    def ensure_xray(self)->bool:
        p=shutil.which("xray")
        if p:
            self.xray_path=p; ok(f"Found xray in PATH: {p}"); return True

        # Check for a user-provided Xray binary bundled next to the script/package (preferred)
        # e.g. copy xray.exe from v2rayN here if you need newer transports like xhttp.
        script_dir = os.path.dirname(os.path.abspath(__file__))
        candidates = [
            os.path.join(script_dir, "xray.exe"),
            os.path.join(script_dir, "xray", "xray.exe"),
            os.path.join(script_dir, "bin", "xray.exe"),
            os.path.join(script_dir, "xray-core", "xray.exe"),
        ]
        for c in candidates:
            if os.path.isfile(c):
                self.xray_path = c
                ok(f"Using bundled/user-provided xray: {c}")
                return True

        if self.is_installed():
            ok(f"Found xray: {self.xray_path}"); return True
        warn("Xray not found — downloading...")
        return self.download_xray()

    def parse_config(self, config_str:str)->dict:
        config_str = (config_str or "").strip()
        # URI shortcuts
        if config_str.startswith("vless://"): return self._parse_vless(config_str)
        if config_str.startswith("vmess://"): return self._parse_vmess(config_str)
        if config_str.startswith("trojan://"): return self._parse_trojan(config_str)

        # If a path to a file was provided, try to load it
        try:
            if os.path.isfile(config_str):
                with open(config_str, "r", encoding="utf-8") as f:
                    txt = f.read().strip()
                    # If file contains a single URI, recurse
                    if txt.startswith(("vless://","vmess://","trojan://")):
                        return self.parse_config(txt)
                    # If file contains JSON, try to parse
                    try:
                        obj = json.loads(txt)
                    except Exception:
                        obj = None
                    if obj:
                        # Expecting full Xray config JSON with outbounds
                        if isinstance(obj, dict) and obj.get("outbounds"):
                            # Use first outbound as parsed representation
                            out = obj["outbounds"][0]
                            return {"outbound": out, "original_host": out.get("settings",{}).get("vnext", [{}])[0].get("address", ""), "original_port": out.get("settings",{}).get("vnext", [{}])[0].get("port", 0)}

        except Exception:
            pass

        # Try raw JSON string (user pasted full xray config)
        if config_str.startswith("{"):
            try:
                obj = json.loads(config_str)
                if isinstance(obj, dict) and obj.get("outbounds"):
                    out = obj["outbounds"][0]
                    return {"outbound": out, "original_host": out.get("settings",{}).get("vnext", [{}])[0].get("address", ""), "original_port": out.get("settings",{}).get("vnext", [{}])[0].get("port", 0)}
            except Exception:
                pass

        # Fallback: not a supported URI or JSON-config
        raise ValueError("Unsupported config. Use vless:// vmess:// trojan:// or pass full Xray JSON config/file")

    def _parse_vless(self, uri:str)->dict:
        import urllib.parse
        uri=uri[8:]
        if "#" in uri: uri,_=uri.rsplit("#",1)
        uuid,rest=uri.split("@",1)
        if "?" in rest:
            host_port,params_str=rest.split("?",1)
            params=dict(urllib.parse.parse_qsl(params_str))
        else:
            host_port,params=rest,{}
        if host_port.startswith("["):
            host,port=host_port.rsplit(":",1); host=host.strip("[]")
        else:
            host,port=host_port.rsplit(":",1)
        port=int(port)
        outbound={"protocol":"vless","settings":{"vnext":[{"address":host,"port":port,"users":[{"id":uuid,"encryption":params.get("encryption","none"),"flow":params.get("flow","")}]}]},"streamSettings":self._build_stream_settings(params,host)}
        if not outbound["settings"]["vnext"][0]["users"][0].get("flow"):
            outbound["settings"]["vnext"][0]["users"][0].pop("flow",None)
        return {"outbound":outbound,"original_host":host,"original_port":port,"sni":params.get("sni",params.get("serverName",host)),"host":params.get("host",host),"protocol":outbound.get("protocol",""),"transport":outbound.get("streamSettings",{}).get("network",""),"security":outbound.get("streamSettings",{}).get("security","none")}

    def _parse_vmess(self, uri:str)->dict:
        import base64
        enc=uri[8:]
        pad=4-len(enc)%4
        if pad!=4: enc+="="*pad
        try:
            obj=json.loads(base64.b64decode(enc).decode("utf-8"))
        except Exception:
            raise ValueError("Invalid VMess config")
        host=obj.get("add",""); port=int(obj.get("port",443))
        uuid=obj.get("id",""); aid=int(obj.get("aid",0))
        net=obj.get("net","tcp"); tls=obj.get("tls",""); sni=obj.get("sni",host)
        ws_host=obj.get("host",host); ws_path=obj.get("path","/")
        params={"type":net,"security":tls if tls else "none","sni":sni,"host":ws_host,"path":ws_path}
        outbound={"protocol":"vmess","settings":{"vnext":[{"address":host,"port":port,"users":[{"id":uuid,"alterId":aid,"security":obj.get("scy","auto")}]}]},"streamSettings":self._build_stream_settings(params,host)}
        return {"outbound":outbound,"original_host":host,"original_port":port,"sni":sni,"host":ws_host}

    def _parse_trojan(self, uri:str)->dict:
        import urllib.parse
        uri=uri[9:]
        if "#" in uri: uri,_=uri.rsplit("#",1)
        password,rest=uri.split("@",1)
        if "?" in rest:
            host_port,params_str=rest.split("?",1)
            params=dict(urllib.parse.parse_qsl(params_str))
        else:
            host_port,params=rest,{}
        if host_port.startswith("["):
            host,port=host_port.rsplit(":",1); host=host.strip("[]")
        else:
            host,port=host_port.rsplit(":",1)
        port=int(port)
        params.setdefault("security","tls")
        outbound={"protocol":"trojan","settings":{"servers":[{"address":host,"port":port,"password":password}]},"streamSettings":self._build_stream_settings(params,host)}
        return {"outbound":outbound,"original_host":host,"original_port":port,"sni":params.get("sni",host),"host":params.get("host",host)}

    def _build_stream_settings(self, params:dict, default_host:str)->dict:
        network=params.get("type", params.get("network","tcp"))
        # Note: transports like "xhttp" require a recent Xray binary. We still parse them and let Xray handle support.
        security=params.get("security","tls")
        sni=params.get("sni",params.get("serverName",default_host))
        alpn=params.get("alpn","h2,http/1.1")
        fp=params.get("fp","chrome")
        host=params.get("host",default_host)
        path=params.get("path","/")
        stream={"network":network}
        if security=="tls":
            stream["security"]="tls"
            stream["tlsSettings"]={"serverName":sni,"allowInsecure":False,"alpn":alpn.split(",") if alpn else ["h2","http/1.1"],"fingerprint":fp}
        elif security=="reality":
            stream["security"]="reality"
            stream["realitySettings"]={"serverName":sni,"fingerprint":fp,"publicKey":params.get("pbk",""),"shortId":params.get("sid",""),"spiderX":params.get("spx","")}
        else:
            stream["security"]="none"
        if network=="ws":
            stream["wsSettings"]={"path":path,"headers":{"Host":host}}
        elif network=="grpc":
            stream["grpcSettings"]={"serviceName":params.get("serviceName",""),"multiMode":params.get("mode","gun")=="multi"}
        elif network=="tcp" and params.get("headerType")=="http":
            stream["tcpSettings"]={"header":{"type":"http","request":{"path":[path],"headers":{"Host":[host]}}}}
        elif network in ("h2","http"):
            stream["httpSettings"]={"host":[host],"path":path}
        elif network == "xhttp":
            xhs = {}
            # Common params: host, path, mode, extra (JSON string)
            if params.get("host"):
                xhs["host"] = params.get("host")
            if params.get("path"):
                xhs["path"] = params.get("path")
            if params.get("mode"):
                xhs["mode"] = params.get("mode")
            extra = params.get("extra")
            if extra:
                try:
                    xhs["extra"] = json.loads(extra)
                except Exception:
                    # If extra is not valid JSON, pass it through as-is
                    xhs["extra"] = extra
            stream["xhttpSettings"] = xhs

        return stream

    def generate_xray_config(self, parsed:dict, target_ip:str, target_port:int, socks_port:int)->dict:
        outbound=json.loads(json.dumps(parsed["outbound"]))
        if "vnext" in outbound.get("settings",{}):
            outbound["settings"]["vnext"][0]["address"]=target_ip
            outbound["settings"]["vnext"][0]["port"]=target_port
        elif "servers" in outbound.get("settings",{}):
            outbound["settings"]["servers"][0]["address"]=target_ip
            outbound["settings"]["servers"][0]["port"]=target_port
        return {"log":{"loglevel":"error"},
                "inbounds":[{"tag":"socks-in","port":socks_port,"listen":"127.0.0.1","protocol":"socks","settings":{"auth":"noauth","udp":False}}],
                "outbounds":[outbound,{"protocol":"freedom","tag":"direct"},{"protocol":"blackhole","tag":"block"}]}

    def start_xray(self, config:dict, instance_id:str):
        cfg_path=os.path.join(self.work_dir, f"config_{instance_id}.json")
        with open(cfg_path,"w",encoding="utf-8") as f:
            json.dump(config,f,indent=2)
        proc=subprocess.Popen([self.xray_path,"run","-c",cfg_path], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
                              creationflags=subprocess.CREATE_NO_WINDOW if sys.platform=="win32" else 0)
        return proc,cfg_path

def _sample_ipv4_hosts(net, k):
    start=int(net.network_address)+1
    end=int(net.broadcast_address)-1
    if end<start: return []
    n=end-start+1
    k=min(k,n)
    picked=set()
    while len(picked)<k:
        picked.add(random.randint(start,end))
    return [str(ipaddress.ip_address(i)) for i in picked]

def _sample_ipv6_hosts(net, k):
    # IPv6 sampling without iterating the full space.
    # We generate random interface identifiers within the prefix.
    if net.prefixlen>=127:
        return [str(net.network_address)]
    host_bits=128-net.prefixlen
    k=max(1,int(k))
    out=set()
    base=int(net.network_address)
    maxv=(1<<host_bits)-1
    for _ in range(min(k*4, 50000)):
        rnd=random.getrandbits(host_bits)
        out.add(base+rnd)
        if len(out)>=k:
            break
    return [str(ipaddress.ip_address(i)) for i in out]

def parse_ip_ranges(text, max_ips=10000, ip_version=4, scan_order:str='random'):
    ips=[]
    for line in text.strip().splitlines():
        line=line.strip()
        if not line or line.startswith("#"): continue
        try:
            if "/" in line:
                net=ipaddress.ip_network(line, strict=False)
                if net.version!=ip_version: continue
                if ip_version==4:
                    ips.extend(_sample_ipv4_hosts(net, 1000))
                else:
                    ips.extend(_sample_ipv6_hosts(net, 400))
            elif "-" in line:
                a,b=[p.strip() for p in line.split("-",1)]
                start_ip=ipaddress.ip_address(a)
                if start_ip.version!=ip_version: continue
                if ip_version==6:
                    # For IPv6 ranges, require full end address: a-b
                    end_ip=ipaddress.ip_address(b)
                else:
                    if "." in b: end_ip=ipaddress.ip_address(b)
                    else:
                        base=str(start_ip).rsplit(".",1)[0]
                        end_ip=ipaddress.ip_address(f"{base}.{b}")
                si,ei=int(start_ip),int(end_ip)
                if ei<si: si,ei=ei,si
                span=ei-si+1
                if span>1000:
                    take=1000 if ip_version==4 else 400
                    # random.sample on huge IPv6 ranges is expensive; use randint.
                    if ip_version==6:
                        picked={random.randint(si,ei) for _ in range(take*2)}
                        picked=list(picked)[:take]
                    else:
                        if scan_order=='ordered':
                            picked=list(range(si, min(ei, si+take-1)+1))
                        else:
                            picked=random.sample(range(si,ei+1), take)
                    ips.extend([str(ipaddress.ip_address(i)) for i in picked])
                else:
                    ips.extend([str(ipaddress.ip_address(i)) for i in range(si,ei+1)])
            else:
                ip=ipaddress.ip_address(line)
                if ip.version==ip_version: ips.append(str(ip))
        except ValueError:
            continue
    seen=set(); clean=[]
    for ip in ips:
        if ip in seen: continue
        seen.add(ip)
        if ip_version==4:
            last=int(ip.split(".")[-1])
            if last not in (0,255): clean.append(ip)
        else:
            clean.append(ip)
    if scan_order!='ordered':
        random.shuffle(clean)
    return clean[:max_ips]


class RichTUI:
    """
    Professional Rich-based Dashboard Layout:
    ┌─────────────────────────────────────────┐
    │  Header (title, runtime, counters)      │
    ├───────────────────────┬─────────────────┤
    │  Results Table (70%)   │ Stats (30%)     │
    │                       │ Yellow border   │
    ├───────────────────────┴─────────────────┤
    │  Footer (progress bar + status)         │
    └─────────────────────────────────────────┘
    """
    def __init__(self, title: str = "Clean IP Scanner", sort_key: str = "score", top: int = 15):
        self.title = title
        self.sort_key = sort_key or "score"
        self.top = top
        self._items = []
        self._status_line = ""
        self._stats = {"clean": 0, "dirty": 0, "timeout": 0, "error": 0, "total": 0, "done": 0}
        self._lock = threading.Lock()
        self._stop = False
        self._live = None
        self._spin_frames = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏"
        self._spin_i = 0
        self._console = Console() if HAS_RICH else None
        self._start_time = None
        self._recent_results = []
        self._current_testing = None  # Track IP currently being tested (ip, port, start_time)
        self._testing_history = []  # Recent IPs being tested

    def start(self):
        self._stop = False
        self._start_time = time.time()

    def stop(self):
        self._stop = True
        try:
            if self._live:
                self._live.stop()
        except Exception:
            pass

    def set_current_testing(self, ip: str, port: int):
        """Mark an IP as currently being tested (for visual indicator)."""
        with self._lock:
            self._current_testing = {"ip": ip, "port": port, "start_time": time.time()}
            self._testing_history.append({"ip": ip, "port": port})
            if len(self._testing_history) > 20:
                self._testing_history.pop(0)

    def clear_current_testing(self):
        """Clear the current testing IP (when test completes)."""
        with self._lock:
            self._current_testing = None

    def _get_breathing_color(self):
        """Get a smoothly breathing color that cycles through blue/cyan gradient."""
        # Create smooth color cycle: blue -> dodger_blue2 -> deep_sky_blue1 -> cyan -> white -> back
        colors = ["blue", "dodger_blue2", "deep_sky_blue1", "cyan", "cyan3", "dodger_blue2", "blue"]
        elapsed = time.time() - self._start_time if self._start_time else 0
        # Slower cycle: complete cycle every 3 seconds
        cycle_position = (elapsed / 3.0) % len(colors)
        color_index = int(cycle_position) % len(colors)
        return colors[color_index]

    def update(self, items: list, status_line: str = "", stats: dict = None):
        if self._stop or not HAS_RICH:
            return
        with self._lock:
            self._items = list(items)
            self._status_line = status_line or ""
            if stats:
                self._stats.update(stats)
            self._spin_i = (self._spin_i + 1) % len(self._spin_frames)
            # Track recent results for sidebar
            if items and (not self._recent_results or items[-1].ip != self._recent_results[-1].ip):
                self._recent_results.append(items[-1])
                if len(self._recent_results) > 15:
                    self._recent_results.pop(0)

        layout = self._render_dashboard()
        if self._live is None:
            self._live = Live(layout, refresh_per_second=15, console=self._console, transient=False)
            self._live.start()
        else:
            try:
                self._live.update(layout)
            except Exception:
                pass

    def _render_dashboard(self):
        """Build the dashboard layout with header, body (table + stats), and footer."""
        layout = Layout()
        layout.split_column(
            Layout(name="header", size=6),
            Layout(name="main", ratio=1),
            Layout(name="footer", size=5)
        )
        layout["header"].update(self._render_header())
        layout["main"].split_row(
            Layout(name="table", ratio=7, minimum_size=40),
            Layout(name="sidebar", ratio=3, minimum_size=20)
        )
        layout["main"]["table"].update(self._render_table())
        layout["main"]["sidebar"].update(self._render_sidebar())
        layout["footer"].update(self._render_footer())
        return layout

    def _render_header(self):
        """Header: Title + Runtime | Clean/Dirty/Timeout/Error Counters."""
        elapsed = int(time.time() - self._start_time) if self._start_time else 0
        with self._lock:
            stats = self._stats.copy()
        
        title_text = Text()
        title_text.append("🔎 ", style="bold white")
        title_text.append("Clean IP Scanner", style="bold cyan")
        title_text.append(f" | Runtime: {elapsed}s", style="cyan")
        
        counters_text = Text()
        counters_text.append("✅ ", style="bold green")
        counters_text.append(f"{stats['clean']}  ", style="green")
        counters_text.append("❌ ", style="bold red")
        counters_text.append(f"{stats['dirty']}  ", style="red")
        counters_text.append("⏱ ", style="bold yellow")
        counters_text.append(f"{stats['timeout']}  ", style="yellow")
        counters_text.append("⚠ ", style="bold magenta")
        counters_text.append(f"{stats['error']}", style="magenta")
        counters_text.append(" │ ", style="dim")
        counters_text.append(f"Total: {stats['done']}/{stats['total']}", style="cyan")
        
        content = Text.assemble(title_text, "\n", counters_text)
        return Panel(
            Align.center(content),
            border_style=self._get_breathing_color(),
            box=box.ROUNDED,
            padding=(0, 2)
        )

    def _render_table(self):
        """Main results table with IP, Delay, Jitter, Loss, DL, UL, Score."""
        with self._lock:
            items = list(self._items)
            status_line = self._status_line  # Capture status to check stage

        table = Table(
            title="🏆 Top Results (All Clean IPs)",
            box=box.ROUNDED,
            header_style="bold cyan",
            border_style="blue",
            expand=True,
            padding=(0, 1)
        )
        
        # Check if we're in Stage 1 (latency-only testing)
        is_stage_1 = "Stage 1" in status_line
        
        table.add_column("IP", style="cyan", no_wrap=True)
        table.add_column("Delay", justify="right", width=7, style="magenta")
        table.add_column("JIT", justify="right", width=6)
        table.add_column("LOSS%", justify="right", width=7)
        # Only add DL/UL columns if NOT in Stage 1
        if not is_stage_1:
            table.add_column("DL", justify="right", width=8)
            table.add_column("UL", justify="right", width=8)
        table.add_column("SCORE", justify="right", width=8, style="white")

        def safe(v, d=0.0):
            try:
                return float(v) if v else d
            except Exception:
                return d

        rows = [r for r in items if getattr(r, "status", None) is not None]
        rows.sort(key=lambda r: safe(getattr(r, "score", 99999.0)))

        scores = [safe(getattr(r, "score", None)) for r in rows if getattr(r, "status", "").lower() == "clean"]
        smin = min(scores) if scores else 0.0
        smax = max(scores) if scores else 1.0
        if smax == smin:
            smax = smin + 1

        def stars_for(score):
            try:
                sc = safe(score, None)
                if sc is None:
                    return "—"
                norm = (sc - smin) / (smax - smin)
                norm = max(0, min(1, norm))
                n_stars = max(1, int((1 - norm) * 5))
                return "★" * n_stars
            except Exception:
                return "—"

        # Display all clean rows (not limited to self.top)
        clean_rows = [r for r in rows if getattr(r, "status", "").lower() == "clean"]
        for r in clean_rows:
            ip = f"{getattr(r, 'ip', '')}:{getattr(r, 'port', '')}"
            r_lat = safe(getattr(r, "real_delay_ms", None) or getattr(r, "latency_median", None))
            jit = safe(getattr(r, "jitter_stddev", None))
            loss = safe(getattr(r, "loss", 0.0))
            dl = safe(getattr(r, "download_speed", 0.0))
            ul = safe(getattr(r, "upload_speed", 0.0))
            score = safe(getattr(r, "score", 99999.0))

            lat_style = "green" if r_lat < 100 else "yellow" if r_lat < 200 else "red"
            loss_style = "green" if loss < 5 else "yellow" if loss < 15 else "red"

            # Build row based on stage (Stage 1 = no DL/UL)
            if is_stage_1:
                table.add_row(
                    ip,
                    f"[{lat_style}]{r_lat:.1f}ms[/{lat_style}]",
                    f"{jit:.1f}",
                    f"[{loss_style}]{loss:.1f}[/{loss_style}]",
                    f"{score:.1f}",
                )
            else:
                table.add_row(
                    ip,
                    f"[{lat_style}]{r_lat:.1f}ms[/{lat_style}]",
                    f"{jit:.1f}",
                    f"[{loss_style}]{loss:.1f}[/{loss_style}]",
                    f"{dl:.2f}",
                    f"{ul:.2f}",
                    f"{score:.1f}",
                )

        return Panel(table, border_style=self._get_breathing_color(), box=box.ROUNDED)

    def _render_sidebar(self):
        """Right panel: Statistics (yellow border per spec)."""
        def safe(v, d=0.0):
            try:
                return float(v) if v else d
            except Exception:
                return d
        
        with self._lock:
            items = list(self._items)
            recent = list(self._recent_results)
            current = self._current_testing
        
        content = Text()
        
        # Current testing (animated)
        if current:
            spinner_frames = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏"
            spin_idx = int(time.time() * 10) % len(spinner_frames)
            spin_char = spinner_frames[spin_idx]
            content.append(f"⏳ Testing\n", style="bold magenta")
            content.append(f"{spin_char} {current['ip']}:{current.get('port', '?')}\n\n", style="magenta on black")
        
        # Statistics
        content.append("📊 Statistics\n", style="bold yellow")
        clean_items = [r for r in items if getattr(r, "status", "").lower() == "clean"]
        
        if clean_items:
            avg_lat = sum(safe(getattr(r, "real_delay_ms", 0) or getattr(r, "latency_median", 0)) for r in clean_items) / len(clean_items)
            content.append("Avg Delay: ", style="white")
            content.append(f"{avg_lat:.1f}ms\n", style="cyan")
            content.append("Clean IPs: ", style="white")
            content.append(f"{len(clean_items)}\n\n", style="green bold")
        
        # Recent (with highlight for currently testing)
        content.append("🕘 Recent\n", style="bold yellow")
        current_ip = current.get("ip") if current else None
        for r in reversed(recent[-10:]):
            status = getattr(r, "status", "").lower()
            ip = getattr(r, "ip", "?")
            is_current = (ip == current_ip)
            
            # Highlight current testing IP
            if is_current:
                content.append(f"⟳ {ip} (testing)\n", style="bold magenta")
            elif status == "clean":
                content.append(f"✓ {ip}\n", style="green")
            elif status == "dirty":
                content.append(f"✗ {ip}\n", style="red")
            elif status == "timeout":
                content.append(f"⏱ {ip}\n", style="yellow")
            else:
                content.append(f"? {ip}\n", style="dim")
        
        return Panel(content, title="Info", border_style="yellow", box=box.ROUNDED)

    def _render_footer(self):
        """Footer: Status + Progress Bar with ETA."""
        with self._lock:
            status = self._status_line
            stats = self._stats.copy()
            spin = self._spin_frames[self._spin_i % len(self._spin_frames)]

        total = max(stats['total'], 1)
        done = stats['done']
        pct = (done / total) * 100 if total > 0 else 0
        
        bar_width = 40
        filled = int(bar_width * done / total) if total > 0 else 0
        
        # Pulsing status text (fade in/out effect)
        pulse_cycle = (time.time() * 2) % 1.0  # 0.5s pulse
        status_text = status or "Scanning..."
        # Apply brightness based on pulse
        if pulse_cycle < 0.5:
            status_style = "bold white"
        else:
            status_style = "dim white"
        
        progress_line = Text()
        progress_line.append(f"{spin} ", style="cyan bold")
        progress_line.append(status_text, style=status_style)
        progress_line.append("\n")
        
        progress_line.append("[", style="dim")
        progress_line.append("█" * filled, style="cyan")
        progress_line.append("░" * (bar_width - filled), style="dim")
        progress_line.append("] ", style="dim")
        progress_line.append(f"{pct:5.1f}%", style="bold cyan")
        
        elapsed = int(time.time() - self._start_time) if self._start_time else 0
        rate = done / elapsed if elapsed > 0 else 0
        eta = int((total - done) / rate) if rate > 0 and total > done else 0
        progress_line.append(f"  ETA: {eta}s", style="yellow")
        
        return Panel(progress_line, border_style=self._get_breathing_color(), box=box.ROUNDED, padding=(0, 2))


class IPTester:
    def __init__(
        self,
        xray,
        parsed,
        timeout,
        max_latency,
        test_dl,
        test_ul,
        dl_kb,
        ul_kb,
        latency_urls:List[str],
        latency_rounds:int=3,
        speed_rounds:int=3,
        speed_seconds:float=SPEED_TEST_SECONDS,
        speed_kb:int=SPEED_TEST_CHUNK_KB,
        adaptive:bool=True,
        warmup:bool=False,
        cdn_check:bool=False,
        strict_cdn:bool=False,
        icmp_loss:bool=False,
        tcp_loss:bool=False,
        test_h2:bool=False,
        test_h3:bool=False,
        geoip:bool=False,
        ipinfo_token:str=None,
        debug:bool=False,
        ping_count:int=3,
        log_dir:str=LOG_DIR,
        tui_obj=None,
        xray_semaphore=None,
        stop_event:threading.Event|None=None,
        real_delay:bool=True,
        real_delay_host:str=DEFAULT_REAL_DELAY_HOST,
        real_delay_port:int=DEFAULT_REAL_DELAY_PORT,
    ):
        self.xray=xray; self.parsed=parsed; self.timeout=timeout; self.max_latency=max_latency
        self.test_dl=test_dl; self.test_ul=test_ul
        self.dl_bytes=dl_kb*1024; self.ul_bytes=ul_kb*1024
        self.speed_seconds=max(0.0, float(speed_seconds or 0.0))
        self.speed_kb=max(32, int(speed_kb or SPEED_TEST_CHUNK_KB))
        self.speed_chunk_bytes=self.speed_kb*1024
        self.latency_urls=latency_urls or DEFAULT_LATENCY_URLS
        self.latency_rounds=max(1,int(latency_rounds))
        self.speed_rounds=max(1,int(speed_rounds or 1))
        self.adaptive=bool(adaptive)
        self.warmup=warmup
        self.cdn_check=cdn_check
        self.strict_cdn=strict_cdn
        self.icmp_loss_enabled=icmp_loss
        self.tcp_loss_enabled=tcp_loss
        self.test_h2=test_h2
        self.test_h3=test_h3
        self.geoip=geoip
        self.ipinfo_token=ipinfo_token
        self.debug=debug
        self.log_dir=log_dir
        self._tui_obj=tui_obj
        self._xray_semaphore=xray_semaphore
        self._stop_event=stop_event
        self.real_delay_enabled=bool(real_delay)
        self.real_delay_host=(real_delay_host or DEFAULT_REAL_DELAY_HOST).strip()
        self.real_delay_port=int(real_delay_port or DEFAULT_REAL_DELAY_PORT)
        self._proc_lock=threading.Lock()
        self._active_procs=set()
        if self.debug:
            os.makedirs(self.log_dir, exist_ok=True)
        self._lock=threading.Lock()

    def _is_stopping(self)->bool:
        return bool(self._stop_event and self._stop_event.is_set())

    def _reserve_local_port(self)->tuple[socket.socket,int]:
        """Reserve a local port using PortManager (thread-safe)."""
        port = PortManager.acquire_port()
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            s.bind(("127.0.0.1", port))
            return s, port
        except OSError:
            PortManager.release_port(port)
            s.close()
            raise RuntimeError(f"Failed to bind port {port}")


    def _register_proc(self, proc):
        if not proc:
            return
        with self._proc_lock:
            self._active_procs.add(proc)

    def _unregister_proc(self, proc):
        if not proc:
            return
        with self._proc_lock:
            self._active_procs.discard(proc)

    def stop_all_processes(self):
        with self._proc_lock:
            procs=list(self._active_procs)
        for p in procs:
            try:
                if p and p.poll() is None:
                    p.terminate()
            except Exception:
                pass
        for p in procs:
            try:
                if p and p.poll() is None:
                    p.kill()
            except Exception:
                pass

    def _session(self, socks_port):
        _check_deps()
        s = requests.Session()
        s.proxies = {"http": f"socks5h://127.0.0.1:{socks_port}", "https": f"socks5h://127.0.0.1:{socks_port}"}
        return s

    def _write_debug(self, instance_id:str, proc_stderr_bytes:bytes|None=None, extra:str=""):
        if not self.debug:
            return
        try:
            p=os.path.join(self.log_dir, f"{instance_id}.log")
            with open(p, "ab") as f:
                if extra:
                    f.write(extra.encode("utf-8", errors="ignore")+b"\n")
                if proc_stderr_bytes:
                    f.write(proc_stderr_bytes)
        except Exception:
            pass

    def _stat_latency(self, values:List[float])->Tuple[float,float,float,float]:
        # returns min, mean, median, stddev
        if not values:
            return 0.0,0.0,0.0,0.0
        v=sorted(values)
        n=len(v)
        mn=v[0]
        mean=sum(v)/n
        if n%2==1:
            med=v[n//2]
        else:
            med=(v[n//2-1]+v[n//2])/2
        # stddev
        var=sum((x-mean)*(x-mean) for x in v)/n
        std=var**0.5
        return mn,mean,med,std

    def _proxy_tcp_delay(self, socks_port:int, host:str, port:int, count:int=3)->float:
        """Best-effort real tunnel delay via SOCKS TCP connect."""
        if not HAS_SOCKS:
            return 0.0
        vals=[]
        for _ in range(max(1, int(count))):
            if self._is_stopping():
                break
            s=None
            t0=time.time()
            try:
                s=socks.socksocket()
                s.set_proxy(socks.SOCKS5, "127.0.0.1", int(socks_port), rdns=True)
                s.settimeout(max(0.8, float(self.timeout)))
                s.connect((host, int(port)))
                vals.append((time.time()-t0)*1000.0)
            except Exception:
                pass
            finally:
                try:
                    if s:
                        s.close()
                except Exception:
                    pass
        return round(statistics.median(vals), 1) if vals else 0.0

    def _download_chunk_once(self, sess, url_template:str, chunk_bytes:int)->tuple[int,str]:
        headers={"User-Agent":"Mozilla/5.0"}
        url=url_template
        if "{}" in url_template:
            url=url_template.format(int(chunk_bytes))
        else:
            headers["Range"]=f"bytes=0-{max(0, int(chunk_bytes)-1)}"
        try:
            with sess.get(
                url,
                timeout=(self.timeout, self.timeout + 15),
                headers=headers,
                stream=True,
                allow_redirects=True,
            ) as resp:
                resp.raise_for_status()
                got=0
                for ch in resp.iter_content(chunk_size=16384):
                    if self._is_stopping():
                        break
                    if not ch:
                        continue
                    got += len(ch)
                    if got >= chunk_bytes:
                        break
                return got, ""
        except Exception as e:
            return 0, str(e)

    def _duration_download_speed(self, sess, min_seconds:float)->tuple[float,str]:
        start=time.time()
        deadline=start+max(1.0, float(min_seconds))
        total=0
        last_err=""
        chunk=max(int(self.speed_chunk_bytes), int(self.dl_bytes))
        while (time.time()<deadline) and (not self._is_stopping()):
            got_round=False
            for u in DOWNLOAD_TEST_URLS:
                if self._is_stopping():
                    break
                got,err_txt=self._download_chunk_once(sess, u, chunk)
                if got>0:
                    total += got
                    got_round=True
                    break
                if err_txt:
                    last_err=err_txt
            if not got_round:
                break
        dt=time.time()-start
        if dt<=0 or total<=0:
            return 0.0, (last_err or "no-data")
        return (total*8)/(dt*1_000_000), ""

    def _duration_upload_speed(self, sess, min_seconds:float)->tuple[float,str]:
        start=time.time()
        deadline=start+max(1.0, float(min_seconds))
        total=0
        last_err=""
        payload=os.urandom(max(32768, min(int(self.speed_chunk_bytes), max(int(self.ul_bytes), 32768))))
        while (time.time()<deadline) and (not self._is_stopping()):
            sent_round=False
            for u in UPLOAD_TEST_URLS:
                if self._is_stopping():
                    break
                try:
                    headers={"Content-Type":"application/octet-stream","User-Agent":"Mozilla/5.0"}
                    resp=sess.post(u, data=payload, timeout=(self.timeout, self.timeout + 20), headers=headers, allow_redirects=True)
                    resp.raise_for_status()
                    total += len(payload)
                    sent_round=True
                    break
                except Exception as e:
                    last_err=str(e)
                    continue
            if not sent_round:
                break
        dt=time.time()-start
        if dt<=0 or total<=0:
            return 0.0, (last_err or "no-data")
        return (total*8)/(dt*1_000_000), ""

    def _which(self, name:str)->str:
        try:
            return shutil.which(name) or ""
        except Exception:
            return ""

    def _run_cmd(self, cmd:list, timeout:float=8.0)->tuple[int,str,str]:
        try:
            p=subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
            return p.returncode, p.stdout or "", p.stderr or ""
        except Exception as e:
            return 1, "", str(e)

    def _icmp_ping_loss(self, ip:str, count:int=5, timeout_ms:int=1000)->float:
        """Returns loss percentage using system ping (best-effort).
        Note: ICMP may require privileges / be blocked; we treat parse failures as 100% loss.
        """
        if ":" in ip:
            # IPv6 ping flags differ
            cmd = ["ping", "-n", str(count), "-w", str(timeout_ms), "-6", ip] if os.name=="nt" else ["ping","-6","-c",str(count),"-W",str(max(1,timeout_ms//1000)),ip]
        else:
            cmd = ["ping", "-n", str(count), "-w", str(timeout_ms), ip] if os.name=="nt" else ["ping","-c",str(count),"-W",str(max(1,timeout_ms//1000)),ip]
        rc,out,err = self._run_cmd(cmd, timeout=max(3.0, count*(timeout_ms/1000)+2))
        txt = (out + "\n" + err).lower()
        # Windows: "Lost = X (Y% loss)"
        m = re.search(r"(\d+)%\s*loss", txt)
        if m:
            try: return float(m.group(1))
            except Exception: pass
        # Linux: "X% packet loss"
        m = re.search(r"(\d+(?:\.\d+)?)%\s*packet\s*loss", txt)
        if m:
            try: return float(m.group(1))
            except Exception: pass
        # Fallback: if ping failed, assume 100% loss; else 0
        return 100.0 if rc!=0 else 0.0

    def _tcp_syn_loss(self, ip:str, port:int, count:int=5, timeout:float=1.0)->float:
        """Best-effort TCP connect loss % (proxy-less)."""
        fails=0
        for _ in range(count):
            try:
                s=socket.socket(socket.AF_INET6 if ":" in ip else socket.AF_INET, socket.SOCK_STREAM)
                s.settimeout(timeout)
                s.connect((ip, port))
            except Exception:
                fails += 1
            finally:
                try: s.close()
                except Exception: pass
        return round((fails / max(1,count))*100, 1)

    def _curl_supported(self)->bool:
        return bool(self._which("curl"))

    def _curl_head(self, ip:str, port:int, host:str, url_path:str="/", extra_args:list=None, timeout:float=8.0)->tuple[int,str,str]:
        extra_args = extra_args or []
        # Use --resolve to force the host to the candidate IP
        url=f"https://{host}{url_path}"
        resolve=f"{host}:{port}:{ip}"
        cmd=["curl","-sS","-I","--connect-timeout",str(int(max(1,timeout-2))),"--max-time",str(int(timeout)),"--resolve",resolve] + extra_args + [url]
        return self._run_cmd(cmd, timeout=timeout+2)

    def _curl_get(self, ip:str, port:int, host:str, url_path:str, extra_args:list=None, timeout:float=8.0)->tuple[int,str,str]:
        extra_args = extra_args or []
        url=f"https://{host}{url_path}"
        resolve=f"{host}:{port}:{ip}"
        cmd=["curl","-sS","--connect-timeout",str(int(max(1,timeout-2))),"--max-time",str(int(timeout)),"--resolve",resolve] + extra_args + [url]
        return self._run_cmd(cmd, timeout=timeout+2)

    def _detect_cdn(self, ip:str, port:int, host:str)->tuple[bool,str,str]:
        """Returns (ok, vendor, reason). Best-effort."""
        if not self._curl_supported():
            return True, "", "curl_not_found"
        # 1) Cloudflare trace check
        rc,out,err = self._curl_get(ip, port, host, "/cdn-cgi/trace", timeout=self.timeout+3)
        body=(out or "").strip()
        if rc==0 and ("colo=" in body or "warp=" in body or "http=" in body and "ip=" in body):
            return True, "cloudflare", "trace_ok"
        # 2) Header heuristics
        rc2,hdrs,err2 = self._curl_head(ip, port, host, "/", timeout=self.timeout+3)
        h=(hdrs or "").lower()
        if "server: cloudflare" in h or "cf-ray:" in h:
            return True, "cloudflare", "header_ok"
        if "x-served-by:" in h or "via: 1.1 varnish" in h or "fastly" in h:
            return True, "fastly", "header_ok"
        # Unknown / maybe fake
        return False, "", "cdn_not_detected"

    def _http_version_probe(self, ip:str, port:int, host:str, mode:str)->bool:
        """Real HTTP2/HTTP3 capability probes (no guessing).

        - HTTP/2: real TLS ALPN negotiation on TCP.
        - HTTP/3: real signal via Alt-Svc header advertising h3 on HTTPS response.
        """
        try:
            af = socket.AF_INET6 if ":" in ip else socket.AF_INET

            if mode == "h2":
                raw = socket.socket(af, socket.SOCK_STREAM)
                raw.settimeout(max(1.0, self.timeout))
                raw.connect((ip, port))
                ctx = ssl.create_default_context()
                ctx.check_hostname = False
                ctx.verify_mode = ssl.CERT_NONE
                try:
                    ctx.set_alpn_protocols(["h2", "http/1.1"])
                except Exception:
                    pass
                ssock = ctx.wrap_socket(raw, server_hostname=host)
                ssock.do_handshake()
                proto = ""
                try:
                    proto = ssock.selected_alpn_protocol() or ""
                except Exception:
                    proto = ""
                try:
                    ssock.close()
                except Exception:
                    pass
                try:
                    raw.close()
                except Exception:
                    pass
                return proto.lower() == "h2"

            if mode == "h3":
                raw = socket.socket(af, socket.SOCK_STREAM)
                raw.settimeout(max(1.0, self.timeout))
                raw.connect((ip, port))
                ctx = ssl.create_default_context()
                ctx.check_hostname = False
                ctx.verify_mode = ssl.CERT_NONE
                ssock = ctx.wrap_socket(raw, server_hostname=host)
                req = (
                    f"HEAD / HTTP/1.1\r\n"
                    f"Host: {host}\r\n"
                    f"User-Agent: CleanIPScanner\r\n"
                    f"Accept: */*\r\n"
                    f"Connection: close\r\n\r\n"
                ).encode("utf-8", "ignore")
                ssock.sendall(req)
                data = b""
                while b"\r\n\r\n" not in data and len(data) < 65536:
                    chunk = ssock.recv(4096)
                    if not chunk:
                        break
                    data += chunk
                head = data.split(b"\r\n\r\n", 1)[0].decode("iso-8859-1", "ignore").lower()
                try:
                    ssock.close()
                except Exception:
                    pass
                try:
                    raw.close()
                except Exception:
                    pass
                for line in head.split("\r\n"):
                    if line.startswith("alt-svc:"):
                        v = line.split(":", 1)[1]
                        if "h3=" in v or "h3-" in v:
                            return True
                return False

            return False
        except Exception:
            return False

    def _tls_handshake_once(self, ip:str, port:int, host:str, timeout:float)->tuple[bool,float]:
        t0=time.time()
        try:
            af=socket.AF_INET6 if ":" in ip else socket.AF_INET
            raw=socket.socket(af, socket.SOCK_STREAM)
            raw.settimeout(timeout)
            raw.connect((ip, port))
            ctx=ssl.create_default_context()
            ctx.check_hostname=False
            ctx.verify_mode=ssl.CERT_NONE
            ssock=ctx.wrap_socket(raw, server_hostname=host)
            ssock.do_handshake()
            try: ssock.close()
            except Exception: pass
            return True, (time.time()-t0)*1000.0
        except Exception:
            return False, (time.time()-t0)*1000.0
        finally:
            try: raw.close()
            except Exception: pass

    def _start_xray_instance(self, ip:str, port:int, instance_id:str, max_retries:int=8):
        """Start one xray process with port-race retry handling."""
        for _ in range(max(1, int(max_retries))):
            if self._is_stopping():
                return None, None, None, "stopped"
            reserve_sock=None
            cfg_path=None
            proc=None
            try:
                reserve_sock,socks_port=self._reserve_local_port()
                xcfg=self.xray.generate_xray_config(self.parsed, ip, port, socks_port)
                try:
                    reserve_sock.close()
                except Exception:
                    pass
                reserve_sock=None

                proc,cfg_path=self.xray.start_xray(xcfg, instance_id)
                self._register_proc(proc)
                _register_global_proc(proc)  # Global cleanup tracking
                time.sleep(0.35)
                if proc.poll() is None:
                    return proc,cfg_path,socks_port,""

                stderr=proc.stderr.read().decode("utf-8", errors="ignore") if proc.stderr else ""
                low=(stderr or "").lower()
                try:
                    self._unregister_proc(proc)
                    proc.kill()
                except Exception:
                    pass
                try:
                    if cfg_path and os.path.isfile(cfg_path):
                        os.remove(cfg_path)
                except Exception:
                    pass
                if ("address already in use" in low) or ("only one usage of each socket address" in low) or ("bind" in low):
                    continue
                return None,None,None,(stderr[:200] if stderr else "xray exited")
            except Exception as e:
                try:
                    if proc:
                        self._unregister_proc(proc)
                        proc.kill()
                except Exception:
                    pass
                try:
                    if cfg_path and os.path.isfile(cfg_path):
                        os.remove(cfg_path)
                except Exception:
                    pass
                try:
                    if reserve_sock:
                        reserve_sock.close()
                except Exception:
                    pass
                return None,None,None,str(e)
        return None,None,None,"xray bind retry exhausted"

    def stability_check(self, ip:str, port:int, host:str, seconds:int, interval:float=5.0)->dict:
        """Real-traffic stability check through the tunnel (not just TLS handshake)."""
        sem_acquired=False
        proc=None
        cfg_path=None
        try:
            if self._xray_semaphore is not None:
                self._xray_semaphore.acquire()
                sem_acquired=True
            instance_id=f"stability_{ip.replace('.','_').replace(':','_')}_{port}_{random.randint(1000,9999)}"
            proc,cfg_path,socks_port,start_err=self._start_xray_instance(ip, port, instance_id)
            if not proc:
                return {"uptime":0.0,"variance_ms":0.0,"spikes":0,"samples":0,"traffic_mbps":0.0,"error":start_err or "xray start failed"}

            sess=self._session(socks_port)
            if sess is None:
                return {"uptime":0.0,"variance_ms":0.0,"spikes":0,"samples":0,"traffic_mbps":0.0,"error":"missing requests/pysocks"}

            end=time.time()+max(1,int(seconds))
            lat_samples=[]
            traffic_samples=[]
            ok=0
            total=0
            probe_url=(self.latency_urls[0] if self.latency_urls else "https://cp.cloudflare.com/generate_204")
            while (time.time()<end) and (not self._is_stopping()):
                total += 1
                success=False
                probe_ms=0.0
                t0=time.time()
                try:
                    resp=sess.get(probe_url, timeout=self.timeout, allow_redirects=False)
                    if resp.status_code in (200,204,301,302):
                        probe_ms=(time.time()-t0)*1000.0
                        dl_mbps,dl_err=self._duration_download_speed(sess, min(2.0, max(1.0, interval*0.5)))
                        if dl_mbps>0:
                            success=True
                            traffic_samples.append(dl_mbps)
                        elif not dl_err:
                            success=True
                    if success:
                        ok += 1
                        lat_samples.append(probe_ms)
                except Exception:
                    pass
                sleep_left=max(0.2, float(interval))
                t_spent=time.time()-t0
                if sleep_left>t_spent:
                    time.sleep(sleep_left-t_spent)

            uptime=round((ok/max(1,total))*100,1)
            if len(lat_samples)>=2:
                var=statistics.pvariance(lat_samples)
                std=math.sqrt(var)
                med=statistics.median(lat_samples)
                spikes=sum(1 for x in lat_samples if x>med+3*std and x>med*1.8)
                return {
                    "uptime":uptime,
                    "variance_ms":round(var,1),
                    "spikes":int(spikes),
                    "samples":len(lat_samples),
                    "traffic_mbps":round(statistics.mean(traffic_samples),2) if traffic_samples else 0.0,
                }
            return {
                "uptime":uptime,
                "variance_ms":0.0,
                "spikes":0,
                "samples":len(lat_samples),
                "traffic_mbps":round(statistics.mean(traffic_samples),2) if traffic_samples else 0.0,
            }
        finally:
            try:
                if proc:
                    self._unregister_proc(proc)
                    if proc.poll() is None:
                        proc.terminate()
                        proc.wait(timeout=2)
            except Exception:
                try:
                    if proc:
                        proc.kill()
                except Exception:
                    pass
            try:
                if cfg_path and os.path.isfile(cfg_path):
                    os.remove(cfg_path)
            except Exception:
                pass
            if sem_acquired and self._xray_semaphore is not None:
                try:
                    self._xray_semaphore.release()
                except Exception:
                    pass

    def geoip_lookup(self, ip:str)->dict:
        if not self.geoip:
            return {}
        if not self.ipinfo_token:
            return {}
        try:
            import requests
            url=f"https://ipinfo.io/{ip}/json?token={self.ipinfo_token}"
            r=requests.get(url, timeout=5)
            if r.status_code==200:
                d=r.json()
                keep={k:d.get(k) for k in ["ip","hostname","city","region","country","loc","org","postal","timezone","asn","company"] if k in d}
                return keep
        except Exception:
            return {}
        return {}

    def test_ip(self, ip, port):
        r=ScanResult(ip=ip, port=port)
        r._phase="latency"
        safe_ip=ip.replace('.', '_').replace(':', '_')
        instance_id=f"{safe_ip}_{port}_{random.randint(1000,9999)}"
        proc=None
        cfg_path=None
        sess=None
        sem_acquired=False
        try:
            if self._is_stopping():
                r.status="error"
                r.error_msg="Stopped by user"
                return r

            if self._xray_semaphore is not None:
                self._xray_semaphore.acquire()
                sem_acquired=True

            proc,cfg_path,socks_port,start_err=self._start_xray_instance(ip, port, instance_id)
            if not proc:
                r.status="error"
                r.error_msg=f"Xray start failed: {start_err}"
                return r

            sess=self._session(socks_port)
            if sess is None:
                r.status="error"
                r.error_msg="Missing dependency: requests+pysocks"
                return r

            if self.warmup:
                try:
                    for u in self.latency_urls[:2]:
                        if self._is_stopping():
                            break
                        sess.get(u, timeout=self.timeout, allow_redirects=False)
                except Exception:
                    pass

            r._phase="latency"
            all_lat=[]
            endpoint_breakdown={}
            region=""
            ok_count=0
            fail_count=0
            total_attempts=0
            for u in self.latency_urls:
                if self._is_stopping():
                    break
                vals=[]
                succ=0
                for _ in range(self.latency_rounds):
                    if self._is_stopping():
                        break
                    total_attempts += 1
                    try:
                        t0=time.time()
                        resp=sess.get(u, timeout=self.timeout, allow_redirects=False)
                        dt=(time.time()-t0)*1000
                        if resp.status_code in (200,204,301,302):
                            vals.append(dt)
                            all_lat.append(dt)
                            succ += 1
                            ok_count += 1
                            ray=resp.headers.get("cf-ray","")
                            if (not region) and ("-" in ray):
                                region=ray.split("-")[-1]
                        else:
                            fail_count += 1
                    except Exception:
                        fail_count += 1
                endpoint_breakdown[u]={"ok":succ,"total":self.latency_rounds,"min_ms":round(min(vals),1) if vals else 0.0}

            r.endpoint_breakdown=endpoint_breakdown
            r.endpoints_ok=ok_count
            r.endpoints_total=total_attempts
            r.samples=len(all_lat)
            if not all_lat:
                r.status="timeout"
                r.error_msg="Timeout"
                return r

            mn,mean,med,std=self._stat_latency(all_lat)
            r.latency_min=round(mn,1)
            r.latency_mean=round(mean,1)
            r.latency_median=round(med,1)
            r.jitter_stddev=round(std,1)
            r.latency=r.latency_median
            r.loss=round((fail_count/total_attempts)*100,1) if total_attempts else 0.0
            r.region=region

            if self.real_delay_enabled and (not self._is_stopping()):
                # Measure Real Delay by routing real HTTP requests through the Xray SOCKS5 tunnel.
                # Use socks5h to force remote DNS resolution; send 5 requests and take median of successful RTTs.
                try:
                    probe_url = (self.latency_urls[0] if self.latency_urls else "https://cp.cloudflare.com/generate_204")
                    delays = []
                    failures = 0
                    rounds = 5
                    for _ in range(rounds):
                        if self._is_stopping():
                            break
                        try:
                            t0 = time.time()
                            resp = sess.get(probe_url, timeout=min(5, max(3, int(self.timeout))), allow_redirects=False)
                            dt = (time.time() - t0) * 1000.0
                            if resp.status_code in (200, 204, 301, 302):
                                delays.append(dt)
                            else:
                                failures += 1
                        except Exception:
                            failures += 1
                    if delays:
                        r.real_delay_ms = round(statistics.median(delays), 1)
                    else:
                        r.real_delay_ms = 0.0
                    r.real_delay_loss = round((failures / max(1, rounds)) * 100.0, 1)
                except Exception:
                    r.real_delay_ms = 0.0
                    r.real_delay_loss = 100.0

            if r.latency_median > self.max_latency:
                r.status="dirty"
                r.error_msg=f"High median latency: {r.latency_median}ms"
                return r

            try:
                if self.icmp_loss_enabled:
                    r.icmp_loss=self._icmp_ping_loss(ip, count=5, timeout_ms=int(self.timeout*1000))
                if self.tcp_loss_enabled:
                    r.tcp_loss=self._tcp_syn_loss(ip, port, count=5, timeout=max(0.6, self.timeout/2))
            except Exception:
                pass

            if self.cdn_check:
                ok_cdn,vendor,reason=self._detect_cdn(ip, port, self.parsed.get("sni") or self.parsed.get("host") or "")
                r.cdn_ok=bool(ok_cdn)
                r.cdn_vendor=vendor
                if self.strict_cdn and (not ok_cdn):
                    r.status="dirty"
                    r.error_msg=f"CDN not detected ({reason})"
                    return r

            try:
                host_for_probe=self.parsed.get("sni") or self.parsed.get("host") or ""
                if host_for_probe:
                    if self.test_h2:
                        r.h2=bool(self._http_version_probe(ip, port, host_for_probe, "h2"))
                    if self.test_h3:
                        r.h3=bool(self._http_version_probe(ip, port, host_for_probe, "h3"))
            except Exception:
                pass

            try:
                if self.geoip:
                    r.geo=self.geoip_lookup(ip)
            except Exception:
                r.geo={}

            r._phase="tls"
            try:
                t0=time.time()
                sess.get(self.latency_urls[0], timeout=self.timeout, allow_redirects=False)
                r.tls_time=round((time.time()-t0)*1000,1)
            except Exception:
                r.tls_time=r.latency_median*2

            if self.test_dl and (not self._is_stopping()):
                r._phase="dl"
                dl_speeds=[]
                dl_err=""
                skip_speed=False
                if self.adaptive and self.warmup:
                    if (r.latency_median or r.latency) > max(2000, self.max_latency*3) or (r.loss or 0) > 8 or (r.jitter_stddev or 0) > 1200:
                        skip_speed=True
                        dl_err="adaptive-skip"

                if not skip_speed:
                    for _ in range(self.speed_rounds):
                        if self._is_stopping():
                            break
                        if self.speed_seconds > 0:
                            sp,err_txt=self._duration_download_speed(sess, self.speed_seconds)
                            if sp > 0:
                                dl_speeds.append(sp)
                                dl_err=""
                                continue
                            dl_err=err_txt or "download failed"
                            break
                        got_speed=False
                        for u in DOWNLOAD_TEST_URLS:
                            t0=time.time()
                            got,err_txt=self._download_chunk_once(sess, u, self.dl_bytes)
                            dt=time.time()-t0
                            if got > 0 and dt > 0:
                                got_speed=True
                                dl_speeds.append((got*8)/(dt*1_000_000))
                                dl_err=""
                                break
                            if err_txt:
                                dl_err=err_txt
                        if not got_speed:
                            break

                if dl_speeds:
                    r.download_speed=round(statistics.mean(dl_speeds), 2)
                    r.download_stddev=round(statistics.pstdev(dl_speeds) if len(dl_speeds) > 1 else 0.0, 2)
                else:
                    r.download_speed=0.0
                    r.download_stddev=0.0
                r.speed_rounds=int(self.speed_rounds)
                if self.debug and dl_err:
                    self._write_debug(instance_id, extra=f"[DL] {dl_err}")

            if self.test_ul and (not self._is_stopping()):
                r._phase="ul"
                ul_speeds=[]
                ul_err=""
                skip_speed=False
                if self.adaptive and self.warmup:
                    if (r.latency_median or r.latency) > max(2000, self.max_latency*3) or (r.loss or 0) > 8 or (r.jitter_stddev or 0) > 1200:
                        skip_speed=True
                        ul_err="adaptive-skip"

                if not skip_speed:
                    for _ in range(self.speed_rounds):
                        if self._is_stopping():
                            break
                        if self.speed_seconds > 0:
                            sp,err_txt=self._duration_upload_speed(sess, self.speed_seconds)
                            if sp > 0:
                                ul_speeds.append(sp)
                                ul_err=""
                                continue
                            ul_err=err_txt or "upload failed"
                            break
                        payload=os.urandom(self.ul_bytes)
                        got_speed=False
                        for u in UPLOAD_TEST_URLS:
                            try:
                                headers={"Content-Type":"application/octet-stream","User-Agent":"Mozilla/5.0"}
                                t0=time.time()
                                resp=sess.post(u, data=payload, timeout=(self.timeout, self.timeout + 20), headers=headers, allow_redirects=True)
                                resp.raise_for_status()
                                dt=time.time()-t0
                                if dt>0:
                                    ul_speeds.append((len(payload)*8)/(dt*1_000_000))
                                    got_speed=True
                                    ul_err=""
                                    break
                            except Exception as e:
                                ul_err=str(e)
                                continue
                        if not got_speed:
                            break

                if ul_speeds:
                    r.upload_speed=round(statistics.mean(ul_speeds), 2)
                    r.upload_stddev=round(statistics.pstdev(ul_speeds) if len(ul_speeds) > 1 else 0.0, 2)
                else:
                    r.upload_speed=0.0
                    r.upload_stddev=0.0
                r.speed_rounds=int(self.speed_rounds)
                if self.debug and ul_err:
                    self._write_debug(instance_id, extra=f"[UL] {ul_err}")

            if self._is_stopping():
                r.status="error"
                r.error_msg="Stopped by user"
            elif r.latency <= 0:
                r.status="timeout"
            elif r.loss >= 20:
                r.status="dirty"
                r.error_msg=f"Packet loss too high: {r.loss}%"
            elif r.real_delay_ms > 0 and r.real_delay_ms > max(1500, self.max_latency * 1.5):
                r.status="dirty"
                r.error_msg=f"Real tunnel delay too high: {r.real_delay_ms}ms"
            elif self.warmup and self.test_dl and r.download_speed <= 0:
                r.status="dirty"
                r.error_msg="No sustained download throughput"
            else:
                r.status="clean"

            r._phase="finished"
            return r
        except TimeoutError:
            r.status="timeout"
            r.error_msg="Request timeout"
            r._phase="finished"
            return r
        except Exception as e:
            r.status="error"
            r.error_msg=str(e)[:120]
            r._phase="finished"
            return r
        finally:
            try:
                if sess:
                    sess.close()
            except Exception:
                pass
            try:
                if self.debug and proc and proc.stderr:
                    try:
                        b=proc.stderr.read() or b""
                    except Exception:
                        b=b""
                    if b:
                        self._write_debug(instance_id, b)
            except Exception:
                pass
            try:
                if proc:
                    self._unregister_proc(proc)
                    if proc.poll() is None:
                        proc.terminate()
                        proc.wait(timeout=2)
            except Exception:
                try:
                    if proc:
                        proc.kill()
                except Exception:
                    pass
            try:
                if cfg_path and os.path.isfile(cfg_path):
                    os.remove(cfg_path)
            except Exception:
                pass
            if sem_acquired and self._xray_semaphore is not None:
                try:
                    self._xray_semaphore.release()
                except Exception:
                    pass

class CleanIPScanner:
    def __init__(
        self,
        config_str,
        ip_ranges,
        ports,
        workers,
        timeout,
        max_latency,
        count,
        test_dl,
        test_ul,
        dl_kb,
        ul_kb,
        speed_rounds:int=3,
        speed_seconds:float=SPEED_TEST_SECONDS,
        speed_kb:int=SPEED_TEST_CHUNK_KB,
        latency_urls:List[str]=None,
        stage2_top:int=30,
        use_cache:bool=True,
        debug:bool=False,
        ping_count:int=3,
        ip_version:int=4,
        smart_concurrency:bool=True,
        tui:bool=False,
        sort_key:str="score",
        scan_order:str="random",
        cdn_check:bool=False,
        strict_cdn:bool=False,
        icmp_loss:bool=False,
        tcp_loss:bool=False,
        test_h2:bool=False,
        test_h3:bool=False,
        stability:int=0,
        stability_interval:float=5.0,
        stability_top:int=10,
        xray_workers:int|None=None,
        real_delay:bool=True,
        real_delay_host:str=DEFAULT_REAL_DELAY_HOST,
        real_delay_port:int=DEFAULT_REAL_DELAY_PORT,
        geoip:bool=False,
        ipinfo_token:str=None,
    ):
        self.config_str=config_str; self.ip_ranges=ip_ranges; self.ports=ports
        self.workers=max(1,int(workers)); self.timeout=float(timeout); self.max_latency=int(max_latency); self.count=int(count); self.scan_order=str(scan_order or 'random').lower().strip() or 'random'
        if self.scan_order not in ('random','ordered'):
            self.scan_order='random'
        self.test_dl=bool(test_dl); self.test_ul=bool(test_ul); self.dl_kb=int(dl_kb); self.ul_kb=int(ul_kb); self.speed_rounds=int(speed_rounds or 1)
        self.speed_seconds=max(0.0, float(speed_seconds or 0.0))
        self.speed_kb=max(32, int(speed_kb or SPEED_TEST_CHUNK_KB))
        self.latency_urls=latency_urls or DEFAULT_LATENCY_URLS
        self.stage2_top=max(1,int(stage2_top))
        self.use_cache=use_cache
        self.debug=debug
        self.ping_count=int(ping_count) if ping_count else 3
        self.ip_version=ip_version
        self.smart_concurrency=smart_concurrency
        self.tui=tui
        self.sort_key=sort_key
        self.cdn_check=cdn_check
        self.strict_cdn=strict_cdn
        self.icmp_loss=icmp_loss
        self.tcp_loss=tcp_loss
        self.test_h2=test_h2
        self.test_h3=test_h3
        self.stability=int(stability or 0)
        self.stability_interval=float(stability_interval or 5.0)
        self.stability_top=int(stability_top or 10)
        _xw = int(xray_workers) if xray_workers else min(DEFAULT_XRAY_WORKERS, self.workers)
        self.xray_workers=max(1, min(self.workers, _xw))
        self.real_delay=bool(real_delay)
        self.real_delay_host=(real_delay_host or DEFAULT_REAL_DELAY_HOST).strip()
        self.real_delay_port=int(real_delay_port or DEFAULT_REAL_DELAY_PORT)
        self.geoip=geoip
        self.ipinfo_token=ipinfo_token
        self.results=[]
        self.stage1_results=[]
        self._lock=threading.Lock()
        self.stop_flag=False
        self.stop_event=threading.Event()
        self._xray_semaphore=threading.BoundedSemaphore(self.xray_workers)
        self.xray=XrayManager()
        self._cache=self._load_cache() if use_cache else {}

    def _cache_key(self, ip:str, port:int)->str:
        h=hashlib.sha256()
        h.update(self.config_str.encode("utf-8",errors="ignore"))
        h.update(f"|{ip}:{port}|{self.timeout}|{self.max_latency}|{self.dl_kb}|{self.ul_kb}|{self.speed_rounds}|{self.speed_seconds}|{self.speed_kb}|{self.ping_count}|{int(self.test_dl)}|{int(self.test_ul)}|{int(self.real_delay)}|{self.real_delay_host}:{self.real_delay_port}|{','.join(self.latency_urls)}".encode("utf-8"))
        return h.hexdigest()

    def _load_cache(self)->dict:
        try:
            if os.path.isfile(CACHE_FILE):
                with open(CACHE_FILE,"r",encoding="utf-8") as f:
                    obj=json.load(f)
                    return obj if isinstance(obj,dict) else {}
        except Exception:
            pass
        return {}

    def _save_cache(self):
        if not self.use_cache:
            return
        try:
            with open(CACHE_FILE,"w",encoding="utf-8") as f:
                json.dump(self._cache,f,indent=2,ensure_ascii=False)
        except Exception:
            pass

    def _check_deps(self):
        missing=[]
        if not HAS_REQUESTS:
            missing.append("requests")
        if not HAS_SOCKS:
            missing.append("pysocks")
        if not missing:
            return
        section("Dependencies")
        warn(f"Missing packages: {' '.join(missing)}")
        info("Install with: py -3.12 -m pip install requests pysocks")
        raise SystemExit(1)

    def _make_live_progress_panel(self, current_ip=None, port=None, elapsed=0.0, done=0, total=0, clean=0, dirty=0, tout=0, errs=0, eta=0, stage=""):
        """Build a Rich renderable for live progress (spinner + current IP + stats)."""
        if not HAS_RICH:
            return None
        lines = []
        # Spinner + Current IP
        if current_ip:
            spinner = Spinner("dots", text=f"Testing {current_ip}:{port} | Elapsed: {elapsed:.1f}s", style="cyan")
            lines.append(spinner)
        else:
            lines.append(Text("Waiting for tasks...", style="dim"))
        # Progress bar
        bar_len = 30
        filled = int(bar_len * done / total) if total else bar_len
        bar = '█' * filled + '░' * (bar_len - filled)
        pct = (done / total) * 100 if total else 100
        progress_line = f"[{bar}] {pct:5.1f}% ({done}/{total}) | ✓{clean} ✗{dirty} ⏱{tout} ⚠{errs} | ETA: {eta}s"
        lines.append(Text(progress_line, style="green"))
        return Panel(
            "\n".join(str(line) if isinstance(line, str) else str(line.plain) for line in lines),
            title=f"[cyan]{stage}[/cyan]",
            border_style="cyan",
            padding=(1, 2)
        )

    def _print_one(self, res:ScanResult, stage:str=""):
        prefix=(f"[{stage}] " if stage else "")
        if str(res.status).lower()=="clean":
            icon=f"{Colors.GREEN}✓ CLEAN{Colors.RESET}"
            detail=f"Real Delay:{getattr(res,'real_delay_ms',0.0)}ms  mean:{res.latency_mean}ms  jitter:{res.jitter_stddev}ms  loss:{res.loss}%"
            if res.tls_time>0: detail+=f"  TLS:{res.tls_time}ms"
            if res.download_speed>0: detail+=f"  DL:{res.download_speed}Mbps"
            if res.upload_speed>0: detail+=f"  UL:{res.upload_speed}Mbps"
            # region tag removed from line output
        elif res.status=="dirty":
            icon=f"{Colors.RED}✗ DIRTY{Colors.RESET}"
            detail=res.error_msg or f"median:{res.latency_median}ms"
        elif res.status=="timeout":
            icon=f"{Colors.YELLOW}⏱ TIMEOUT{Colors.RESET}"
            detail=res.error_msg or "Timeout"
        else:
            icon=f"{Colors.DIM}⚠ ERROR{Colors.RESET}"
            detail=(res.error_msg or "Unknown")[:80]
        print(f"\n  {prefix}{icon} {res.ip}:{res.port} → {detail}")

    def _scan_tasks(self, tasks:List[Tuple[str,int]], tester:IPTester, stage:str, write_cache:bool=False):
        total=len(tasks)
        start=time.time()
        done=0
        clean=dirty=tout=errs=0
        inflight_limit=max(1,min(self.workers,len(tasks)))
        max_workers=self.workers
        history=[]

        current_ip_info={"ip":None,"port":None,"start_time":None}
        info_lock=threading.Lock()
        scanning_active={"flag":True}

        def adjust_inflight():
            nonlocal inflight_limit
            if not self.smart_concurrency:
                return
            if len(history)<30:
                return
            last=history[-30:]
            tout_ratio=sum(1 for s in last if s=="timeout")/len(last)
            err_ratio=sum(1 for s in last if s=="error")/len(last)
            bad=tout_ratio+err_ratio
            if bad>=0.35 and inflight_limit>1:
                inflight_limit=max(1,int(inflight_limit*0.75))
            elif bad<=0.08 and inflight_limit<max_workers:
                inflight_limit=min(max_workers, inflight_limit+1)

        if not self.tui:  # Skip section print if Rich TUI is active
            section(f"Scanning {stage}")
        live_display=None
        if HAS_RICH and not getattr(self, "_tui_obj", None):
            live_display=Live(refresh_per_second=10, transient=False)
            live_display.start()

        def update_live_loop():
            while scanning_active["flag"]:
                try:
                    if live_display:
                        with info_lock:
                            current_ip=current_ip_info.get("ip")
                            port=current_ip_info.get("port")
                            ip_start=current_ip_info.get("start_time", time.time())
                        ip_elapsed=time.time()-ip_start if ip_start else 0
                        lines=[]
                        elapsed_total=time.time()-start
                        rate=done/elapsed_total if elapsed_total>0 else 0
                        eta=int((total-done)/rate) if rate>0 else 0
                        if current_ip:
                            timeout_indicator=""
                            if ip_elapsed>15:
                                timeout_indicator=" ⚠ (Still working...)"
                            lines.append(f"⏳ Testing: {current_ip}:{port} | Elapsed: {ip_elapsed:.1f}s{timeout_indicator}")
                        pct=(done/total)*100 if total else 100
                        bar_len=28
                        filled=int(bar_len*done/total) if total else bar_len
                        bar="█"*filled + "░"*(bar_len-filled)
                        lines.append(f"[{bar}] {pct:5.1f}%")
                        lines.append(f"Progress: {done}/{total} | ✓{clean} ✗{dirty} ⏱{tout} ⚠{errs}")
                        lines.append(f"Inflight: {inflight_limit} | ETA: {eta}s")
                        content="\n".join(lines)
                        live_display.update(Panel(content, title=f"[cyan]{stage}[/cyan]", border_style="cyan", padding=(0, 1)))
                except Exception:
                    pass
                time.sleep(0.1)

        live_thread=None
        if live_display:
            live_thread=threading.Thread(target=update_live_loop, daemon=True)
            live_thread.start()

        ex=ThreadPoolExecutor(max_workers=max_workers)
        futs={}
        try:
            idx=0
            stop_grace_started=None

            def _bump(res: ScanResult):
                nonlocal done, clean, dirty, tout, errs
                done += 1
                st=(getattr(res, "status", None) or "error").lower()
                if st=="clean":
                    clean += 1
                    history.append("clean")
                elif st=="dirty":
                    dirty += 1
                    history.append("dirty")
                elif st=="timeout":
                    tout += 1
                    history.append("timeout")
                else:
                    errs += 1
                    history.append("error")
                if len(history)>500:
                    del history[:-500]

                pct=(done/total)*100 if total else 100
                bar_len=28
                filled=int(bar_len*done/total) if total else bar_len
                bar="█"*filled + "░"*(bar_len-filled)
                elapsed=time.time()-start
                rate=done/elapsed if elapsed>0 else 0
                eta=int((total-done)/rate) if rate>0 else 0
                status_line=f"[{stage}] {pct:5.1f}% | {done}/{total} | ✓{clean} ✗{dirty} ⏱{tout} ⚠{errs} | inflight:{inflight_limit} | ETA:{eta}s"

                if getattr(self, "_tui_obj", None):
                    try:
                        items=self.results if self.results else (self.stage1_results if self.stage1_results else [])
                        stats={"clean":clean,"dirty":dirty,"timeout":tout,"error":errs,"total":total,"done":done}
                        self._tui_obj.update(items, status_line=status_line, stats=stats)
                        self._tui_obj.clear_current_testing()  # Clear "testing" indicator
                    except Exception:
                        pass
                elif not live_display:
                    sys.stdout.write(f"\r  [{bar}] {pct:5.1f}% | {done}/{total} | ✓{clean} ✗{dirty} ⏱{tout} ⚠{errs} | inflight:{inflight_limit} | ETA:{eta}s  ")
                    sys.stdout.flush()

            def submit_more():
                nonlocal idx
                while idx < len(tasks) and len(futs) < inflight_limit and (not self.stop_flag):
                    ip,port=tasks[idx]
                    idx += 1
                    if self.use_cache:
                        key=self._cache_key(ip,port)
                        cached=self._cache.get(key)
                        if cached and isinstance(cached,dict):
                            _dl=float(cached.get("download_speed") or 0)
                            _ul=float(cached.get("upload_speed") or 0)
                            _rd=float(cached.get("real_delay_ms") or 0)
                            if (self.test_dl and _dl<=0) or (self.test_ul and _ul<=0) or (self.real_delay and _rd<=0):
                                cached=None
                        if cached and isinstance(cached,dict):
                            res=ScanResult(ip=ip,port=port)
                            for k,v in cached.items():
                                setattr(res,k,v)
                            res.ip=ip
                            res.port=port
                            self._append_result(res, stage)
                            _bump(res)
                            continue
                    with info_lock:
                        current_ip_info["ip"]=ip
                        current_ip_info["port"]=port
                        current_ip_info["start_time"]=time.time()
                    if getattr(self, "_tui_obj", None):
                        self._tui_obj.set_current_testing(ip, port)
                    futs[ex.submit(tester.test_ip, ip, port)] = (ip,port)

            submit_more()

            while futs:
                if self.stop_flag:
                    self.stop_event.set()
                    tester.stop_all_processes()
                    if stop_grace_started is None:
                        stop_grace_started=time.time()
                    for f in list(futs.keys()):
                        if f.cancel():
                            ip,port=futs.pop(f)
                            res=ScanResult(ip=ip, port=port, status="error", error_msg="Stopped by user")
                            self._append_result(res, stage)
                            _bump(res)
                    if (not futs) or ((time.time()-stop_grace_started)>8.0):
                        break

                try:
                    fut=next(as_completed(list(futs.keys()), timeout=(0.5 if self.stop_flag else self.timeout+60)))
                except Exception:
                    if (not self.stop_flag) and self.smart_concurrency and inflight_limit>1:
                        inflight_limit=max(1,int(inflight_limit*0.9))
                    continue

                ip,port=futs.pop(fut)
                try:
                    res=fut.result(timeout=max(3.0, self.timeout+30))
                except Exception as e:
                    res=ScanResult(ip=ip, port=port, status="error", error_msg=str(e))

                self._append_result(res, stage)
                if write_cache and self.use_cache:
                    key=self._cache_key(ip,port)
                    self._cache[key]={
                        "latency_min":res.latency_min,
                        "latency_mean":res.latency_mean,
                        "latency_median":res.latency_median,
                        "jitter_stddev":res.jitter_stddev,
                        "latency":res.latency,
                        "download_speed":res.download_speed,
                        "upload_speed":res.upload_speed,
                        "download_stddev":getattr(res,"download_stddev",0.0),
                        "upload_stddev":getattr(res,"upload_stddev",0.0),
                        "speed_rounds":getattr(res,"speed_rounds",1),
                        "real_delay_ms":getattr(res,"real_delay_ms",0.0),
                        "loss":res.loss,
                        "tls_time":res.tls_time,
                        "samples":res.samples,
                        "endpoints_ok":res.endpoints_ok,
                        "endpoints_total":res.endpoints_total,
                        "endpoint_breakdown":res.endpoint_breakdown,
                        "status":res.status,
                        "region":res.region,
                        "error_msg":res.error_msg,
                    }

                _bump(res)
                if self.stop_flag:
                    continue
                adjust_inflight()
                submit_more()
        finally:
            try:
                if self.stop_flag:
                    ex.shutdown(wait=False, cancel_futures=True)
                else:
                    ex.shutdown(wait=True, cancel_futures=False)
            except Exception:
                pass
            scanning_active["flag"]=False
            if live_thread:
                try:
                    live_thread.join(timeout=2)
                except Exception:
                    pass
            if live_display:
                try:
                    live_display.stop()
                except Exception:
                    pass

        print()

    def _append_result(self, res:ScanResult, stage:str):
        with self._lock:
            if stage=="Stage 1":
                self.stage1_results.append(res)
            else:
                self.results.append(res)
        # Only print text output if TUI is NOT active (avoid flickering)
        if not self.tui:
            self._print_one(res, stage)

    
    def _print_powered_by(self):
        tag = "@Config_Vortex55"
        if HAS_RICH and self.tui:
            c = Console()
            title = Text("Clean IP Scanner", style="bold cyan")
            body = Text.assemble(("Powered by ", "dim"), (tag, "bold yellow"))
            c.print(Panel(Align.center(body), title=title, border_style="cyan", padding=(1, 4)))
        else:
            line = f" Powered by: {tag} "
            w = max(52, len(line) + 6)
            print("\n" + "╔" + "═"*(w-2) + "╗")
            print("║" + line.center(w-2) + "║")
            print("╚" + "═"*(w-2) + "╝\n")

    def run(self):
        banner()
        self._print_powered_by()
        self._tui_obj = (RichTUI(sort_key=self.sort_key) if (self.tui and HAS_RICH) else None)
        if self._tui_obj:
            self._tui_obj.start()

        start=time.time()
        self._check_deps()

        section("Xray Core")
        if not self.xray.ensure_xray():
            err("Cannot continue without xray")
            raise SystemExit(1)

        section("Parse Config")
        parsed=self.xray.parse_config(self.config_str)
        ok(f"Protocol: {parsed.get('protocol','')}")
        ok(f"Original Host: {parsed.get('host','')}")
        ok(f"SNI: {parsed.get('sni','') or parsed.get('host','')}")

        section(f"Parse Ranges (IPv{self.ip_version})")
        ranges_text = self.ip_ranges if isinstance(self.ip_ranges,str) else "\n".join(self.ip_ranges or [])
        ips=parse_ip_ranges(ranges_text, max_ips=max(1,int(self.count)), ip_version=self.ip_version, scan_order=self.scan_order)
        ok(f"IPs selected: {len(ips)}")
        ports = self.ports if self.ports else [int(parsed.get("original_port", DEFAULT_PORTS[0]))]
        tasks=[(ip,p) for ip in ips for p in ports]
        ok(f"Total candidates: {len(tasks)} (IPs × Ports)")
        ok(f"Latency endpoints: {', '.join(self.latency_urls)}")
        ok(f"Cache: {CACHE_FILE} (hits will be skipped)" if self.use_cache else "Cache: disabled")
        ok(f"Xray process cap: {self.xray_workers} (workers={self.workers})")
        if self.speed_seconds > 0:
            ok(f"Speed mode: duration ({self.speed_seconds:.1f}s per round, chunk={self.speed_kb}KB)")
        else:
            ok(f"Speed mode: legacy size ({self.dl_kb}KB DL / {self.ul_kb}KB UL)")
        if self.real_delay:
            ok(f"Real delay probe: {self.real_delay_host}:{self.real_delay_port} via SOCKS tunnel")

        old=signal.getsignal(signal.SIGINT)
        def on_sigint(sig, frame):
            warn("Stopping now... cancelling pending tasks and saving partial results")
            self.stop_flag=True
            self.stop_event.set()
        signal.signal(signal.SIGINT, on_sigint)

        active_testers=[]
        try:
            tester1=IPTester(
                self.xray,
                parsed,
                timeout=self.timeout,
                max_latency=self.max_latency,
                test_dl=False,
                test_ul=False,
                dl_kb=self.dl_kb,
                ul_kb=self.ul_kb,
                latency_urls=self.latency_urls,
                latency_rounds=self.ping_count,
                speed_rounds=self.speed_rounds,
                speed_seconds=self.speed_seconds,
                speed_kb=self.speed_kb,
                adaptive=True,
                warmup=False,
                cdn_check=self.cdn_check,
                strict_cdn=self.strict_cdn,
                icmp_loss=self.icmp_loss,
                tcp_loss=self.tcp_loss,
                test_h2=self.test_h2,
                test_h3=self.test_h3,
                geoip=self.geoip,
                ipinfo_token=self.ipinfo_token,
                debug=self.debug,
                log_dir=LOG_DIR,
                xray_semaphore=self._xray_semaphore,
                stop_event=self.stop_event,
                real_delay=self.real_delay,
                real_delay_host=self.real_delay_host,
                real_delay_port=self.real_delay_port,
            )
            active_testers.append(tester1)
            self._scan_tasks(tasks, tester1, stage="Stage 1", write_cache=True)

            if self.stop_flag or self.stop_event.is_set():
                warn("Scan interrupted. Skipping Stage 2.")
            else:
                # Stage-1 "clean" list (status values like "CLEAN" / "DIRTY" / "TIMEOUT")
                # Use safe getattr + str() so we never crash on missing/None status.
                stage1_clean=[res for res in self.stage1_results if str(getattr(res, "status", "")).lower()=="clean"]
                stage1_clean.sort(key=lambda x: (x.latency_median or 99999, x.jitter_stddev or 9999, x.loss))
                top=stage1_clean[:self.stage2_top]
                if top:
                    if not self.tui:  # Skip section print if Rich TUI is active
                        section("Warm-up & Re-test")
                        ok(f"🔄 Stage 2 will re-test Top {len(top)} (for accuracy)")
                    else:
                        info(f"🔄 Stage 2 will re-test Top {len(top)}")
                    tasks2=[(r.ip,r.port) for r in top]
                    tester2=IPTester(
                        self.xray,
                        parsed,
                        timeout=self.timeout,
                        max_latency=self.max_latency,
                        test_dl=self.test_dl,
                        test_ul=self.test_ul,
                        dl_kb=self.dl_kb,
                        ul_kb=self.ul_kb,
                        latency_urls=self.latency_urls,
                        latency_rounds=max(self.ping_count*2, self.ping_count+1),
                        speed_rounds=self.speed_rounds,
                        speed_seconds=self.speed_seconds,
                        speed_kb=self.speed_kb,
                        adaptive=True,
                        warmup=True,
                        cdn_check=self.cdn_check,
                        strict_cdn=self.strict_cdn,
                        icmp_loss=self.icmp_loss,
                        tcp_loss=self.tcp_loss,
                        test_h2=self.test_h2,
                        test_h3=self.test_h3,
                        geoip=self.geoip,
                        ipinfo_token=self.ipinfo_token,
                        debug=self.debug,
                        log_dir=LOG_DIR,
                        xray_semaphore=self._xray_semaphore,
                        stop_event=self.stop_event,
                        real_delay=self.real_delay,
                        real_delay_host=self.real_delay_host,
                        real_delay_port=self.real_delay_port,
                    )
                    active_testers.append(tester2)
                    self._scan_tasks(tasks2, tester2, stage="Stage 2", write_cache=True)
                else:
                    warn("No clean candidates from Stage 1.")

        except KeyboardInterrupt:
            # Windows can still raise KeyboardInterrupt even with SIGINT handler.
            self.stop_flag = True
            self.stop_event.set()
            warn("Interrupted by user (Ctrl+C). Saving partial results...")

        finally:
            for t in active_testers:
                try:
                    t.stop_all_processes()
                except Exception:
                    pass
            try:
                signal.signal(signal.SIGINT, old)
            except Exception:
                pass
            self._save_cache()

        # Optional stability mode (post-check on top results)
        if (not self.stop_flag) and self.stability and self.stability>0:
            section("Stability Mode")
            host_for_probe = parsed.get("sni") or parsed.get("host") or ""
            tester_stab = IPTester(
                self.xray,
                parsed,
                timeout=self.timeout,
                max_latency=self.max_latency,
                test_dl=False,
                test_ul=False,
                dl_kb=self.dl_kb,
                ul_kb=self.ul_kb,
                latency_urls=self.latency_urls,
                latency_rounds=1,
                speed_rounds=1,
                speed_seconds=min(2.0, max(1.0, self.speed_seconds)),
                speed_kb=self.speed_kb,
                adaptive=False,
                warmup=False,
                cdn_check=False,
                strict_cdn=False,
                icmp_loss=False,
                tcp_loss=False,
                test_h2=False,
                test_h3=False,
                geoip=False,
                ipinfo_token=None,
                debug=self.debug,
                log_dir=LOG_DIR,
                xray_semaphore=self._xray_semaphore,
                stop_event=self.stop_event,
                real_delay=False,
            )
            top_list = [res for res in (self.results or self.stage1_results) if str(getattr(res, "status", "")).lower()=="clean"][:self.stability_top]
            info(f"🔍 Testing stability of {len(top_list)} IPs ({self.stability}s each)")
            
            emoji_frames = ["🟩", "🟧", "🟨", "🟦", "🟪", "🟫"]
            for idx, r in enumerate(top_list):
                if self.stop_flag or self.stop_event.is_set():
                    break
                
                # Show "starting" state
                emoji = emoji_frames[idx % len(emoji_frames)]
                sys.stdout.write(f"\r  {emoji} Testing {r.ip}:{r.port}... ")
                sys.stdout.flush()
                
                st = tester_stab.stability_check(r.ip, r.port, host_for_probe, seconds=self.stability, interval=self.stability_interval)
                r.stability_uptime = st.get("uptime",0.0)
                r.stability_variance_ms = st.get("variance_ms",0.0)
                r.stability_spikes = st.get("spikes",0)
                r.stability_traffic_mbps = st.get("traffic_mbps",0.0)
                
                # Show "completed" state with progress bar
                progress = int(((idx + 1) / len(top_list)) * 100) if len(top_list) > 0 else 0
                bar_len = 20
                filled = int(bar_len * (idx + 1) / len(top_list)) if len(top_list) > 0 else 0
                bar = "█" * filled + "▒" * (bar_len - filled)
                
                sys.stdout.write(f"\r  ✓ [{bar}] {progress:3d}% ({idx+1}/{len(top_list)}) {r.ip}:{r.port}          ")
                sys.stdout.flush()
            
            # Final newline after progress bar
            print()
            ok(f"✅ Stability testing completed for {len(top_list)} IPs")

        self._report(time.time()-start, parsed)
        finished()
        if self._tui_obj:
            try: self._tui_obj.stop()
            except Exception: pass
    def _make_output_dir(self, ts:str)->str:
        out_dir=os.path.join(os.getcwd(), f"output_{ts}")
        os.makedirs(out_dir, exist_ok=True)
        return out_dir

    def _to_dict(self, r:ScanResult)->dict:
        return {
            "ip":r.ip,
            "port":r.port,
            "status":r.status,
            "latency_min_ms":r.latency_min,
            "latency_mean_ms":r.latency_mean,
            "latency_median_ms":r.latency_median,
            "jitter_stddev_ms":r.jitter_stddev,
            "loss_percent":r.loss,
            "tls_time_ms":r.tls_time,
            "real_delay_ms":getattr(r,"real_delay_ms",0.0),
            "download_mbps":r.download_speed,
            "upload_mbps":r.upload_speed,
            "samples":r.samples,
            "endpoints_ok":r.endpoints_ok,
            "endpoints_total":r.endpoints_total,
            "region":r.region,
            "error_msg":r.error_msg,
            "endpoint_breakdown":r.endpoint_breakdown or {},
            "icmp_loss_percent":getattr(r,"icmp_loss",0.0),
            "tcp_loss_percent":getattr(r,"tcp_loss",0.0),
            "cdn_ok":bool(getattr(r,"cdn_ok",True)),
            "cdn_vendor":getattr(r,"cdn_vendor",""),
            "http2":bool(getattr(r,"h2",False)),
            "http3":bool(getattr(r,"h3",False)),
            "stability_uptime_percent":getattr(r,"stability_uptime",0.0),
            "stability_variance_ms":getattr(r,"stability_variance_ms",0.0),
            "stability_spikes":getattr(r,"stability_spikes",0),
            "stability_traffic_mbps":getattr(r,"stability_traffic_mbps",0.0),
            "geo":getattr(r,"geo",{}) or {},
        }

    def _export(self, out_dir:str, clean:List[ScanResult], all_results:List[ScanResult], parsed:dict):
        # TXT
        out_txt=os.path.join(out_dir, "clean_ips.txt")
        with open(out_txt,"w",encoding="utf-8") as f:
            for r in clean:
                f.write(r.ip+"\n")

        # CSV
        out_csv=os.path.join(out_dir, "results.csv")
        with open(out_csv,"w",newline="",encoding="utf-8") as f:
            w=csv.DictWriter(f, fieldnames=list(self._to_dict(all_results[0]).keys()) if all_results else [])
            if all_results:
                w.writeheader()
                for r in all_results:
                    w.writerow(self._to_dict(r))

        # JSON
        out_json=os.path.join(out_dir, "results.json")
        with open(out_json,"w",encoding="utf-8") as f:
            json.dump({
                "generated_at":datetime.now().isoformat(),
                "latency_endpoints":self.latency_urls,
                "stage2_top":self.stage2_top,
                "ip_version":self.ip_version,
                "speed_rounds":self.speed_rounds,
                "speed_seconds":self.speed_seconds,
                "speed_kb":self.speed_kb,
                "xray_workers":self.xray_workers,
                "real_delay":self.real_delay,
                "real_delay_target":f"{self.real_delay_host}:{self.real_delay_port}",
                "results":[self._to_dict(r) for r in all_results],
            }, f, indent=2, ensure_ascii=False)

        # All clean configs (all clean IPs)
        out_cfg=os.path.join(out_dir, "all_clean_configs.txt")
        with open(out_cfg,"w",encoding="utf-8") as f:
            for r in clean:
                f.write(generate_config_with_ip(self.config_str, r.ip)+"\n")
        
        # Auto-generate BEST config (top 1)
        try:
            if clean:
                best = clean[0]
                out_best_ip = os.path.join(out_dir, "best_ip.txt")
                out_best_cfg = os.path.join(out_dir, "best_ip_config.txt")
                with open(out_best_ip, "w", encoding="utf-8") as f:
                    f.write(f"{best.ip}:{best.port}\n")
                with open(out_best_cfg, "w", encoding="utf-8") as f:
                    f.write(generate_config_with_ip(self.config_str, best.ip) + "\n")
        except Exception:
            pass

        # HTML report
        out_html=os.path.join(out_dir, "report.html")
        build_html_report(out_html, clean, all_results, self.latency_urls)

        ok(f"💾 Saved: {out_txt}")
        ok(f"💾 Saved: {out_csv}")
        ok(f"💾 Saved: {out_json}")
        ok(f"💾 Saved: {out_html}")
        ok(f"💾 Saved: {out_cfg}")

    def _report(self, elapsed:float, parsed:dict):
        section("Results")
        # Merge Stage2 results with Stage1 (for IPs not re-tested in Stage2)
        # Stage2 overrides Stage1 results for the same IP, so we keep Stage2 values if present
        final = list(self.results) if self.results else []
        if self.results:
            # Collect IPs that were re-tested in Stage2
            stage2_ips = {res.ip for res in self.results}
            # Add Stage1 results that weren't in Stage2
            for res in self.stage1_results:
                if res.ip not in stage2_ips:
                    final.append(res)
        else:
            # If no Stage2, just use Stage1
            final = list(self.stage1_results)
        
        clean=[res for res in final if str(getattr(res, "status", "")).lower()=="clean"]
        clean.sort(key=lambda x: x.score)

        # Stage1 contains the full population (including timeouts/errors).
        s1=self.stage1_results or []
        s1_dirty=len([r for r in s1 if r.status=='dirty'])
        s1_timeout=len([r for r in s1 if r.status=='timeout'])
        s1_error=len([r for r in s1 if r.status=='error'])

        print()
        print(f"  📊 Stage1 total: {len(self.stage1_results)}")
        print(f"  📊 Stage2 total: {len(self.results)}")
        print(f"  ✅ Clean:        {Colors.GREEN}{len(clean)}{Colors.RESET}")
        print(f"  ❌ Dirty:        {Colors.RED}{s1_dirty}{Colors.RESET}")
        print(f"  ⏱ Timeout:      {Colors.YELLOW}{s1_timeout}{Colors.RESET}")
        print(f"  ⚠ Error:        {s1_error}")
        print(f"  ⏱ Time:         {elapsed:.1f}s")

        ts=datetime.now().strftime("%Y%m%d_%H%M%S")
        out_dir=self._make_output_dir(ts)
        self._export(out_dir, clean, final, parsed)

        if clean:
            section("Top 10 (quick copy)")
            # Emoji score bar based on score percentile among clean results
            try:
                _scores_all=[float(x.score or 0.0) for x in clean if getattr(x,"score",None) is not None]
                _smin=min(_scores_all) if _scores_all else 0.0
                _smax=max(_scores_all) if _scores_all else 0.0
            except Exception:
                _smin=_smax=0.0

            def _stars(score):
                try:
                    if _smax <= _smin:
                        return "[⭐⭐⭐]"
                    t=(float(score)-float(_smin))/(float(_smax)-float(_smin))
                    t=max(0.0,min(1.0,t))
                    # Lower score = better.
                    n=5-int(round(t*4))
                    n=max(1,min(5,int(n)))
                    return "⭐" * n
                except Exception:
                    return "—"
            
            # Display top results with better formatting
            print(f"\n  {Colors.BOLD}{'Rank':<6} {'IP':<20} {'Delay':<10} {'Jitter':<10} {'Loss %':<8} {'DL':<10} {'Score':<15}{Colors.RESET}")
            print(f"  {Colors.BLUE}{'-'*90}{Colors.RESET}")
            
            for idx, r in enumerate(clean[:10], 1):
                # Select medal emoji based on rank
                if idx == 1:
                    rank_emoji = "🥇"
                elif idx == 2:
                    rank_emoji = "🥈"
                elif idx == 3:
                    rank_emoji = "🥉"
                else:
                    rank_emoji = "  "
                
                score_stars = _stars(r.score)
                dl_str = f"{r.download_speed:.2f}M" if r.download_speed > 0 else "N/A"
                
                print(
                    f"  {rank_emoji} {idx:<3} {Colors.CYAN}{r.ip:<20}{Colors.RESET} "
                    f"{getattr(r,'real_delay_ms',0.0):>8.1f}ms {r.jitter_stddev:>8.1f}ms {r.loss:>6.1f}% {dl_str:>9} "
                    f"{score_stars} {float(r.score or 0.0):.0f}"
                )

def _http_get_text(url:str, timeout:int=30)->str:
    req=urllib.request.Request(url, headers={"User-Agent":"Mozilla/5.0"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return r.read().decode("utf-8",errors="ignore")


def update_ranges_now():
    """Download official Cloudflare & Fastly ranges and update local files under ranges/."""
    os.makedirs(RANGES_DIR, exist_ok=True)
    section("Update ranges")
    # Cloudflare
    try:
        info("Downloading Cloudflare IPv4...")
        cf4=_http_get_text("https://www.cloudflare.com/ips-v4").strip()+"\n"
        with open(os.path.join(RANGES_DIR,"cloudflare_ipv4.txt"),"w",encoding="utf-8") as f:
            f.write(cf4)
        ok("Updated ranges/cloudflare_ipv4.txt")
        info("Downloading Cloudflare IPv6...")
        cf6=_http_get_text("https://www.cloudflare.com/ips-v6").strip()+"\n"
        with open(os.path.join(RANGES_DIR,"cloudflare_ipv6.txt"),"w",encoding="utf-8") as f:
            f.write(cf6)
        ok("Updated ranges/cloudflare_ipv6.txt")
    except Exception as e:
        warn(f"Cloudflare IPv4 update failed: {e}")
    # Fastly
    try:
        info("Downloading Fastly public-ip-list...")
        txt=_http_get_text("https://api.fastly.com/public-ip-list")
        obj=json.loads(txt)
        v4=obj.get("addresses",[])
        v6=obj.get("ipv6_addresses", obj.get("ipv6Addresses", obj.get("ipv6", [])))
        with open(os.path.join(RANGES_DIR,"fastly_ipv4.txt"),"w",encoding="utf-8") as f:
            f.write("\n".join(v4)+"\n")
        ok("Updated ranges/fastly_ipv4.txt")
        if isinstance(v6,list) and v6:
            with open(os.path.join(RANGES_DIR,"fastly_ipv6.txt"),"w",encoding="utf-8") as f:
                f.write("\n".join(v6)+"\n")
            ok("Updated ranges/fastly_ipv6.txt")
    except Exception as e:
        warn(f"Fastly IPv4 update failed: {e}")



def generate_config_with_ip(config_str:str, ip:str)->str:
    """Replace the *server address* with a given IP, while keeping SNI/Host params intact.
    Works for vless://, vmess://, trojan:// (best-effort).
    """
    cfg=(config_str or "").strip()
    if not cfg:
        return cfg
    try:
        if cfg.startswith("vless://"):
            head=cfg[:8]
            body=cfg[8:]
            frag=""
            if "#" in body:
                body,frag=body.rsplit("#",1)
                frag="#"+frag
            user,rest=body.split("@",1)
            if "?" in rest:
                hostport,qs=rest.split("?",1)
                q="?"+qs
            else:
                hostport,q=rest,""
            if hostport.startswith("["):
                _,port=hostport.rsplit(":",1)
            else:
                _,port=hostport.rsplit(":",1)
            return f"{head}{user}@{ip}:{port}{q}{frag}"

        if cfg.startswith("trojan://"):
            head=cfg[:9]
            body=cfg[9:]
            frag=""
            if "#" in body:
                body,frag=body.rsplit("#",1)
                frag="#"+frag
            pwd,rest=body.split("@",1)
            if "?" in rest:
                hostport,qs=rest.split("?",1)
                q="?"+qs
            else:
                hostport,q=rest,""
            if hostport.startswith("["):
                _,port=hostport.rsplit(":",1)
            else:
                _,port=hostport.rsplit(":",1)
            return f"{head}{pwd}@{ip}:{port}{q}{frag}"

        if cfg.startswith("vmess://"):
            import base64
            enc=cfg[8:]
            pad=4-len(enc)%4
            if pad!=4: enc+="="*pad
            obj=json.loads(base64.b64decode(enc).decode("utf-8"))
            obj["add"]=ip
            raw=json.dumps(obj, ensure_ascii=False)
            b64=base64.b64encode(raw.encode("utf-8")).decode("utf-8").rstrip("=")
            return "vmess://"+b64
    except Exception:
        return cfg
    return cfg


def build_html_report(out_path:str, clean:List[ScanResult], all_results:List[ScanResult], endpoints:List[str]):
    """Generate a self-contained HTML report (sortable table + Top10)."""
    rows=[]
    for r in all_results:
        rows.append({
            "ip":r.ip,
            "port":r.port,
            "status":r.status,
            "median":r.latency_median,
            "mean":r.latency_mean,
            "jitter":r.jitter_stddev,
            "loss":r.loss,
            "tls":r.tls_time,
            "real_delay":getattr(r,"real_delay_ms",0.0),
            "dl":r.download_speed,
            "ul":r.upload_speed,
            "dl_std":getattr(r,"download_stddev",0.0),
            "ul_std":getattr(r,"upload_stddev",0.0),
            "stab_u":getattr(r,"stability_uptime",0.0),
            "stab_v":getattr(r,"stability_variance_ms",0.0),
            "spikes":getattr(r,"stability_spikes",0),
            "stab_tr":getattr(r,"stability_traffic_mbps",0.0),
            "region":r.region or "",
            "score":round(r.score,2),
        })

    top10=clean[:10]
    top_rows="".join(
        f"<tr><td>{i+1}</td><td>{r.ip}</td><td>{round(r.score,2)}</td><td>{r.latency_median}</td><td>{r.jitter_stddev}</td><td>{r.loss}</td><td>{r.download_speed}</td><td>{r.upload_speed}</td></tr>"
        for i,r in enumerate(top10)
    )
    rows_json=json.dumps(rows, ensure_ascii=False)
    generated=datetime.now().isoformat()
    endpoints_txt=", ".join(endpoints or [])

    # IMPORTANT: this template is NOT an f-string (to avoid escaping JS/CSS braces).
    html="""<!doctype html>
<html lang="en"><head>
<meta charset="utf-8"/>
<meta name="viewport" content="width=device-width, initial-scale=1"/>
<title>Clean IP Scanner Report</title>
<style>
  :root{
    --bg1:#f5f7ff;
    --bg2:#eef8ff;
    --card:#ffffffdd;
    --text:#102136;
    --line:#dbe4ef;
    --head:#edf4ff;
  }
  body{
    font-family:"Segoe UI", "Trebuchet MS", system-ui, -apple-system, sans-serif;
    margin:24px;
    color:var(--text);
    background:
      radial-gradient(circle at 10% 0%, #e4f0ff 0%, transparent 40%),
      radial-gradient(circle at 100% 20%, #dff7ff 0%, transparent 35%),
      linear-gradient(150deg, var(--bg1), var(--bg2));
    min-height:100vh;
  }
  .muted{color:#4b5d74;}
  .card{
    border:1px solid var(--line);
    border-radius:16px;
    padding:16px;
    margin:12px 0;
    background:var(--card);
    backdrop-filter: blur(4px);
    box-shadow:0 10px 30px rgba(22,58,108,.08);
  }
  table{border-collapse:collapse; width:100%; font-size:13px;}
  th,td{border-bottom:1px solid #e6edf5; padding:8px 10px; text-align:left;}
  th{cursor:pointer; background:var(--head); position:sticky; top:0;}
  .tag{display:inline-block; padding:2px 8px; border-radius:999px; font-size:12px;}
  .ok{background:#dcfce7; color:#166534;}
  .bad{background:#fee2e2; color:#991b1b;}
  .to{background:#fef9c3; color:#854d0e;}
  .err{background:#e5e7eb; color:#374151;}
</style>
</head>
<body>
<h1>Clean IP Scanner Report</h1>
<p class="muted">Generated: __GENERATED__<br/>Endpoints: __ENDPOINTS__</p>

<div class="card">
  <h2>Top 10 (best score)</h2>
  <table>
    <thead><tr><th>#</th><th>IP</th><th>Score</th><th>Median (ms)</th><th>Jitter</th><th>Loss %</th><th>DL</th><th>UL</th></tr></thead>
    <tbody>__TOP10__</tbody>
  </table>
</div>

<div class="card">
  <h2>All results</h2>
  <p class="muted">Click headers to sort.</p>
  <table id="t">
    <thead>
      <tr>
        <th data-k="status">Status</th>
        <th data-k="ip">IP</th>
        <th data-k="port">Port</th>
        <th data-k="median">Median</th>
        <th data-k="mean">Mean</th>
        <th data-k="jitter">Jitter</th>
        <th data-k="loss">Loss%</th>
        <th data-k="tls">TLS</th>
        <th data-k="real_delay">RealDelay</th>
        <th data-k="dl">DL</th>
        <th data-k="ul">UL</th>
        <th data-k="dl_std">DL σ</th>
        <th data-k="ul_std">UL σ</th>
        <th data-k="stab_u">Stab%</th>
        <th data-k="stab_v">Var(ms)</th>
        <th data-k="spikes">Spikes</th>
        <th data-k="stab_tr">Stab DL</th>
        <th data-k="region">Region</th>
        <th data-k="score">Score</th>
      </tr>
    </thead>
    <tbody id="tb"></tbody>
  </table>
</div>

<script>
const rows = __ROWS__;
const tb = document.getElementById('tb');
// Heatmap helpers (lower-is-better for score/median/jitter/tls/loss/stddev; higher-is-better for DL/UL/uptime)
function mm(vals){
  const v = vals.filter(x => typeof x === 'number' && isFinite(x));
  if(!v.length) return [0,1];
  return [Math.min(...v), Math.max(...v)];
}
function clamp01(x){ return Math.max(0, Math.min(1, x)); }
function heat(val, min, max, invert){
  if(!(typeof val === 'number') || !isFinite(val)) return '';
  if(max <= min) return '';
  let t = (val - min) / (max - min);
  t = clamp01(t);
  if(invert) t = 1 - t;
  // 120 (green) -> 0 (red)
  const hue = 120 * t;
  return `background-color: hsl(${hue}, 80%, 92%);`;
}

const cleanRows = rows.filter(r => r.status === 'clean');
const [scoreMin, scoreMax] = mm(cleanRows.map(r => r.score));
const [medMin, medMax]     = mm(cleanRows.map(r => r.median));
const [dlMin, dlMax]       = mm(cleanRows.map(r => r.dl));
const [ulMin, ulMax]       = mm(cleanRows.map(r => r.ul));
const [jitMin, jitMax]     = mm(cleanRows.map(r => r.jitter));
const [lossMin, lossMax]   = mm(cleanRows.map(r => r.loss));
function tag(status){
  if(status==='clean') return '<span class="tag ok">CLEAN</span>';
  if(status==='dirty') return '<span class="tag bad">DIRTY</span>';
  if(status==='timeout') return '<span class="tag to">TIMEOUT</span>';
  return '<span class="tag err">ERROR</span>';
}
function render(data){
  tb.innerHTML = data.map(r => `
    <tr>
      <td>${tag(r.status)}</td>
      <td>${r.ip}</td>
      <td>${r.port}</td>
      <td style="${heat(r.median, medMin, medMax, true)}">${r.median}</td>
      <td>${r.mean}</td>
      <td style="${heat(r.jitter, jitMin, jitMax, true)}">${r.jitter}</td>
      <td style="${heat(r.loss, lossMin, lossMax, true)}">${r.loss}</td>
      <td>${r.tls}</td>
      <td>${r.real_delay}</td>
      <td style="${heat(r.dl, dlMin, dlMax, false)}">${r.dl}</td>
      <td style="${heat(r.ul, ulMin, ulMax, false)}">${r.ul}</td>
      <td>${r.dl_std}</td>
      <td>${r.ul_std}</td>
      <td>${r.stab_u}</td>
      <td>${r.stab_v}</td>
      <td>${r.spikes}</td>
      <td>${r.stab_tr}</td>
      <td>${r.region||''}</td>
      <td style="${heat(r.score, scoreMin, scoreMax, true)} font-weight:600;">${r.score}</td>
    </tr>`).join('');
}
let curK='score', asc=true;
function sortBy(k){
  asc = (curK===k) ? !asc : true;
  curK=k;
  const data=[...rows].sort((a,b)=>{
    const av=a[k], bv=b[k];
    const an = (typeof av==='number') ? av : (av||'').toString();
    const bn = (typeof bv==='number') ? bv : (bv||'').toString();
    if(typeof an==='number' && typeof bn==='number') return asc ? an-bn : bn-an;
    return asc ? an.localeCompare(bn) : bn.localeCompare(an);
  });
  rows_sorted = data;
  render(data);
  applyHeatmap();
}

function colorScale(v,min,max,reverse){
  if(max<=min) return '';
  let t=(v-min)/(max-min);
  if(reverse) t=1-t;
  t=Math.max(0,Math.min(1,t));
  // green->yellow->red
  const hue = 120 - (120*t);
  return `hsl(${hue}, 85%, 88%)`;
}

function applyHeatmap(){
  const cols=['median','jitter','loss','tls','real_delay','score','dl','ul','dl_std','ul_std','stab_v','spikes','stab_tr'];
  const stats={};
  cols.forEach(k=>{
    const vs=rows.map(r=>Number(r[k]||0)).filter(v=>!isNaN(v));
    stats[k]={min:Math.min(...vs), max:Math.max(...vs)};
  });
  // For dl/ul and stab_u we want higher better -> reverse scale
  const reverseCols=new Set(['dl','ul','stab_u','stab_tr']);
  document.querySelectorAll('#tb tr').forEach((tr,idx)=>{
    const r = rows_sorted[idx] || null;
    if(!r) return;
  });
  // apply by reading current DOM cells
  const trs=[...document.querySelectorAll('#tb tr')];
  trs.forEach(tr=>{
    const tds=[...tr.children];
    // column indices in the rendered table
    const map={median:3, mean:4, jitter:5, loss:6, tls:7, real_delay:8, dl:9, ul:10, dl_std:11, ul_std:12, stab_u:13, stab_v:14, spikes:15, stab_tr:16, score:18};
    Object.entries(map).forEach(([k,i])=>{
      const td=tds[i];
      if(!td) return;
      const v=Number(td.textContent||0);
      if(isNaN(v)) return;
      const s=stats[k] || {min:0,max:0};
      td.style.background=colorScale(v,s.min,s.max, reverseCols.has(k));
      td.style.borderRadius='6px';
    });
  });
}

let rows_sorted=[...rows].sort((a,b)=>a.score-b.score);
render(rows_sorted);
applyHeatmap();
document.querySelectorAll('th[data-k]').forEach(th=> th.addEventListener('click', ()=>sortBy(th.dataset.k)));
</script>
</body></html>"""

    html=(html
          .replace("__GENERATED__", generated)
          .replace("__ENDPOINTS__", endpoints_txt)
          .replace("__TOP10__", top_rows)
          .replace("__ROWS__", rows_json))
    with open(out_path,"w",encoding="utf-8") as f:
        f.write(html)


def read_text_file(path):
    with open(path,"r",encoding="utf-8") as f: return f.read().strip()

def prompt_int(label, default, mn=1, mx=10000):
    while True:
        s=input(f"{label} [{default}]: ").strip()
        if not s: return default
        if s.isdigit():
            v=int(s)
            if mn<=v<=mx: return v
        print("  Invalid number. Try again.")

def prompt_float(label, default, mn=0.1, mx=120.0):
    while True:
        s=input(f"{label} [{default}]: ").strip()
        if not s: return default
        try:
            v=float(s)
            if mn<=v<=mx: return v
        except ValueError:
            pass
        print("  Invalid number. Try again.")


def prompt(label, default=""):
    """Simple string prompt with default (for backward compatibility)."""
    d = str(default) if default is not None else ""
    try:
        s = input(f"  {label} [{d}]: ").strip()
    except (EOFError, KeyboardInterrupt):
        print("\n  ⚠ Cancelled by user. Exiting...\n")
        raise SystemExit(0)
    return s if s else d

def prompt_yesno(label, default):
    d="Y" if default else "N"
    while True:
        s=input(f"{label} (Y/N) [{d}]: ").strip().lower()
        if not s: return default
        if s in ("y","yes"): return True
        if s in ("n","no"): return False
        print("  Please enter Y or N.")

def wizard():
    banner()
    
    # Offer to install Rich if not already installed
    if not HAS_RICH:
        section("Live UI Enhancement")
        print("  The 'Rich' library provides a professional live dashboard UI (recommended)")
        print("  Install now?")
        if prompt_yesno("  Install Rich?", True):
            print("  Installing Rich...")
            try:
                import subprocess
                subprocess.run([sys.executable, "-m", "pip", "install", "rich", "-q"], check=True, timeout=30)
                print("  ✓ Rich installed successfully! Restart the scanner to enable dashboard.")
            except Exception as e:
                print(f"  ✗ Failed to install: {e}")
                print("  You can install manually with: pip install rich")
        print()
    
    section("Wizard Mode")
    if prompt_yesno("  Update CDN ranges now?", False):
        update_ranges_now()
    print("  Config input:")
    print("   1) Read from config.sample.txt (recommended)")
    print("   2) Paste config link manually (vless/vmess/trojan)")
    ch=input("  Choice [1]: ").strip() or "1"
    if ch=="2":
        cfg=input("  Paste your vless/vmess/trojan link: ").strip()
    else:
        cfg=read_text_file("config.sample.txt")
    
    # Ports (override)
    # Many CDN-fronted configs use port 80 (HTTP) or 443 (TLS). We let the user choose explicitly.
    inferred_port = None
    cfg_one = (cfg or "").strip()

    # Try to infer port from common link formats (vless/vmess/trojan/etc.)
    m = re.search(r'@[^/?#:]+:(\d+)', cfg_one)
    if m:
        inferred_port = m.group(1)
    else:
        # Try to infer from JSON configs: "port": 443
        m2 = re.search(r'"port"\s*:\s*(\d+)', cfg_one)
        if m2:
            inferred_port = m2.group(1)

    default_ports = "443"
    ports_in = prompt("Ports to scan (comma-separated)", default_ports)
    ports = []
    try:
        ports = [int(x) for x in ports_in.replace(" ", "").split(",") if x]
    except Exception:
        ports = [int(default_ports)]

    # Optional: override ranges with specific IP/CIDR targets
    # Custom targets disabled in this build (ranges-only scan)
    custom_targets = None

    # Get script directory for relative paths
    # Find ranges folder - could be in multiple locations
    script_dir = os.path.dirname(os.path.abspath(__file__))
    ranges_path = None
    
    # Check multiple possible locations for ranges folder
    possible_paths = [
        os.path.join(script_dir, "ranges"),  # Same dir as script
        os.path.join(os.path.dirname(script_dir), "ranges"),  # Parent dir
        os.path.join(os.path.dirname(os.path.dirname(script_dir)), "ranges"),  # 2 levels up
        os.path.join(os.path.dirname(os.path.dirname(script_dir)), "CleanIPScanner", "ranges"),  # CleanIPScanner folder
        os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(script_dir))), "CleanIPScanner", "ranges"),  # 3 levels up then CleanIPScanner
    ]
    
    for path_candidate in possible_paths:
        if os.path.isdir(path_candidate):
            ranges_path = path_candidate
            break
    
    # If still not found, use first candidate (will error gracefully with helpful message)
    if not ranges_path:
        ranges_path = possible_paths[0]

    section("CDN / Ranges")
    print("  Select IP ranges:")
    print("   1) Cloudflare (ranges\\cloudflare_ipv4.txt)")
    print("   2) Fastly (ranges\\fastly_ipv4.txt)")
    print("   3) Custom file path")
    print("   4) Enter a specific IP / CIDR / range manually")
    if custom_targets:
        # custom_targets can be a list of IPs/CIDRs or a file path (string)
        ranges_text = custom_targets
        print("  Using custom targets provided by user.")
    else:
        rch=input("  Choice [1]: ").strip() or "1"
        if rch=="2":
            ranges_text=read_text_file(os.path.join(ranges_path, "fastly_ipv4.txt"))
        elif rch=="3":
            rpath=input("  Ranges file path: ").strip()
            ranges_text=read_text_file(rpath)
        elif rch=="4":
            manual=input("  Paste IPs/CIDRs/ranges (one per line, blank to finish). Paste now:\n")
            # If user pasted multiple lines, accept as-is; otherwise allow single-line
            if manual.strip():
                ranges_text = manual
            else:
                ranges_text = ""
        else:
            ranges_text=read_text_file(os.path.join(ranges_path, "cloudflare_ipv4.txt"))

    
    # Scan order (default: random)
    print("  Scan order for IPs from ranges file:")
    print("   1) Random (default)  — shuffle the IPs")
    print("   2) Ordered           — keep file order")
    so = input("  Choice [1]: ").strip() or "1"
    scan_order = "ordered" if so == "2" else "random"

    section("Latency endpoints")
    print("  Default endpoints:")
    for i,u in enumerate(DEFAULT_LATENCY_URLS,1):
        print(f"   {i}) {u}")
    use_default=prompt_yesno("  Use default endpoints?", True)
    if use_default:
        latency_urls=list(DEFAULT_LATENCY_URLS)
    else:
        s=input("  Enter endpoints (comma-separated URLs): ").strip()
        latency_urls=[x.strip() for x in s.split(",") if x.strip()] or list(DEFAULT_LATENCY_URLS)
    section("Performance / Limits")
    count=prompt_int("  How many IPs to sample from ranges?", SCAN_COUNT, 1, 5000)
    workers=prompt_int("  Workers (concurrency)", CONCURRENCY, 1, 256)
    xray_workers=prompt_int("  Max concurrent Xray processes", min(DEFAULT_XRAY_WORKERS, workers), 1, workers)
    timeout=prompt_float("  Timeout seconds", TIMEOUT, 1.0, 60.0)
    max_lat=prompt_int("  Max acceptable latency (ms)", MAX_LATENCY, 50, 60000)
    stage2_top=prompt_int("  Stage2 re-test top N (more accurate)", 30, 1, 500)
    
    section("Stability & Consistency Testing")
    print("  Real stability testing: 60 seconds with 5s intervals")
    print("  Measures: uptime %, variance, and latency spikes")
    stability=prompt_int("  Stability test seconds (60=recommended)", 60, 0, 600)
    stability_interval=prompt_float("  Stability probe interval seconds", 5.0, 1.0, 30.0) if stability>0 else 5.0
    stability_top=prompt_int("  Stability check top N clean IPs", 15, 1, 100) if stability>0 else 15
    use_cache=prompt_yesno("  Use cache (skip already-tested)?", True)
    debug=prompt_yesno("  Debug mode (save logs)?", False)
    # Live UI (Rich)
    # If Rich is already installed, enable the live table automatically.
    # If not installed, offer to install it (default: No). Once installed, it will be used automatically next runs.
    tui = True if HAS_RICH else False
    if not HAS_RICH:
        print(f"  💡 For a colorful live table UI you can install Rich.")
        if prompt_yesno(f"  Install Rich now? (uses: {sys.executable} -m pip install rich)", False):
            try:
                import subprocess
                subprocess.check_call([sys.executable, "-m", "pip", "install", "rich"])
                print("  ✅ Rich installed. Live table UI is now enabled.")
                tui = True
            except Exception as e:
                print(f"  ❌ Failed to install Rich: {e}")
                tui = False

    sort_key = "score"
    section("Speed Tests")
    test_dl=prompt_yesno("  Download test?", True)
    test_ul=prompt_yesno("  Upload test?", True)
    speed_seconds=prompt_float("  Speed test duration per round (seconds, 0=legacy size mode)", SPEED_TEST_SECONDS, 0.0, 30.0)
    speed_kb=prompt_int("  Speed test chunk size (KB)", SPEED_TEST_CHUNK_KB, 32, 16384)
    dl_kb=prompt_int("  Download size (KB)", DOWNLOAD_TEST_SIZE, 10, 5000)
    ul_kb=prompt_int("  Upload size (KB)", UPLOAD_TEST_SIZE, 10, 5000)
    speed_rounds=prompt_int("  Speed test rounds (mean/std)", 3, 1, 10)
    section("Accuracy Options")
    real_delay=prompt_yesno("  Real tunnel delay probe (SOCKS TCP connect)?", True)
    return cfg, ports, ranges_text, scan_order, workers, xray_workers, timeout, max_lat, count, test_dl, test_ul, dl_kb, ul_kb, speed_rounds, speed_seconds, speed_kb, latency_urls, stage2_top, stability, stability_interval, stability_top, use_cache, debug, tui, sort_key, real_delay

def main():
    parser=argparse.ArgumentParser(description="Clean IP Scanner — Xray Core (Wizard + CLI)")
    parser.add_argument("--config","-c",type=str,default=None)
    parser.add_argument("--config-file","-cf",type=str,default=None)
    parser.add_argument("--range","-r",type=str,default=None)
    parser.add_argument("--range-file","-rf",type=str,default=None)
    parser.add_argument("--ip", action="append", default=None, help="Target IP/CIDR (repeatable, or comma-separated) - overrides ranges")
    parser.add_argument("--ip-file", type=str, default=None, help="File with target IPs/CIDRs - overrides ranges")
    parser.add_argument("--ports","-p",type=str,default="443")
    parser.add_argument("--workers","-w",type=int,default=CONCURRENCY)
    parser.add_argument("--timeout","-t",type=float,default=TIMEOUT)
    parser.add_argument("--max-latency","-ml",type=int,default=MAX_LATENCY)
    parser.add_argument("--count","-n",type=int,default=SCAN_COUNT)
    parser.add_argument("--no-download",action="store_true")
    parser.add_argument("--no-upload",action="store_true")
    parser.add_argument("--download-size",type=int,default=DOWNLOAD_TEST_SIZE)
    parser.add_argument("--upload-size",type=int,default=UPLOAD_TEST_SIZE)
    parser.add_argument("--speed-rounds",type=int,default=3, help="Speed test rounds (mean/std); 1-10")
    parser.add_argument("--speed-seconds",type=float,default=SPEED_TEST_SECONDS, help="Duration-based speed test seconds per round (0 = legacy size mode)")
    parser.add_argument("--speed-kb",type=int,default=SPEED_TEST_CHUNK_KB, help="Chunk size for duration-based speed tests (KB)")
    parser.add_argument("--xray-workers",type=int,default=DEFAULT_XRAY_WORKERS, help="Max concurrent xray.exe processes")
    parser.add_argument("--endpoints",type=str,default=None, help="Comma-separated latency URLs")
    parser.add_argument("--stage2-top",type=int,default=30, help="Re-test Top N from Stage1")
    parser.add_argument("--no-cache",action="store_true")
    parser.add_argument("--refresh-cache",action="store_true", help="Ignore old cache and rewrite")
    parser.add_argument("--debug",action="store_true", help="Save xray stderr/logs to logs/")
    parser.add_argument("--update-ranges",action="store_true", help="Update Cloudflare/Fastly ranges then exit")
    parser.add_argument("--ipv6",action="store_true", help="Scan IPv6 ranges (if provided)")
    parser.add_argument("--no-smart-concurrency",action="store_true", help="Disable adaptive inflight control")
    parser.add_argument("--tui",action="store_true", help="Interactive live TUI (htop-like)")
    parser.add_argument("--sort-key",type=str,default="score", help="Initial sort key: score|lat|jitter|loss|tls|dl|ul")
    parser.add_argument("--cdn-check",action="store_true", help="Smart CDN detection (headers/trace; best-effort)")
    parser.add_argument("--strict-cdn",action="store_true", help="If CDN not detected, mark as dirty (requires --cdn-check)")
    parser.add_argument("--icmp-loss",action="store_true", help="Real ICMP packet loss (system ping; best-effort)")
    parser.add_argument("--tcp-loss",action="store_true", help="Real TCP connect loss (best-effort)")
    parser.add_argument("--http2",action="store_true", help="Probe HTTP/2 support (real ALPN handshake)")
    parser.add_argument("--http3",action="store_true", help="Probe HTTP/3 support (Alt-Svc detection)")
    parser.add_argument("--stability",type=int,default=0, help="Stability mode seconds (e.g., --stability 60)")
    parser.add_argument("--stability-interval",type=float,default=5.0, help="Stability probe interval seconds")
    parser.add_argument("--stability-top",type=int,default=10, help="How many top clean IPs to run stability on")
    parser.add_argument("--no-real-delay",action="store_true", help="Disable real tunnel delay probe via SOCKS connect")
    parser.add_argument("--real-delay-host",type=str,default=DEFAULT_REAL_DELAY_HOST, help="Host for real delay probe (default: 1.1.1.1)")
    parser.add_argument("--real-delay-port",type=int,default=DEFAULT_REAL_DELAY_PORT, help="Port for real delay probe (default: 443)")
    # Geo/ASN enrichment removed by request (kept out of the UI).
    parser.add_argument("--no-color",action="store_true")
    args=parser.parse_args()
    if args.no_color: Colors.disable()

    if args.update_ranges:
        update_ranges_now()
        return

    if len(sys.argv)==1:
        cfg, ports, rng, scan_order, workers, xray_workers, timeout, max_lat, count, tdl, tul, dl_kb, ul_kb, speed_rounds, speed_seconds, speed_kb, latency_urls, stage2_top, stability, stability_interval, stability_top, use_cache, debug, tui, sort_key, real_delay = wizard()
        sc=CleanIPScanner(
            cfg, rng, ports, workers, timeout, max_lat, count,
            tdl, tul, dl_kb, ul_kb, speed_rounds,
            speed_seconds=speed_seconds,
            speed_kb=speed_kb,
            latency_urls=latency_urls,
            stage2_top=stage2_top,
            stability=stability,
            stability_interval=stability_interval,
            stability_top=stability_top,
            use_cache=use_cache,
            debug=debug,
            ip_version=4,
            smart_concurrency=True,
            tui=tui,
            sort_key=sort_key,
            scan_order=scan_order,
            xray_workers=xray_workers,
            real_delay=real_delay,
            real_delay_host=DEFAULT_REAL_DELAY_HOST,
            real_delay_port=DEFAULT_REAL_DELAY_PORT,
        )
        sc.run()
        return

    config_str = read_text_file(args.config_file) if args.config_file else (args.config or "")
    if not config_str:
        err("No config provided. Use -c or -cf"); raise SystemExit(1)
    ip_ranges = None
    if getattr(args, "ip_file", None):
        ip_ranges = args.ip_file
    elif getattr(args, "ip", None):
        import re as _re
        tmp = []
        for it in args.ip:
            tmp.extend([t for t in _re.split(r"[\s,]+", (it or "").strip()) if t])
        ip_ranges = tmp
    else:
        ip_ranges = read_text_file(args.range_file) if args.range_file else (args.range or "")
    if not ip_ranges:
        err("No targets provided. Use -r/-rf (ranges) or --ip/--ip-file (custom targets)"); raise SystemExit(1)
    ports=[int(p.strip()) for p in args.ports.split(",") if p.strip().isdigit()] or [443]
    latency_urls=None
    if args.endpoints:
        latency_urls=[x.strip() for x in args.endpoints.split(",") if x.strip()]
    use_cache=(not args.no_cache)
    cache_reset=args.refresh_cache
    sc=CleanIPScanner(
        config_str, ip_ranges, ports, args.workers, args.timeout, args.max_latency, args.count,
        not args.no_download, not args.no_upload, args.download_size, args.upload_size, args.speed_rounds,
        speed_seconds=args.speed_seconds,
        speed_kb=args.speed_kb,
        latency_urls=latency_urls or DEFAULT_LATENCY_URLS,
        stage2_top=args.stage2_top,
        stability=args.stability,
        stability_interval=args.stability_interval,
        stability_top=args.stability_top,
        use_cache=use_cache,
        debug=args.debug,
        ip_version=6 if args.ipv6 else 4,
        smart_concurrency=(not args.no_smart_concurrency),
        tui=args.tui,
        sort_key=args.sort_key,
        cdn_check=args.cdn_check,
        strict_cdn=args.strict_cdn,
        icmp_loss=args.icmp_loss,
        tcp_loss=args.tcp_loss,
        test_h2=args.http2,
        test_h3=args.http3,
        xray_workers=args.xray_workers,
        real_delay=(not args.no_real_delay),
        real_delay_host=args.real_delay_host,
        real_delay_port=args.real_delay_port,
        # Geo/ASN enrichment removed
        geoip=False,
        ipinfo_token="",
    )
    if cache_reset:
        sc._cache={}
    sc.run()


if __name__=="__main__":
    main()