proxy_checker.py

from __future__ import annotations

import argparse
import asyncio
import concurrent.futures
import json
import os
import re
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable

import httpx
import requests


PROXY_SOURCES = {
    "http": [
        "https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/http.txt",
        "https://raw.githubusercontent.com/monosans/proxy-list/main/proxies/http.txt",
        "https://raw.githubusercontent.com/r00tee/Proxy-List/refs/heads/main/Https.txt",
        "https://raw.githubusercontent.com/Vann-Dev/proxy-list/refs/heads/main/proxies/http.txt",
        "https://raw.githubusercontent.com/Vann-Dev/proxy-list/refs/heads/main/proxies/https.txt",
        "https://raw.githubusercontent.com/elliottophellia/proxylist/master/results/pmix_checked.txt",
        "https://raw.githubusercontent.com/proxifly/free-proxy-list/main/proxies/protocols/http/data.txt",
        "https://raw.githubusercontent.com/ALIILAPRO/Proxy/main/http.txt",
        "https://raw.githubusercontent.com/andigwandi/free-proxy/main/proxy_list.txt",
        "https://raw.githubusercontent.com/ErcinDedeoglu/proxies/main/proxies/http.txt",
        "https://raw.githubusercontent.com/mmpx12/proxy-list/master/proxies.txt",
        "https://raw.githubusercontent.com/monosans/proxy-list/main/proxies/all.txt",
        "https://raw.githubusercontent.com/roosterkid/openproxylist/main/HTTPS_RAW.txt",
        "https://raw.githubusercontent.com/SevenworksDev/proxy-list/main/proxies/unknown.txt",
        "https://raw.githubusercontent.com/sunny9577/proxy-scraper/master/proxies.txt",
        "https://raw.githubusercontent.com/themiralay/Proxy-List-World/master/data.txt",
        "https://raw.githubusercontent.com/Tsprnay/Proxy-lists/master/proxies/http.txt",
        "https://raw.githubusercontent.com/TuanMinPay/live-proxy/master/http.txt",
        "https://raw.githubusercontent.com/Zaeem20/FREE_PROXIES_LIST/master/http.txt",
        "https://raw.githubusercontent.com/zevtyardt/proxy-list/main/http.txt",
        "https://raw.githubusercontent.com/dinoz0rg/proxy-list/main/scraped_proxies/http.txt",
        "https://raw.githubusercontent.com/zebbern/Proxy-Scraper/main/http.txt",
        "https://raw.githubusercontent.com/vmheaven/VMHeaven-Free-Proxy-Updated/main/http.txt",
        "https://raw.githubusercontent.com/saisuiu/Lionkings-Http-Proxys-Proxies/refs/heads/main/free.txt",
        "https://raw.githubusercontent.com/FifzzSENZE/Master-Proxy/master/proxies/http.txt",
        "https://raw.githubusercontent.com/fyvri/fresh-proxy-list/archive/storage/classic/http.txt",
        "https://github.com/handeveloper1/Proxy/raw/refs/heads/main/Proxies-Ercin/http.txt",
        "https://github.com/Anonym0usWork1221/Free-Proxies/raw/refs/heads/main/proxy_files/http_proxies.txt",
        "https://raw.githubusercontent.com/BreakingTechFr/Proxy_Free/refs/heads/main/proxies/http.txt",
        "https://raw.githubusercontent.com/VolkanSah/Auto-Proxy-Fetcher/refs/heads/main/proxies.txt",
        "https://raw.githubusercontent.com/databay-labs/free-proxy-list/refs/heads/master/http.txt",
        "https://raw.githubusercontent.com/ProxyScraper/ProxyScraper/refs/heads/main/http.txt",
        "https://raw.githubusercontent.com/variableninja/proxyscraper/refs/heads/main/proxies/http.txt",
        "https://raw.githubusercontent.com/berkay-digital/Proxy-Scraper/refs/heads/main/proxies.txt",
        "https://github.com/XigmaDev/proxy/raw/refs/heads/main/proxies.txt",
        "https://github.com/chekamarue/proxies/raw/refs/heads/main/https.txt",
        "https://github.com/chekamarue/proxies/raw/refs/heads/main/httpss.txt",
        "https://github.com/claude89757/free_https_proxies/raw/refs/heads/main/https_proxies.txt",
        "https://github.com/claude89757/free_https_proxies/raw/refs/heads/main/isz_https_proxies.txt",
        "https://raw.githubusercontent.com/joy-deploy/free-proxy-list/refs/heads/main/data/latest/types/http/proxies.txt",
        "https://github.com/andigwandi/free-proxy/raw/refs/heads/main/proxy_list.txt",
        "https://raw.githubusercontent.com/parserpp/ip_ports/refs/heads/main/proxyinfo.txt",
        "https://raw.githubusercontent.com/Firmfox/Proxify/refs/heads/main/proxies/http.txt",
        "https://raw.githubusercontent.com/ebrasha/abdal-proxy-hub/refs/heads/main/http-proxy-list-by-EbraSha.txt",
        "https://rootjazz.com/proxies/proxies.txt",
        "https://api.proxyscrape.com/v3/free-proxy-list/get?request=displayproxies&protocol=http&proxy_format=ipport&format=text&timeout=20000",
        "https://proxyspace.pro/http.txt",
    ],
    "socks5": [
        "https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/socks5.txt",
        "https://raw.githubusercontent.com/r00tee/Proxy-List/refs/heads/main/Socks5.txt",
        "https://raw.githubusercontent.com/monosans/proxy-list/main/proxies/socks5.txt",
        "https://raw.githubusercontent.com/Vann-Dev/proxy-list/refs/heads/main/proxies/socks5.txt",
        "https://raw.githubusercontent.com/proxifly/free-proxy-list/main/proxies/protocols/socks5/data.txt",
        "https://raw.githubusercontent.com/ALIILAPRO/Proxy/main/socks5.txt",
        "https://raw.githubusercontent.com/andigwandi/free-proxy/main/proxy_list.txt",
        "https://raw.githubusercontent.com/ErcinDedeoglu/proxies/main/proxies/socks5.txt",
        "https://raw.githubusercontent.com/hookzof/socks5_list/master/proxy.txt",
        "https://raw.githubusercontent.com/roosterkid/openproxylist/main/SOCKS5_RAW.txt",
        "https://raw.githubusercontent.com/SevenworksDev/proxy-list/main/proxies/socks5.txt",
        "https://raw.githubusercontent.com/SevenworksDev/proxy-list/main/proxies/unknown.txt",
        "https://raw.githubusercontent.com/sunny9577/proxy-scraper/master/proxies.txt",
        "https://raw.githubusercontent.com/themiralay/Proxy-List-World/master/data.txt",
        "https://raw.githubusercontent.com/Tsprnay/Proxy-lists/master/proxies/socks5.txt",
        "https://raw.githubusercontent.com/TuanMinPay/live-proxy/master/socks5.txt",
        "https://raw.githubusercontent.com/vakhov/fresh-proxy-list/master/socks5.txt",
        "https://raw.githubusercontent.com/Zaeem20/FREE_PROXIES_LIST/master/socks5.txt",
        "https://raw.githubusercontent.com/zevtyardt/proxy-list/main/socks5.txt",
        "https://raw.githubusercontent.com/dinoz0rg/proxy-list/main/scraped_proxies/socks5.txt",
        "https://raw.githubusercontent.com/zebbern/Proxy-Scraper/main/socks5.txt",
        "https://raw.githubusercontent.com/vmheaven/VMHeaven-Free-Proxy-Updated/main/socks5.txt",
        "https://raw.githubusercontent.com/FifzzSENZE/Master-Proxy/master/proxies/socks5.txt",
        "https://raw.githubusercontent.com/fyvri/fresh-proxy-list/archive/storage/classic/socks5.txt",
        "https://github.com/handeveloper1/Proxy/raw/refs/heads/main/Proxies-Ercin/socks5.txt",
        "https://github.com/Anonym0usWork1221/Free-Proxies/raw/refs/heads/main/proxy_files/socks5_proxies.txt",
        "https://raw.githubusercontent.com/BreakingTechFr/Proxy_Free/refs/heads/main/proxies/socks5.txt",
        "https://raw.githubusercontent.com/VolkanSah/Auto-Proxy-Fetcher/refs/heads/main/proxies.txt",
        "https://raw.githubusercontent.com/databay-labs/free-proxy-list/refs/heads/master/socks5.txt",
        "https://raw.githubusercontent.com/ProxyScraper/ProxyScraper/refs/heads/main/socks5.txt",
        "https://raw.githubusercontent.com/variableninja/proxyscraper/refs/heads/main/proxies/socks.txt",
        "https://raw.githubusercontent.com/joy-deploy/free-proxy-list/refs/heads/main/data/latest/types/socks5/proxies.txt",
        "https://raw.githubusercontent.com/parserpp/ip_ports/refs/heads/main/proxyinfo.txt",
        "https://raw.githubusercontent.com/Firmfox/Proxify/refs/heads/main/proxies/socks5.txt",
        "https://raw.githubusercontent.com/ebrasha/abdal-proxy-hub/refs/heads/main/socks5-proxy-list-by-EbraSha.txt",
    ],
}

PROXY_PATTERN = re.compile(
    r"(?:(?P<scheme>https?|socks4|socks5)://)?"
    r"(?P<host>(?:\d{1,3}\.){3}\d{1,3}):(?P<port>\d{1,5})",
    re.IGNORECASE,
)

HTTP_SCHEMES = {"http", "https"}
USER_AGENT = "proxy-checker/1.0"


def normalize_proxy(proxy_type: str, scheme: str | None, host: str, port: str) -> str | None:
    scheme = scheme.lower() if scheme else None
    if scheme:
        if proxy_type == "http" and scheme not in HTTP_SCHEMES:
            return None
        if proxy_type != "http" and scheme != proxy_type:
            return None

    octets = host.split(".")
    if any(int(octet) > 255 for octet in octets):
        return None

    port_number = int(port)
    if not 1 <= port_number <= 65535:
        return None

    return f"{host}:{port_number}"


def extract_proxies(proxy_type: str, text: str) -> set[str]:
    proxies: set[str] = set()
    for match in PROXY_PATTERN.finditer(text):
        proxy = normalize_proxy(
            proxy_type,
            match.group("scheme"),
            match.group("host"),
            match.group("port"),
        )
        if proxy:
            proxies.add(proxy)
    return proxies


def fetch_source(proxy_type: str, url: str, timeout: float) -> tuple[str, str, set[str], str | None]:
    try:
        response = requests.get(
            url,
            timeout=timeout,
            headers={"User-Agent": USER_AGENT},
        )
        response.raise_for_status()
        return proxy_type, url, extract_proxies(proxy_type, response.text), None
    except requests.RequestException as exc:
        return proxy_type, url, set(), str(exc)


def fetch_all_sources(fetch_workers: int, timeout: float) -> tuple[dict[str, set[str]], list[dict[str, str]]]:
    merged = {proxy_type: set() for proxy_type in PROXY_SOURCES}
    failures: list[dict[str, str]] = []
    jobs = [
        (proxy_type, url)
        for proxy_type, urls in PROXY_SOURCES.items()
        for url in urls
    ]

    with concurrent.futures.ThreadPoolExecutor(max_workers=fetch_workers) as executor:
        futures = [
            executor.submit(fetch_source, proxy_type, url, timeout)
            for proxy_type, url in jobs
        ]
        for future in concurrent.futures.as_completed(futures):
            proxy_type, url, proxies, error = future.result()
            merged[proxy_type].update(proxies)
            if error:
                failures.append({"type": proxy_type, "url": url, "error": error})

    return merged, failures


def proxy_url(proxy_type: str, proxy: str) -> str:
    if proxy_type == "http":
        return f"http://{proxy}"
    return f"{proxy_type}://{proxy}"


def is_valid_probe_response(response: httpx.Response) -> bool:
    return response.status_code == 204 and not response.content


async def probe_url(client: httpx.AsyncClient, url: str) -> tuple[bool, float | None]:
    started_at = time.perf_counter()
    try:
        response = await client.get(url)
        latency = round(time.perf_counter() - started_at, 3)
        return is_valid_probe_response(response), latency
    except (httpx.HTTPError, OSError, ImportError):
        return False, None


async def check_proxy(
    proxy_type: str,
    proxy: str,
    test_url: str,
    https_test_url: str,
    timeout: float,
) -> tuple[str, bool, float | None, bool, float | None]:
    url = proxy_url(proxy_type, proxy)
    client_timeout = httpx.Timeout(timeout)
    limits = httpx.Limits(max_connections=2, max_keepalive_connections=0)
    try:
        async with httpx.AsyncClient(
            proxy=url,
            timeout=client_timeout,
            headers={"User-Agent": USER_AGENT},
            follow_redirects=False,
            limits=limits,
        ) as client:
            (http_ok, http_latency), (https_ok, https_latency) = await asyncio.gather(
                probe_url(client, test_url),
                probe_url(client, https_test_url),
            )
    except (httpx.HTTPError, OSError, ImportError):
        return proxy, False, None, False, None
    return proxy, http_ok, http_latency, https_ok, https_latency


def check_proxy_batch(
    proxy_type: str,
    proxies: Iterable[str],
    check_workers: int,
    test_url: str,
    https_test_url: str,
    timeout: float,
    show_progress: bool,
) -> dict[str, list[dict[str, float | str]]]:
    return asyncio.run(
        check_proxy_batch_async(
            proxy_type,
            proxies,
            check_workers,
            test_url,
            https_test_url,
            timeout,
            show_progress,
        )
    )


async def check_proxy_batch_async(
    proxy_type: str,
    proxies: Iterable[str],
    check_workers: int,
    test_url: str,
    https_test_url: str,
    timeout: float,
    show_progress: bool,
) -> dict[str, list[dict[str, float | str]]]:
    proxy_list = list(proxies)
    alive = {"http": [], "https": []}
    total = len(proxy_list)
    if not total:
        if show_progress:
            print(f"check {proxy_type}: 0/0 http_alive=0 https_alive=0", flush=True)
        return alive

    if show_progress:
        print(f"check {proxy_type}: starting {total} proxies with {check_workers} workers", flush=True)

    checked = 0
    failed = 0
    worker_count = max(1, min(check_workers, total))
    report_every = max(1, min(1000, total // 100 or 1))
    queue: asyncio.Queue[str | None] = asyncio.Queue()
    for proxy in proxy_list:
        queue.put_nowait(proxy)
    for _ in range(worker_count):
        queue.put_nowait(None)

    async def worker() -> None:
        nonlocal checked, failed
        while True:
            proxy = await queue.get()
            try:
                if proxy is None:
                    return
                result = await check_proxy(proxy_type, proxy, test_url, https_test_url, timeout)
                _, http_ok, http_latency, https_ok, https_latency = result
                checked += 1
                if http_ok:
                    alive["http"].append({"proxy": proxy, "latency": http_latency or 0.0})
                if https_ok:
                    alive["https"].append({"proxy": proxy, "latency": https_latency or 0.0})
                if not http_ok and not https_ok:
                    failed += 1
                if show_progress and (checked == total or checked % report_every == 0):
                    percent = checked / total * 100
                    print(
                        f"check {proxy_type}: {checked}/{total} "
                        f"({percent:.1f}%) http_alive={len(alive['http'])} "
                        f"https_alive={len(alive['https'])} failed={failed}",
                        flush=True,
                    )
            finally:
                queue.task_done()

    workers = [asyncio.create_task(worker()) for _ in range(worker_count)]
    await queue.join()
    await asyncio.gather(*workers)

    return {
        target: sorted(items, key=lambda item: (float(item["latency"]), str(item["proxy"])))
        for target, items in alive.items()
    }


def write_lines(path: Path, lines: Iterable[str]) -> None:
    data = "\n".join(lines)
    if data:
        data += "\n"
    path.write_text(data, encoding="utf-8")


def read_lines(path: Path) -> list[str]:
    if not path.exists():
        return []
    return [
        line.strip()
        for line in path.read_text(encoding="utf-8").splitlines()
        if line.strip()
    ]


def read_merged_outputs(output_dir: Path) -> dict[str, set[str]]:
    merged_dir = output_dir / "merged"
    return {
        proxy_type: set(read_lines(merged_dir / f"{proxy_type}.txt"))
        for proxy_type in PROXY_SOURCES
    }


def remove_legacy_live_outputs(live_dir: Path) -> None:
    for name in ("all", *PROXY_SOURCES.keys()):
        path = live_dir / f"{name}.txt"
        if path.exists():
            path.unlink()


def write_fetch_outputs(
    output_dir: Path,
    merged: dict[str, set[str]],
    failures: list[dict[str, str]],
    started_at: float,
) -> None:
    merged_dir = output_dir / "merged"
    merged_dir.mkdir(parents=True, exist_ok=True)

    for proxy_type in PROXY_SOURCES:
        write_lines(merged_dir / f"{proxy_type}.txt", sorted(merged[proxy_type]))

    stats = {
        "phase": "fetch",
        "updated_at": datetime.now(timezone.utc).isoformat(),
        "duration_seconds": round(time.perf_counter() - started_at, 3),
        "counts": {
            proxy_type: {
                "sources": len(PROXY_SOURCES[proxy_type]),
                "merged": len(merged[proxy_type]),
            }
            for proxy_type in PROXY_SOURCES
        },
        "source_failures": failures,
    }
    (output_dir / "stats.json").write_text(
        json.dumps(stats, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )


def read_source_failures(output_dir: Path) -> list[dict[str, str]]:
    stats_path = output_dir / "stats.json"
    if not stats_path.exists():
        return []
    try:
        stats = json.loads(stats_path.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        return []
    failures = stats.get("source_failures", [])
    return failures if isinstance(failures, list) else []


def write_check_outputs(
    output_dir: Path,
    merged: dict[str, set[str]],
    alive: dict[str, dict[str, list[dict[str, float | str]]]],
    failures: list[dict[str, str]],
    started_at: float,
) -> None:
    live_dir = output_dir / "live"
    live_dir.mkdir(parents=True, exist_ok=True)
    remove_legacy_live_outputs(live_dir)

    for target in ("http", "https"):
        target_dir = live_dir / target
        target_dir.mkdir(parents=True, exist_ok=True)
        for proxy_type in PROXY_SOURCES:
            write_lines(
                target_dir / f"{proxy_type}.txt",
                [str(item["proxy"]) for item in alive[proxy_type][target]],
            )

        write_lines(
            target_dir / "all.txt",
            [
                f"{proxy_type}://{item['proxy']}"
                for proxy_type in PROXY_SOURCES
                for item in alive[proxy_type][target]
            ],
        )

    stats = {
        "phase": "check",
        "updated_at": datetime.now(timezone.utc).isoformat(),
        "duration_seconds": round(time.perf_counter() - started_at, 3),
        "counts": {
            proxy_type: {
                "sources": len(PROXY_SOURCES[proxy_type]),
                "merged": len(merged[proxy_type]),
                "http_alive": len(alive[proxy_type]["http"]),
                "https_alive": len(alive[proxy_type]["https"]),
            }
            for proxy_type in PROXY_SOURCES
        },
        "source_failures": failures,
    }
    (output_dir / "stats.json").write_text(
        json.dumps(stats, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Fetch, deduplicate and check free proxies.")
    parser.add_argument(
        "command",
        nargs="?",
        choices=("all", "fetch", "check"),
        default="all",
        help="all: fetch and check, fetch: only write merged proxies, check: read merged proxies and write live proxies",
    )
    parser.add_argument("--output-dir", default=os.getenv("OUTPUT_DIR", "proxies"))
    parser.add_argument("--fetch-workers", type=int, default=int(os.getenv("FETCH_WORKERS", "64")))
    parser.add_argument("--check-workers", type=int, default=int(os.getenv("CHECK_WORKERS", "3000")))
    parser.add_argument("--fetch-timeout", type=float, default=float(os.getenv("FETCH_TIMEOUT", "20")))
    parser.add_argument("--check-timeout", type=float, default=float(os.getenv("CHECK_TIMEOUT", "5")))
    parser.add_argument(
        "--test-url",
        default=os.getenv("TEST_URL", "http://connectivitycheck.gstatic.com/generate_204"),
    )
    parser.add_argument(
        "--https-test-url",
        default=os.getenv("HTTPS_TEST_URL", "https://www.gstatic.com/generate_204"),
    )
    parser.add_argument("--no-progress", action="store_true", help="disable check progress logs")
    return parser.parse_args()


def fetch_command(args: argparse.Namespace) -> dict[str, set[str]]:
    started_at = time.perf_counter()
    merged, failures = fetch_all_sources(args.fetch_workers, args.fetch_timeout)
    write_fetch_outputs(Path(args.output_dir), merged, failures, started_at)

    for proxy_type in PROXY_SOURCES:
        print(f"fetch {proxy_type}: merged={len(merged[proxy_type])}", flush=True)
    if failures:
        print(f"source_failures={len(failures)}", flush=True)
    return merged


def check_command(args: argparse.Namespace, merged: dict[str, set[str]] | None = None) -> None:
    started_at = time.perf_counter()
    output_dir = Path(args.output_dir)
    if merged is None:
        merged = read_merged_outputs(output_dir)

    alive = {
        proxy_type: check_proxy_batch(
            proxy_type,
            proxies,
            args.check_workers,
            args.test_url,
            args.https_test_url,
            args.check_timeout,
            not args.no_progress,
        )
        for proxy_type, proxies in merged.items()
    }
    write_check_outputs(
        output_dir,
        merged,
        alive,
        read_source_failures(output_dir),
        started_at,
    )

    for proxy_type in PROXY_SOURCES:
        print(
            f"result {proxy_type}: merged={len(merged[proxy_type])} "
            f"http_alive={len(alive[proxy_type]['http'])} "
            f"https_alive={len(alive[proxy_type]['https'])}",
            flush=True,
        )


def main() -> int:
    args = parse_args()
    if args.command == "fetch":
        fetch_command(args)
    elif args.command == "check":
        check_command(args)
    else:
        merged = fetch_command(args)
        check_command(args, merged)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())