Як написати проксі-чекер на Python

Проксі-чекер на Python перевіряє функціональність, швидкість, анонімність та географічне розташування проксі-сервера, здійснюючи через нього HTTP-запити до відомої цільової URL-адреси та аналізуючи відповідь.

Основи перевірки проксі

Суть проксі-чекера полягає у надсиланні HTTP-запиту через вказаний проксі та оцінці результату. Цей процес зазвичай використовує бібліотеку requests, яка спрощує HTTP-взаємодію в Python.

Базова перевірка HTTP-проксі

Щоб перевірити базовий HTTP або HTTPS проксі, налаштуйте словник proxies у методі requests.get(). Надійна цільова URL-адреса, така як httpbin.org/ip або ipinfo.io/json, є важливою. Ці сервіси повертають публічну IP-адресу, з якої надійшов запит, що дозволяє здійснити перевірку.

import requests
import time

def check_http_proxy(proxy_address: str, timeout: float = 10.0) -> dict:
    """
    Checks an HTTP/HTTPS proxy for basic connectivity and response time.

    Args:
        proxy_address: The proxy string (e.g., "http://user:pass@ip:port").
        timeout: Maximum time in seconds to wait for a response.

    Returns:
        A dictionary with proxy status, response time, and error message if any.
    """
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    target_url = "http://httpbin.org/ip" # Use http for httpbin, or https for ipinfo
    start_time = time.time()
    try:
        response = requests.get(target_url, proxies=proxies, timeout=timeout)
        response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
        end_time = time.time()
        latency = round((end_time - start_time) * 1000, 2) # Latency in ms
        origin_ip = response.json().get('origin')

        return {
            "proxy": proxy_address,
            "status": "Alive",
            "latency_ms": latency,
            "origin_ip": origin_ip,
            "error": None
        }
    except requests.exceptions.Timeout:
        return {"proxy": proxy_address, "status": "Timeout", "latency_ms": None, "origin_ip": None, "error": "Request timed out"}
    except requests.exceptions.ConnectionError:
        return {"proxy": proxy_address, "status": "Dead", "latency_ms": None, "origin_ip": None, "error": "Connection failed"}
    except requests.exceptions.HTTPError as e:
        return {"proxy": proxy_address, "status": "Error", "latency_ms": None, "origin_ip": None, "error": f"HTTP Error: {e}"}
    except Exception as e:
        return {"proxy": proxy_address, "status": "Error", "latency_ms": None, "origin_ip": None, "error": f"An unexpected error occurred: {e}"}

# Example usage:
# print(check_http_proxy("http://203.0.113.45:8080"))
# print(check_http_proxy("https://198.51.100.22:443"))

Обробка різних типів проксі

Бібліотека requests нативно підтримує HTTP та HTTPS проксі. Для SOCKS проксі (SOCKS4 та SOCKS5) потрібна додаткова залежність requests[socks]. Встановіть її за допомогою pip install requests[socks]. Схема URL проксі тоді змінюється на socks5:// або socks4://.

# Example for SOCKS5 proxy
def check_socks_proxy(proxy_address: str, timeout: float = 10.0) -> dict:
    """
    Checks a SOCKS proxy. Requires 'requests[socks]' to be installed.
    """
    proxies = {
        "http": f"socks5://{proxy_address}",
        "https": f"socks5://{proxy_address}",
    }
    target_url = "http://httpbin.org/ip"
    start_time = time.time()
    try:
        response = requests.get(target_url, proxies=proxies, timeout=timeout)
        response.raise_for_status()
        end_time = time.time()
        latency = round((end_time - start_time) * 1000, 2)
        origin_ip = response.json().get('origin')

        return {
            "proxy": proxy_address,
            "status": "Alive",
            "latency_ms": latency,
            "origin_ip": origin_ip,
            "error": None
        }
    except requests.exceptions.Timeout:
        return {"proxy": proxy_address, "status": "Timeout", "latency_ms": None, "origin_ip": None, "error": "Request timed out"}
    except requests.exceptions.ConnectionError:
        return {"proxy": proxy_address, "status": "Dead", "latency_ms": None, "origin_ip": None, "error": "Connection failed"}
    except Exception as e:
        return {"proxy": proxy_address, "status": "Error", "latency_ms": None, "origin_ip": None, "error": f"SOCKS error: {e}"}

# Example usage:
# print(check_socks_proxy("user:pass@192.0.2.10:1080"))

Оцінка якості проксі

Окрім базового підключення, якість проксі визначається його швидкістю, анонімністю та географічним розташуванням.

Швидкість та затримка

Затримка вимірюється часом, що минув між надсиланням запиту та отриманням першого байта відповіді. Функція time.time() може використовуватися для фіксації часу початку та закінчення, при цьому різниця вказує на затримку. Вона зазвичай повідомляється в мілісекундах.

# The latency calculation is already included in the check_http_proxy and check_socks_proxy functions.
# latency = round((end_time - start_time) * 1000, 2) # Latency in ms

Рівні анонімності

Анонімність проксі стосується того, наскільки ефективно проксі приховує оригінальну IP-адресу клієнта та його ідентичність. Це визначається аналізом конкретних HTTP-заголовків, які проксі може додавати або змінювати. Сервіси, такі як httpbin.org/headers або ipinfo.io/json, можуть використовуватися для перевірки цих заголовків.

• Елітний/Висока анонімність: Проксі приховує IP-адресу клієнта і не надсилає жодних ідентифікуючих заголовків, таких як X-Forwarded-For, Via або Proxy-Connection. Цільовий сервер бачить лише IP-адресу проксі.
• Анонімний: Проксі приховує IP-адресу клієнта, але може надсилати заголовки, такі як Via або X-Forwarded-For з IP-адресою проксі, вказуючи на використання проксі. Оригінальна IP-адреса клієнта не розкривається безпосередньо.
• Прозорий: Проксі пересилає IP-адресу клієнта в заголовках, таких як X-Forwarded-For. Цільовий сервер знає, що запит надійшов від проксі, і може ідентифікувати оригінальну IP-адресу клієнта.

Щоб перевірити анонімність, надішліть запит на URL-адресу, яка повертає заголовки запиту, а потім перевірте відповідь. Вам також потрібно знати свою власну публічну IP-адресу (без проксі) для порівняння.

import requests

def get_my_ip():
    """Returns the client's public IP address without a proxy."""
    try:
        response = requests.get("http://httpbin.org/ip", timeout=5)
        response.raise_for_status()
        return response.json().get('origin')
    except requests.exceptions.RequestException:
        return None

def check_anonymity(proxy_address: str, my_ip: str, timeout: float = 10.0) -> str:
    """
    Checks the anonymity level of a proxy.
    """
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    target_url = "http://httpbin.org/headers" # Returns all request headers
    try:
        response = requests.get(target_url, proxies=proxies, timeout=timeout)
        response.raise_for_status()
        headers = response.json().get('headers', {})

        # Check for X-Forwarded-For, Via, Proxy-Connection headers
        if 'X-Forwarded-For' in headers and headers['X-Forwarded-For'] == my_ip:
            return "Transparent"
        if 'Via' in headers or 'X-Forwarded-For' in headers:
            return "Anonymous"

        # If the origin IP (from httpbin.org/ip) is different from my_ip and no identifying headers are present.
        # This requires a separate check to httpbin.org/ip through the proxy.
        # For a more precise check, one would compare the 'origin' IP from httpbin.org/ip through proxy
        # with my_ip and then check headers.

        # Simplified check: if no common identifying headers, assume Elite (requires careful validation)
        return "Elite"

    except requests.exceptions.RequestException:
        return "Unknown (Proxy Dead or Error)"

# Example usage:
# my_public_ip = get_my_ip()
# if my_public_ip:
#     print(f"My IP: {my_public_ip}")
#     print(check_anonymity("http://203.0.113.45:8080", my_public_ip))

Геолокація

Геолокація ідентифікує фізичне розташування (країна, місто, регіон) проксі-сервера. Сервіси, такі як ipinfo.io/json, надають цю інформацію у відповіді JSON на основі IP-адреси походження запиту.

import requests

def get_proxy_geolocation(proxy_address: str, timeout: float = 10.0) -> dict:
    """
    Retrieves geolocation data for a proxy.
    """
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    target_url = "https://ipinfo.io/json" # ipinfo.io is good for geolocation
    try:
        response = requests.get(target_url, proxies=proxies, timeout=timeout)
        response.raise_for_status()
        data = response.json()
        return {
            "country": data.get('country'),
            "region": data.get('region'),
            "city": data.get('city'),
            "org": data.get('org'),
            "timezone": data.get('timezone')
        }
    except requests.exceptions.RequestException:
        return {"country": None, "region": None, "city": None, "org": None, "timezone": None}

# Example usage:
# geo_data = get_proxy_geolocation("http://203.0.113.45:8080")
# print(geo_data)

Створення паралельного чекера

Послідовна перевірка проксі є неефективною для великих списків. Паралелізм, використовуючи багатопоточність, дозволяє перевіряти кілька проксі одночасно. concurrent.futures.ThreadPoolExecutor у Python спрощує це.

import concurrent.futures
import csv
import time

# Assume check_http_proxy, check_anonymity, get_proxy_geolocation are defined above

def comprehensive_proxy_check(proxy_address: str, my_ip: str, timeout: float = 10.0) -> dict:
    """
    Performs a comprehensive check for a single proxy.
    """
    result = check_http_proxy(proxy_address, timeout) # Basic connectivity and latency
    if result["status"] == "Alive":
        # Only proceed with anonymity and geolocation if proxy is alive
        anonymity = check_anonymity(proxy_address, my_ip, timeout)
        geolocation = get_proxy_geolocation(proxy_address, timeout)
        result.update({"anonymity": anonymity})
        result.update(geolocation)
    else:
        result.update({"anonymity": None, "country": None, "region": None, "city": None, "org": None, "timezone": None})
    return result

def run_concurrent_checker(proxy_list_file: str, output_file: str, max_workers: int = 10, timeout: float = 10.0):
    """
    Reads proxies from a file, checks them concurrently, and writes results to CSV.
    """
    proxies_to_check = []
    try:
        with open(proxy_list_file, 'r') as f:
            for line in f:
                proxy = line.strip()
                if proxy:
                    proxies_to_check.append(proxy)
    except FileNotFoundError:
        print(f"Error: Proxy list file '{proxy_list_file}' not found.")
        return

    my_ip = get_my_ip()
    if not my_ip:
        print("Could not determine client's public IP. Anonymity checks might be inaccurate.")

    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_proxy = {executor.submit(comprehensive_proxy_check, proxy, my_ip, timeout): proxy for proxy in proxies_to_check}
        for future in concurrent.futures.as_completed(future_to_proxy):
            proxy_address = future_to_proxy[future]
            try:
                result = future.result()
                results.append(result)
                print(f"Checked {proxy_address}: Status={result['status']}, Latency={result['latency_ms']}ms, Anonymity={result['anonymity']}")
            except Exception as exc:
                print(f"Proxy {proxy_address} generated an exception: {exc}")
                results.append({"proxy": proxy_address, "status": "Error", "error": str(exc)})

    # Write results to CSV
    if results:
        fieldnames = list(results[0].keys())
        with open(output_file, 'w', newline='') as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(results)
        print(f"Results written to {output_file}")
    else:
        print("No proxies checked or no results to write.")

# Example usage:
# Create a file named 'proxies.txt' with one proxy per line, e.g.:
# http://user:pass@192.0.2.1:8080
# socks5://user:pass@198.51.100.2:1080
# http://203.0.113.3:80
# run_concurrent_checker('proxies.txt', 'proxy_results.csv', max_workers=20, timeout=15)

Надійна обробка помилок та найкращі практики

• Конкретні винятки: Перехоплюйте конкретні requests.exceptions (наприклад, Timeout, ConnectionError, HTTPError), щоб надавати детальний зворотний зв'язок. Загальний перехват Exception має бути останнім засобом.
• User-Agent: Включайте заголовок User-Agent у запити, щоб імітувати браузер і уникнути блокування цільовими сервісами.
  headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'}
• Вибір цільової URL-адреси: Вибирайте надійні, швидкі та публічно доступні сервіси, які надають необхідну інформацію (IP, заголовки, геолокація). Уникайте надмірного використання одного сервісу, щоб запобігти блокуванню IP-адреси.
• Введення/Виведення: Зчитуйте списки проксі з текстових файлів (один проксі на рядок) та записуйте результати у структуровані формати, такі як CSV або JSON, для легкого аналізу та інтеграції.
• Автентифікація проксі: Переконайтеся, що URL-адреси проксі правильно відформатовані для автентифікації (наприклад, http://user:pass@ip:port).
• Повторні спроби: Для чекерів промислового рівня розгляньте можливість реалізації логіки повторних спроб для тимчасових помилок. Бібліотека requests може бути розширена за допомогою requests.adapters.HTTPAdapter та urllib3.util.retry.Retry для цієї мети.