Wie man einen Proxy-Checker in Python schreibt

Ein Proxy-Checker in Python überprüft die Funktionalität, Geschwindigkeit, Anonymität und den geografischen Standort eines Proxy-Servers, indem er HTTP-Anfragen über diesen an eine bekannte Ziel-URL sendet und die Antwort analysiert.

Grundlagen der Proxy-Überprüfung

Der Kern eines Proxy-Checkers besteht darin, eine HTTP-Anfrage über einen bestimmten Proxy zu senden und das Ergebnis zu bewerten. Dieser Prozess verwendet typischerweise die requests-Bibliothek, die HTTP-Interaktionen in Python vereinfacht.

Grundlegende HTTP-Proxy-Überprüfung

Um einen grundlegenden HTTP- oder HTTPS-Proxy zu überprüfen, konfigurieren Sie das proxies-Wörterbuch in der requests.get()-Methode. Eine zuverlässige Ziel-URL, wie httpbin.org/ip oder ipinfo.io/json, ist unerlässlich. Diese Dienste geben die öffentliche IP-Adresse zurück, von der die Anfrage stammt, was eine Überprüfung ermöglicht.

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"))

Umgang mit verschiedenen Proxy-Typen

Die requests-Bibliothek unterstützt nativ HTTP- und HTTPS-Proxys. Für SOCKS-Proxys (SOCKS4 und SOCKS5) ist eine zusätzliche Abhängigkeit requests[socks] erforderlich. Installieren Sie diese mit pip install requests[socks]. Das Proxy-URL-Schema ändert sich dann zu socks5:// oder 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"))

Bewertung der Proxy-Qualität

Über die grundlegende Konnektivität hinaus wird die Qualität eines Proxys durch seine Geschwindigkeit, Anonymität und seinen geografischen Standort definiert.

Geschwindigkeit und Latenz

Die Latenz wird durch die Zeit gemessen, die zwischen dem Senden der Anfrage und dem Empfangen des ersten Bytes der Antwort vergeht. Die Funktion time.time() kann verwendet werden, um Start- und Endzeiten zu erfassen, wobei die Differenz die Latenz angibt. Sie wird typischerweise in Millisekunden angegeben.

# 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

Anonymitätsstufen

Die Proxy-Anonymität bezieht sich darauf, wie effektiv ein Proxy die ursprüngliche IP-Adresse und Identität des Clients verbirgt. Dies wird durch die Analyse spezifischer HTTP-Header bestimmt, die der Proxy hinzufügen oder ändern könnte. Dienste wie httpbin.org/headers oder ipinfo.io/json können verwendet werden, um diese Header zu überprüfen.

• Elite/Hohe Anonymität: Der Proxy verbirgt die IP-Adresse des Clients und sendet keine identifizierenden Header wie X-Forwarded-For, Via oder Proxy-Connection. Der Zielserver sieht nur die IP des Proxys.
• Anonym: Der Proxy verbirgt die IP-Adresse des Clients, kann aber Header wie Via oder X-Forwarded-For mit der IP des Proxys senden, was auf die Verwendung eines Proxys hinweist. Die ursprüngliche IP des Clients wird nicht direkt offengelegt.
• Transparent: Der Proxy leitet die IP-Adresse des Clients in Headern wie X-Forwarded-For weiter. Der Zielserver weiß, dass die Anfrage von einem Proxy kam und kann die ursprüngliche IP des Clients identifizieren.

Um die Anonymität zu überprüfen, senden Sie eine Anfrage an eine URL, die die Anfrage-Header zurückgibt, und überprüfen Sie dann die Antwort. Sie müssen auch Ihre eigene öffentliche IP-Adresse (ohne Proxy) zum Vergleich kennen.

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))

Geolokalisierung

Die Geolokalisierung identifiziert den physischen Standort (Land, Stadt, Region) des Proxy-Servers. Dienste wie ipinfo.io/json liefern diese Informationen in einer JSON-Antwort basierend auf der Ursprungs-IP der Anfrage.

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)

Aufbau eines nebenläufigen Checkers

Das sequentielle Überprüfen von Proxys ist für große Listen ineffizient. Nebenläufigkeit, unter Verwendung von Multithreading, ermöglicht die gleichzeitige Überprüfung mehrerer Proxys. Pythons concurrent.futures.ThreadPoolExecutor vereinfacht dies.

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)

Robuste Fehlerbehandlung und Best Practices

• Spezifische Ausnahmen: Fangen Sie spezifische requests.exceptions (z.B. Timeout, ConnectionError, HTTPError) ab, um detailliertes Feedback zu geben. Ein allgemeiner Exception-Fang sollte eine letzte Möglichkeit sein.
• User-Agent: Fügen Sie einen User-Agent-Header in Anfragen ein, um einen Browser zu imitieren und zu vermeiden, von Zieldiensten blockiert zu werden.
  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'}
• Ziel-URL-Auswahl: Wählen Sie zuverlässige, schnelle und öffentlich verfügbare Dienste, die die notwendigen Informationen (IP, Header, Geolokalisierung) bereitstellen. Vermeiden Sie es, einen einzelnen Dienst zu überlasten, um IP-Sperren zu verhindern.
• Ein-/Ausgabe: Lesen Sie Proxy-Listen aus Textdateien (ein Proxy pro Zeile) und schreiben Sie die Ergebnisse in strukturierte Formate wie CSV oder JSON zur einfachen Analyse und Integration.
• Proxy-Authentifizierung: Stellen Sie sicher, dass Proxy-URLs für die Authentifizierung korrekt formatiert sind (z.B. http://user:pass@ip:port).
• Wiederholungen: Für Checker in Produktionsqualität sollten Sie eine Wiederholungslogik für vorübergehende Fehler in Betracht ziehen. Die requests-Bibliothek kann zu diesem Zweck mit requests.adapters.HTTPAdapter und urllib3.util.retry.Retry erweitert werden.