proxies.py

import json
import re
from datetime import datetime
from pathlib import Path

import requests

BASE_DIR = Path(__file__).resolve().parent
PROXIES_TXT_PATH = BASE_DIR / "proxies.txt"
PROXIES_JSON_PATH = BASE_DIR / "proxies.json"
SUPPORTED_SCHEMES = {"http", "https"}
PROXY_PATTERN = re.compile(r"(?:(?P<scheme>https?)://)?(?P<ip>(?:\d{1,3}\.){3}\d{1,3}):(?P<port>\d{1,5})")


def clean_host_port(ip_address: str, port_text: str) -> str | None:
    try:
        octets = [int(part) for part in ip_address.split(".")]
        port = int(port_text)
    except ValueError:
        return None

    if len(octets) != 4 or any(octet < 0 or octet > 255 for octet in octets):
        return None

    if port < 1 or port > 65535:
        return None

    cleaned_ip = ".".join(str(octet) for octet in octets)
    return f"{cleaned_ip}:{port}"


def parse_proxy_entries(raw_text: str) -> dict[str, set[str]]:
    inventory: dict[str, set[str]] = {}

    for match in PROXY_PATTERN.finditer(raw_text):
        scheme = (match.group("scheme") or "").strip().lower()
        host_port = clean_host_port(match.group("ip"), match.group("port"))
        if not host_port:
            continue

        schemes = {scheme} if scheme in SUPPORTED_SCHEMES else set(SUPPORTED_SCHEMES)
        inventory.setdefault(host_port, set()).update(schemes)

    return inventory


def write_proxy_outputs(proxy_inventory: dict[str, set[str]]) -> None:
    sorted_hosts = sorted(proxy_inventory)
    proxy_urls: list[str] = []
    proxy_items: list[dict] = []

    for host_port in sorted_hosts:
        candidate_schemes = sorted(proxy_inventory[host_port])
        urls = [f"{scheme}://{host_port}" for scheme in candidate_schemes]
        proxy_urls.extend(urls)
        proxy_items.append(
            {
                "proxy_host_port": host_port,
                "candidate_schemes": candidate_schemes,
                "proxy_urls": urls,
            }
        )

    PROXIES_TXT_PATH.write_text("\n".join(proxy_urls) + ("\n" if proxy_urls else ""), encoding="utf-8")
    PROXIES_JSON_PATH.write_text(
        json.dumps(
            {
                "generated_at": datetime.now().isoformat(),
                "total_proxy_hosts": len(sorted_hosts),
                "total_proxy_urls": len(proxy_urls),
                "proxies": proxy_items,
            },
            indent=2,
        ),
        encoding="utf-8",
    )


def collect_proxies() -> None:
    sources = [
        "https://api.proxyscrape.com/v4/free-proxy-list/get?request=display_proxies&proxy_format=protocolipport&format=text"
    ]

    merged_inventory: dict[str, set[str]] = {}

    print("--- Proxy Collection ---")
    for url in sources:
        try:
            response = requests.get(url, timeout=15)
            response.raise_for_status()
            source_inventory = parse_proxy_entries(response.text)

            for host_port, schemes in source_inventory.items():
                merged_inventory.setdefault(host_port, set()).update(schemes)

            print(
                f"Success: {url[:60]}... "
                f"({len(source_inventory)} hosts, "
                f"{sum(len(schemes) for schemes in source_inventory.values())} typed proxy URLs)"
            )
        except Exception as exc:
            print(f"Error: {url[:60]}... : {exc}")

    write_proxy_outputs(merged_inventory)

    total_urls = sum(len(schemes) for schemes in merged_inventory.values())
    print("")
    print(f"Done. Saved {len(merged_inventory)} proxy hosts to '{PROXIES_TXT_PATH.name}'.")
    print(f"Typed proxy inventory: {PROXIES_JSON_PATH.name} ({total_urls} proxy URLs).")


if __name__ == "__main__":
    collect_proxies()