check_proxies.py

"""Asynchronously check proxy lists and record results."""

import argparse
import asyncio
import ipaddress
import json
import logging
import ssl
from datetime import datetime
from pathlib import Path

import aiohttp
import certifi
from colorama import Fore, Style, init as colorama_init
from tqdm import tqdm

from proxy_parser import load_domains, load_proxy_file, normalize_proxy, select_domains

try:
    from aiohttp_socks import ProxyConnector
    from python_socks import ProxyError as SocksProxyError
except ImportError:
    ProxyConnector = None

    class SocksProxyError(Exception):
        """Fallback SOCKS exception type when optional deps are missing."""

# ------------------- CONFIGURATION -------------------
PROXY_DIR = Path("proxy")                       # Folder with proxy lists
PROXY_TYPES = ("http", "socks4", "socks5")
CHECK_SERVICE_URL = "https://api.myip.com"
GEOLOOKUP_URL = "https://ipwho.is/{ip}"
GEO_CACHE_FILE = "geo_ip_cache.json"
OK_PROXIES_WITH_IP_FILE = "ok_proxies_with_ip.txt"  # Working proxies with all observed IPs
OK_PROXIES_FILE = "ok_proxies.txt"             # Working proxies only (no IPs)
BAD_PROXIES_FILE = "bad_proxies.txt"           # Proxies that never returned IP
DOMAIN_RESULTS_FILE = "domain_check_results.jsonl"
LOG_FILE = "actions.log"                       # Log file name
TIMEOUT = 5                                    # Request timeout in seconds
MAX_CONCURRENCY = 200                          # Max simultaneous proxy checks
RETRIES = 1                                    # Additional retries per proxy check
RETRY_BACKOFF = 0.2                            # Base retry delay in seconds
GEO_MAX_CONCURRENCY = 20                       # Max simultaneous geolocation lookups
GEO_RETRIES = 3                                # Additional retries for geolocation lookups
GEO_RETRY_BACKOFF = 1.0                        # Base geolocation retry delay in seconds
GEO_RPS = 5.0                                  # Rate limit for geolocation requests per second
# ------------------------------------------------------

# Initialize colorama for colored console output
colorama_init(autoreset=True)

# Create SSL context using certifi CA bundle
ssl_ctx = ssl.create_default_context(cafile=certifi.where())

# Configure logger
logger = logging.getLogger("ProxyChecker")
logger.setLevel(logging.DEBUG)

# File logging (detailed)
file_handler = logging.FileHandler(LOG_FILE, mode="w", encoding="utf-8")
file_handler.setLevel(logging.DEBUG)
file_formatter = logging.Formatter(
    "%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
)
file_handler.setFormatter(file_formatter)

# Console logging (colored)
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)


class ColorFormatter(logging.Formatter):
    """Custom formatter to add colors to console log levels."""
    COLORS = {
        logging.DEBUG: Style.DIM,
        logging.INFO: Fore.CYAN,
        logging.WARNING: Fore.YELLOW,
        logging.ERROR: Fore.RED,
        logging.CRITICAL: Fore.RED + Style.BRIGHT,
    }

    def format(self, record):
        color = self.COLORS.get(record.levelno, "")
        return f"{color}{super().format(record)}{Style.RESET_ALL}"


console_formatter = ColorFormatter("%(message)s")
console_handler.setFormatter(console_formatter)

# Attach handlers to logger
logger.addHandler(file_handler)
logger.addHandler(console_handler)


def parse_args():
    """Parse CLI args."""
    parser = argparse.ArgumentParser(
        description="Asynchronously check proxy list from proxy/<file_name>.",
    )
    parser.add_argument("proxy_file_name", help="Proxy list file name inside proxy/ folder")
    parser.add_argument(
        "--proxy-type",
        "-t",
        choices=PROXY_TYPES,
        default="http",
        help="Proxy type for entries without scheme (default: http)",
    )
    parser.add_argument(
        "--iterations",
        "-i",
        type=int,
        default=1,
        help="Number of check iterations (default: 1)",
    )
    parser.add_argument(
        "--resolve-location",
        dest="resolve_location",
        action=argparse.BooleanOptionalAction,
        default=True,
        help="Resolve country code for each proxy IP and write it to output (default: true)",
    )
    parser.add_argument(
        "--check-url",
        default=CHECK_SERVICE_URL,
        help=(
            "URL used to detect proxy egress IP "
            "(default: https://api.myip.com)"
        ),
    )
    parser.add_argument(
        "--max-concurrency",
        type=int,
        default=MAX_CONCURRENCY,
        help="Maximum simultaneous proxy checks (default: 200)",
    )
    parser.add_argument(
        "--retries",
        type=int,
        default=RETRIES,
        help="Additional retries per proxy on transient errors (default: 1)",
    )
    parser.add_argument(
        "--retry-backoff",
        type=float,
        default=RETRY_BACKOFF,
        help="Base retry delay in seconds (default: 0.2)",
    )
    parser.add_argument(
        "--geo-max-concurrency",
        type=int,
        default=GEO_MAX_CONCURRENCY,
        help="Maximum simultaneous geolocation lookups (default: 20)",
    )
    parser.add_argument(
        "--geo-retries",
        type=int,
        default=GEO_RETRIES,
        help="Additional retries for geolocation lookup errors (default: 3)",
    )
    parser.add_argument(
        "--geo-retry-backoff",
        type=float,
        default=GEO_RETRY_BACKOFF,
        help="Base geolocation retry delay in seconds (default: 1.0)",
    )
    parser.add_argument(
        "--geo-rps",
        type=float,
        default=GEO_RPS,
        help="Max geolocation requests per second, 0 disables throttling (default: 5.0)",
    )
    parser.add_argument(
        "--geo-cache-file",
        default=GEO_CACHE_FILE,
        help="Path to local geolocation cache file (default: geo_ip_cache.json)",
    )
    parser.add_argument(
        "--check-domains",
        metavar="N|all",
        help=(
            "Run extra checks for working proxies against N random domains or all domains "
            "from domains.json"
        ),
    )
    parser.add_argument(
        "--domains-file",
        default="domains.json",
        help="Path to domains JSON file for --check-domains (default: domains.json)",
    )
    parser.add_argument(
        "--domain-results-file",
        default=DOMAIN_RESULTS_FILE,
        help=(
            "File for domain check results in JSON Lines format "
            f"(default: {DOMAIN_RESULTS_FILE})"
        ),
    )
    return parser.parse_args()


def extract_ip_from_response(response_body: str):
    """
    Extract an IP address from response text.

    Supported response formats:
      - JSON object with key "ip" / "origin" / "query"
      - JSON string with raw IP
      - Plain text IP
    """
    stripped = response_body.strip()
    if not stripped:
        return None

    parsed = stripped
    try:
        parsed = json.loads(stripped)
    except json.JSONDecodeError:
        parsed = stripped

    candidates = []
    if isinstance(parsed, dict):
        for key in ("ip", "origin", "query"):
            value = parsed.get(key)
            if isinstance(value, str):
                candidates.append(value)
    elif isinstance(parsed, str):
        candidates.append(parsed)

    for candidate in candidates:
        # "origin" may contain a comma-separated IP list.
        first_part = candidate.split(",")[0].strip()
        try:
            ipaddress.ip_address(first_part)
            return first_part
        except ValueError:
            continue
    return None


def format_error(error: BaseException) -> str:
    """Convert exception to a non-empty, readable error message."""
    message = str(error).strip()
    return message if message else error.__class__.__name__


async def check_proxy(
    proxy: str,
    check_url: str,
    session: aiohttp.ClientSession,
    semaphore: asyncio.Semaphore,
    retries: int,
    retry_backoff: float,
):
    """Try connecting to the check URL through the provided proxy."""
    proxy_scheme = proxy.split("://", 1)[0].lower() if "://" in proxy else "http"
    is_socks_proxy = proxy_scheme in {"socks4", "socks5"}

    for attempt in range(retries + 1):
        try:
            async with semaphore:
                if is_socks_proxy:
                    if ProxyConnector is None:
                        return {
                            "status": False,
                            "message": (
                                "SOCKS proxy support requires aiohttp-socks package. "
                                "Install dependencies from requirements.txt"
                            ),
                            "proxy": proxy,
                        }

                    connector = ProxyConnector.from_url(proxy)
                    async with aiohttp.ClientSession(
                        connector=connector,
                        timeout=session.timeout,
                    ) as socks_session:
                        async with socks_session.get(
                            url=check_url,
                            ssl=ssl_ctx,
                        ) as response:
                            response.raise_for_status()
                            body = await response.text()
                else:
                    async with session.get(
                        url=check_url,
                        proxy=proxy,
                        ssl=ssl_ctx,
                    ) as response:
                        response.raise_for_status()
                        body = await response.text()
            ip = extract_ip_from_response(body)
            if ip is None:
                return {
                    "status": False,
                    "message": f"IP not found in response from {check_url}",
                    "proxy": proxy,
                }
            return {"status": True, "message": {"ip": ip}, "proxy": proxy}
        except aiohttp.InvalidURL:
            return {"status": False, "message": f"Invalid proxy URL: {proxy}", "proxy": proxy}
        except (
            aiohttp.ClientError,
            asyncio.TimeoutError,
            ssl.SSLError,
            SocksProxyError,
            OSError,
        ) as error:
            if attempt < retries:
                await asyncio.sleep(retry_backoff * (attempt + 1))
                continue
            return {"status": False, "message": format_error(error), "proxy": proxy}


async def check_domain_with_proxy(
    proxy: str,
    domain: str,
    session: aiohttp.ClientSession,
    semaphore: asyncio.Semaphore,
):
    """Check whether a domain opens through a working proxy."""
    proxy_scheme = proxy.split("://", 1)[0].lower() if "://" in proxy else "http"
    is_socks_proxy = proxy_scheme in {"socks4", "socks5"}
    urls = [f"https://{domain}", f"http://{domain}"]

    async with semaphore:
        for url in urls:
            try:
                if is_socks_proxy:
                    if ProxyConnector is None:
                        return {
                            "status": False,
                            "proxy": proxy,
                            "domain": domain,
                            "url": url,
                            "status_code": None,
                            "message": (
                                "SOCKS proxy support requires aiohttp-socks package. "
                                "Install dependencies from requirements.txt"
                            ),
                        }
                    connector = ProxyConnector.from_url(proxy)
                    async with aiohttp.ClientSession(
                        connector=connector,
                        timeout=session.timeout,
                    ) as socks_session:
                        async with socks_session.get(url=url, ssl=ssl_ctx) as response:
                            return {
                                "status": response.status < 500,
                                "proxy": proxy,
                                "domain": domain,
                                "url": url,
                                "status_code": response.status,
                                "message": response.reason,
                            }
                async with session.get(url=url, proxy=proxy, ssl=ssl_ctx) as response:
                    return {
                        "status": response.status < 500,
                        "proxy": proxy,
                        "domain": domain,
                        "url": url,
                        "status_code": response.status,
                        "message": response.reason,
                    }
            except (
                aiohttp.ClientError,
                asyncio.TimeoutError,
                ssl.SSLError,
                SocksProxyError,
                OSError,
            ) as error:
                last_error = error

        return {
            "status": False,
            "proxy": proxy,
            "domain": domain,
            "url": urls[-1],
            "status_code": None,
            "message": format_error(last_error),
        }


async def run_domain_checks(proxies, domains, max_concurrency: int, output_file: Path):
    """Run domain checks for working proxies and save JSON Lines results."""
    if not proxies or not domains:
        return []

    results = []
    semaphore = asyncio.Semaphore(max_concurrency)
    connector = aiohttp.TCPConnector(
        limit=max_concurrency,
        limit_per_host=max_concurrency,
    )
    client_timeout = aiohttp.ClientTimeout(total=TIMEOUT)

    async with aiohttp.ClientSession(connector=connector, timeout=client_timeout) as session:
        checks = [(proxy, domain) for proxy in proxies for domain in domains]
        batch_size = max(max_concurrency * 10, max_concurrency)
        with tqdm(total=len(checks), desc="Domain checks") as progress:
            for start in range(0, len(checks), batch_size):
                batch = checks[start:start + batch_size]
                tasks = [
                    asyncio.create_task(check_domain_with_proxy(proxy, domain, session, semaphore))
                    for proxy, domain in batch
                ]
                for task in asyncio.as_completed(tasks):
                    result = await task
                    results.append(result)
                    progress.update(1)

    with open(output_file, "w", encoding="utf-8") as file_obj:
        for result in results:
            file_obj.write(json.dumps(result, ensure_ascii=False, sort_keys=True))
            file_obj.write("\n")

    return results


class AsyncRateLimiter:
    """Simple async rate limiter with a global minimum interval."""

    def __init__(self, rate_per_second: float):
        self.interval = 1.0 / rate_per_second if rate_per_second > 0 else 0.0
        self._lock = asyncio.Lock()
        self._next_allowed = 0.0

    async def wait(self):
        """Wait until the next request slot is available."""
        if self.interval <= 0:
            return

        loop = asyncio.get_running_loop()
        async with self._lock:
            now = loop.time()
            if now < self._next_allowed:
                await asyncio.sleep(self._next_allowed - now)
                now = loop.time()
            self._next_allowed = max(self._next_allowed, now) + self.interval


def load_ip_location_cache(cache_file: Path):
    """Load geolocation cache from disk (legacy dict JSON and append-only JSON stream)."""
    if not cache_file.exists():
        return {}

    try:
        content = cache_file.read_text(encoding="utf-8")
    except OSError as error:
        logger.warning("Failed to read geo cache %s: %s", cache_file, error)
        return {}

    content = content.strip()
    if not content:
        return {}

    cache = {}
    decoder = json.JSONDecoder()
    idx = 0
    length = len(content)

    while idx < length:
        while idx < length and content[idx].isspace():
            idx += 1
        if idx >= length:
            break

        try:
            value, next_idx = decoder.raw_decode(content, idx)
        except json.JSONDecodeError as error:
            logger.warning("Failed to parse geo cache %s near pos %s: %s", cache_file, idx, error)
            break

        if isinstance(value, dict):
            # Legacy format: {"ip":"country", ...}
            # Append-only format: {"ip":"1.2.3.4","country":"US"}
            if "ip" in value and "country" in value:
                ip = value.get("ip")
                country = value.get("country")
                if isinstance(ip, str) and isinstance(country, str) and country:
                    cache[ip] = country.upper()
            else:
                for ip, country in value.items():
                    if isinstance(ip, str) and isinstance(country, str) and country:
                        cache[ip] = country.upper()

        idx = next_idx

    return cache


def append_ip_location_cache(cache_file: Path, previous_cache, current_cache):
    """Append only new/updated geolocation entries to cache file."""
    to_append = []
    for ip, country in current_cache.items():
        if not (isinstance(ip, str) and isinstance(country, str) and country and country != "N/A"):
            continue
        if previous_cache.get(ip) == country:
            continue
        to_append.append({"ip": ip, "country": country})

    if not to_append:
        return 0

    try:
        prefix = ""
        if cache_file.exists() and cache_file.stat().st_size > 0:
            prefix = "\n"
        with open(cache_file, "a", encoding="utf-8") as cache_fp:
            if prefix:
                cache_fp.write(prefix)
            for entry in to_append:
                cache_fp.write(json.dumps(entry, ensure_ascii=False, sort_keys=True))
                cache_fp.write("\n")
        return len(to_append)
    except OSError as error:
        logger.warning("Failed to append geo cache %s: %s", cache_file, error)
        return 0


async def resolve_country_for_ip(
    ip: str,
    session: aiohttp.ClientSession,
    geo_retries: int,
    geo_retry_backoff: float,
    rate_limiter: AsyncRateLimiter,
) -> str:
    """Resolve IP country code with retries and rate-limit handling."""
    for attempt in range(geo_retries + 1):
        try:
            await rate_limiter.wait()
            async with session.get(
                url=GEOLOOKUP_URL.format(ip=ip),
                ssl=ssl_ctx,
            ) as response:
                response.raise_for_status()
                data = await response.json()
        except (
            aiohttp.ClientError,
            asyncio.TimeoutError,
            json.JSONDecodeError,
            ssl.SSLError,
        ) as error:
            if attempt < geo_retries:
                await asyncio.sleep(geo_retry_backoff * (attempt + 1))
                continue
            logger.debug("Location lookup failed for %s: %s", ip, format_error(error))
            return "N/A"

        if not isinstance(data, dict):
            logger.debug("Location lookup returned unexpected payload for %s", ip)
            return "N/A"

        country = data.get("country_code")
        if isinstance(country, str) and country:
            return country.upper()

        message = data.get("message")
        if isinstance(message, str) and "rate limit" in message.lower() and attempt < geo_retries:
            await asyncio.sleep(geo_retry_backoff * (attempt + 1))
            continue

        if isinstance(message, str):
            logger.debug("Location lookup returned no country for %s: %s", ip, message)
        return "N/A"

    return "N/A"


async def resolve_locations(
    ips,
    resolve_location: bool,
    ip_location_cache,
    geo_max_concurrency: int,
    geo_retries: int,
    geo_retry_backoff: float,
    geo_rps: float,
):
    """Resolve locations for IPs and update cache."""
    if not resolve_location:
        return

    unresolved_ips = [ip for ip in ips if ip not in ip_location_cache]
    if not unresolved_ips:
        return

    if geo_rps > 0:
        estimated_minutes = len(unresolved_ips) / geo_rps / 60
        logger.info(
            "Resolving geolocation for %s new IPs (rate=%.2f req/s, est ~%.1f min)",
            len(unresolved_ips),
            geo_rps,
            estimated_minutes,
        )
    else:
        logger.info(
            "Resolving geolocation for %s new IPs (unthrottled mode)",
            len(unresolved_ips),
        )

    semaphore = asyncio.Semaphore(geo_max_concurrency)
    connector = aiohttp.TCPConnector(
        limit=geo_max_concurrency,
        limit_per_host=geo_max_concurrency,
    )
    client_timeout = aiohttp.ClientTimeout(total=TIMEOUT)
    rate_limiter = AsyncRateLimiter(geo_rps)

    async def lookup_ip(ip: str, session: aiohttp.ClientSession):
        async with semaphore:
            country = await resolve_country_for_ip(
                ip,
                session,
                geo_retries,
                geo_retry_backoff,
                rate_limiter,
            )
            return ip, country

    async with aiohttp.ClientSession(connector=connector, timeout=client_timeout) as session:
        batch_size = max(geo_max_concurrency * 20, geo_max_concurrency)
        with tqdm(total=len(unresolved_ips), desc="Geo lookup") as geo_progress:
            for start in range(0, len(unresolved_ips), batch_size):
                batch = unresolved_ips[start:start + batch_size]
                tasks = [asyncio.create_task(lookup_ip(ip, session)) for ip in batch]
                for task in asyncio.as_completed(tasks):
                    ip, country = await task
                    ip_location_cache[ip] = country
                    geo_progress.update(1)


async def run_iteration(
    proxy_list,
    iteration_num,
    all_ok_proxies,
    bad_proxy_stats,
    check_url,
    max_concurrency,
    retries,
    retry_backoff,
    resolve_location,
    ip_location_cache,
    geo_max_concurrency,
    geo_retries,
    geo_retry_backoff,
    geo_rps,
):
    """
    Run a single iteration of proxy checks.
    Side effects:
      - Updates all_ok_proxies: dict {proxy: {ip1: country1, ip2: country2}}
      - Updates bad_proxy_stats: dict {
          proxy: {"fail_count": int, "last_error": str, "last_check": str}
        }
    """
    oks = 0
    bads = 0
    results = []
    semaphore = asyncio.Semaphore(max_concurrency)
    connector = aiohttp.TCPConnector(
        limit=max_concurrency,
        limit_per_host=max_concurrency,
    )
    client_timeout = aiohttp.ClientTimeout(total=TIMEOUT)

    async with aiohttp.ClientSession(connector=connector, timeout=client_timeout) as session:
        # Keep a bounded number of in-memory tasks while still streaming progress.
        batch_size = max(max_concurrency * 10, max_concurrency)
        with tqdm(total=len(proxy_list), desc=f"Iteration {iteration_num}") as progress:
            for start in range(0, len(proxy_list), batch_size):
                batch = proxy_list[start:start + batch_size]
                tasks = [
                    asyncio.create_task(
                        check_proxy(
                            proxy,
                            check_url,
                            session,
                            semaphore,
                            retries,
                            retry_backoff,
                        )
                    )
                    for proxy in batch
                ]
                for task in asyncio.as_completed(tasks):
                    result = await task
                    results.append(result)
                    progress.update(1)

    iteration_ips = [
        result["message"]["ip"]
        for result in results
        if result["status"] and isinstance(result["message"], dict) and "ip" in result["message"]
    ]
    await resolve_locations(
        iteration_ips,
        resolve_location,
        ip_location_cache,
        geo_max_concurrency,
        geo_retries,
        geo_retry_backoff,
        geo_rps,
    )

    for result in results:
        proxy = result["proxy"]
        if result["status"] and isinstance(result["message"], dict) and "ip" in result["message"]:
            ip = result["message"]["ip"]
            country_code = ip_location_cache.get(ip, "N/A") if resolve_location else ""
            if resolve_location:
                logger.info("OK: %s -> IP: %s (%s)", proxy, ip, country_code)
            else:
                logger.info("OK: %s -> IP: %s", proxy, ip)
            oks += 1
            # Store all observed IPs with country code (if enabled)
            if proxy not in all_ok_proxies:
                all_ok_proxies[proxy] = {}
            if ip not in all_ok_proxies[proxy]:
                all_ok_proxies[proxy][ip] = country_code
        else:
            err = result["message"]
            logger.warning("BAD: %s -> %s", proxy, err)
            bads += 1
            if proxy not in bad_proxy_stats:
                bad_proxy_stats[proxy] = {"fail_count": 0, "last_error": "", "last_check": ""}
            bad_proxy_stats[proxy]["fail_count"] += 1
            bad_proxy_stats[proxy]["last_error"] = str(err)
            bad_proxy_stats[proxy]["last_check"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    oks_percent = round(oks / len(proxy_list) * 100)
    bads_percent = round(bads / len(proxy_list) * 100)
    logger.info(
        "Iteration %s summary: OK: %s (%s%%) / BAD: %s (%s%%)",
        iteration_num,
        oks,
        oks_percent,
        bads,
        bads_percent,
    )
    return oks, bads


async def main():
    """Load proxies, run checks, and write summary output files."""
    args = parse_args()
    if args.iterations < 1:
        logger.error("Iterations must be >= 1.")
        return
    if args.max_concurrency < 1:
        logger.error("max-concurrency must be >= 1.")
        return
    if args.retries < 0:
        logger.error("retries must be >= 0.")
        return
    if args.retry_backoff < 0:
        logger.error("retry-backoff must be >= 0.")
        return
    if args.geo_max_concurrency < 1:
        logger.error("geo-max-concurrency must be >= 1.")
        return
    if args.geo_retries < 0:
        logger.error("geo-retries must be >= 0.")
        return
    if args.geo_retry_backoff < 0:
        logger.error("geo-retry-backoff must be >= 0.")
        return
    if args.geo_rps < 0:
        logger.error("geo-rps must be >= 0.")
        return
    if args.check_domains is not None:
        if args.check_domains != "all":
            try:
                domain_check_count = int(args.check_domains)
            except ValueError:
                logger.error("check-domains must be a positive integer or 'all'.")
                return
            if domain_check_count < 1:
                logger.error("check-domains count must be >= 1.")
                return
    check_url_lower = args.check_url.lower()
    if not check_url_lower.startswith(("http://", "https://")):
        logger.error("check-url must start with http:// or https://")
        return

    proxy_file = PROXY_DIR / Path(args.proxy_file_name).name

    # Load proxies from file
    try:
        proxy_list, skipped_lines = load_proxy_file(proxy_file, args.proxy_type)
    except FileNotFoundError:
        logger.error("Proxy file '%s' not found.", proxy_file)
        return

    if not proxy_list:
        logger.error("Proxy list is empty.")
        return

    logger.info(
        "Loaded %s proxies from %s (default_type_for_scheme_less=%s)",
        len(proxy_list),
        proxy_file,
        args.proxy_type,
    )
    logger.info("Check URL: %s", args.check_url)
    logger.info(
        "Concurrency=%s, retries=%s, retry_backoff=%ss",
        args.max_concurrency,
        args.retries,
        args.retry_backoff,
    )
    logger.info(
        "Geo settings: concurrency=%s, retries=%s, retry_backoff=%ss, rps=%s, cache=%s",
        args.geo_max_concurrency,
        args.geo_retries,
        args.geo_retry_backoff,
        args.geo_rps,
        args.geo_cache_file,
    )
    if skipped_lines:
        logger.info("Skipped %s non-proxy lines (headers/comments/empty).", skipped_lines)

    total_oks = 0
    total_bads = 0
    all_ok_proxies = {}   # {proxy: {ip1: country1, ip2: country2}}
    bad_proxy_stats = {}  # {proxy: {"fail_count": int, "last_error": str, "last_check": str}}
    geo_cache_file = Path(args.geo_cache_file)
    ip_location_cache = load_ip_location_cache(geo_cache_file) if args.resolve_location else {}
    initial_ip_location_cache = dict(ip_location_cache)
    if args.resolve_location:
        logger.info("Loaded %s geo cache entries.", len(ip_location_cache))

    # Multiple iterations
    for i in range(1, args.iterations + 1):
        oks, bads = await run_iteration(
            proxy_list,
            i,
            all_ok_proxies,
            bad_proxy_stats,
            args.check_url,
            args.max_concurrency,
            args.retries,
            args.retry_backoff,
            args.resolve_location,
            ip_location_cache,
            args.geo_max_concurrency,
            args.geo_retries,
            args.geo_retry_backoff,
            args.geo_rps,
        )
        total_oks += oks
        total_bads += bads

    if args.resolve_location:
        appended_entries = append_ip_location_cache(
            geo_cache_file,
            initial_ip_location_cache,
            ip_location_cache,
        )
        logger.info("Appended %s geo cache entries to %s", appended_entries, geo_cache_file)

    # ---- Write GOOD proxies to two files ----
    # 1) Proxies WITH all observed IPs
    if args.resolve_location:
        ok_with_ip_lines = [
            "{} -> {}".format(
                proxy,
                ", ".join(
                    f"{ip} ({country})" for ip, country in sorted(all_ok_proxies[proxy].items())
                ),
            )
            for proxy in sorted(all_ok_proxies.keys())
        ]
    else:
        ok_with_ip_lines = [
            f"{proxy} -> {sorted(all_ok_proxies[proxy].keys())}"
            for proxy in sorted(all_ok_proxies.keys())
        ]
    Path(OK_PROXIES_WITH_IP_FILE).write_text("\n".join(ok_with_ip_lines), encoding="utf-8")

    # 2) Proxies ONLY (no IPs)
    Path(OK_PROXIES_FILE).write_text(
        "\n".join(proxy for proxy in sorted(all_ok_proxies.keys())),
        encoding="utf-8",
    )

    domain_results = []
    if args.check_domains:
        domains_file = Path(args.domains_file)
        try:
            domains = load_domains(domains_file)
            selected_domains = select_domains(domains, args.check_domains)
        except FileNotFoundError:
            logger.error("Domains file '%s' not found.", domains_file)
            selected_domains = []
        except (json.JSONDecodeError, ValueError) as error:
            logger.error("Failed to load domains from %s: %s", domains_file, error)
            selected_domains = []

        if selected_domains and all_ok_proxies:
            logger.info("%s", "=" * 50)
            logger.info(
                "DOMAIN CHECKS: %s working proxies x %s domains",
                len(all_ok_proxies),
                len(selected_domains),
            )
            domain_results = await run_domain_checks(
                sorted(all_ok_proxies.keys()),
                selected_domains,
                args.max_concurrency,
                Path(args.domain_results_file),
            )
            domain_ok = sum(1 for result in domain_results if result["status"])
            domain_bad = len(domain_results) - domain_ok
            logger.info("Domain OK: %s / BAD: %s", domain_ok, domain_bad)
            logger.info("Domain check results saved to %s", args.domain_results_file)
        elif args.check_domains:
            Path(args.domain_results_file).write_text("", encoding="utf-8")
            logger.info(
                "Domain checks skipped: %s working proxies, %s selected domains.",
                len(all_ok_proxies),
                len(selected_domains),
            )

    # ---- Write NEVER-SUCCESSFUL proxies (sorted by fails desc) ----
    never_ok = {
        proxy: stats
        for proxy, stats in bad_proxy_stats.items()
        if proxy not in all_ok_proxies
    }
    with open(BAD_PROXIES_FILE, "w", encoding="utf-8") as f:
        for proxy, stats in sorted(
            never_ok.items(),
            key=lambda x: x[1]["fail_count"],
            reverse=True,
        ):
            f.write(
                f"{proxy} | Fails: {stats['fail_count']} | "
                f"Last error: {stats['last_error']} | Last check: {stats['last_check']}\n"
            )

    # Final summary
    total_checks = total_oks + total_bads
    success_rate = round(total_oks / total_checks * 100) if total_checks else 0

    logger.info("%s", "=" * 50)
    logger.info("FINAL SUMMARY for %s iterations:", args.iterations)
    logger.info("Total checks: %s", total_checks)
    logger.info("Total OK: %s", total_oks)
    logger.info("Total BAD: %s", total_bads)
    logger.info("Success rate: %s%%", success_rate)
    logger.info("Working proxies with IPs saved to %s", OK_PROXIES_WITH_IP_FILE)
    logger.info("Working proxies only saved to %s", OK_PROXIES_FILE)
    logger.info("Never-successful proxies saved to %s", BAD_PROXIES_FILE)
    logger.info("%s", "=" * 50)


if __name__ == "__main__":
    asyncio.run(main())