Python ile proxy checker nasıl yazılır

Python ile yazılan bir proxy checker, proxy sunucusu üzerinden bilinen bir hedef URL'ye HTTP istekleri göndererek ve yanıtı analiz ederek proxy'nin çalışırlığını, hızını, anonimliğini ve coğrafi konumunu doğrular.

Proxy kontrolünün temelleri

Bir proxy checker'ın özü, belirtilen proxy üzerinden bir HTTP isteği göndermek ve sonucu değerlendirmektir. Bu süreçte genellikle, Python'da HTTP etkileşimlerini basitleştiren requests kütüphanesi kullanılır.

Temel HTTP proxy kontrolü

Temel bir HTTP veya HTTPS proxy'sini kontrol etmek için requests.get() metodundaki proxies sözlüğünü yapılandırın. httpbin.org/ip veya ipinfo.io/json gibi güvenilir bir hedef URL şarttır. Bu servisler isteğin çıktığı genel IP adresini döndürür ve böylece doğrulama yapılabilir.

import requests
import time

def check_http_proxy(proxy_address: str, timeout: float = 10.0) -> dict:
    """
    Bir HTTP/HTTPS proxy'sini temel bağlanabilirlik ve yanıt süresi açısından kontrol eder.

    Args:
        proxy_address: Proxy dizesi (örn. "http://user:pass@ip:port").
        timeout: Yanıt için beklenecek saniye cinsinden maksimum süre.

    Returns:
        Proxy durumunu, yanıt süresini ve varsa hata mesajını içeren bir sözlük.
    """
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    target_url = "http://httpbin.org/ip" # httpbin için http, ipinfo için https kullanın
    start_time = time.time()
    try:
        response = requests.get(target_url, proxies=proxies, timeout=timeout)
        response.raise_for_status() # Hatalı yanıtlarda (4xx veya 5xx) HTTPError fırlatır
        end_time = time.time()
        latency = round((end_time - start_time) * 1000, 2) # Gecikme, ms cinsinden
        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}"}

# Kullanım örneği:
# print(check_http_proxy("http://203.0.113.45:8080"))
# print(check_http_proxy("https://198.51.100.22:443"))

Farklı proxy türleriyle çalışma

requests kütüphanesi HTTP ve HTTPS proxy'lerini yerel olarak destekler. SOCKS proxy'leri (SOCKS4 ve SOCKS5) için ek bir bağımlılık olan requests[socks] gerekir. pip install requests[socks] komutuyla kurun. Ardından proxy URL şeması socks5:// veya socks4:// olur.

# SOCKS5 proxy örneği
def check_socks_proxy(proxy_address: str, timeout: float = 10.0) -> dict:
    """
    Bir SOCKS proxy'sini kontrol eder. 'requests[socks]' kurulu olmalıdır.
    """
    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}"}

# Kullanım örneği:
# print(check_socks_proxy("user:[email protected]:1080"))

Proxy kalitesinin değerlendirilmesi

Temel bağlanabilirliğin ötesinde, bir proxy'nin kalitesi hızı, anonimliği ve coğrafi konumuyla belirlenir.

Hız ve gecikme

Gecikme, isteğin gönderilmesi ile yanıtın ilk baytının alınması arasında geçen süreyle ölçülür. Başlangıç ve bitiş zamanlarını yakalamak için time.time() fonksiyonu kullanılabilir; aradaki fark gecikmeyi verir. Genellikle milisaniye cinsinden raporlanır.

# Gecikme hesabı zaten check_http_proxy ve check_socks_proxy fonksiyonlarında yer alıyor.
# latency = round((end_time - start_time) * 1000, 2) # Gecikme, ms cinsinden

Anonimlik seviyeleri

Proxy anonimliği, proxy'nin istemcinin özgün IP adresini ve kimliğini ne kadar etkili gizlediğini ifade eder. Bu, proxy'nin ekleyebileceği veya değiştirebileceği belirli HTTP başlıkları incelenerek belirlenir. Bu başlıkları incelemek için httpbin.org/headers veya ipinfo.io/json gibi servisler kullanılabilir.

• Elite/Yüksek anonimlik: proxy istemcinin IP adresini gizler ve X-Forwarded-For, Via veya Proxy-Connection gibi tanımlayıcı başlıklar göndermez. Hedef sunucu yalnızca proxy'nin IP'sini görür.
• Anonim: proxy istemcinin IP adresini gizler, ancak proxy'nin IP'siyle Via ya da X-Forwarded-For gibi başlıklar göndererek proxy kullanıldığını belli edebilir. İstemcinin özgün IP'si doğrudan açığa çıkmaz.
• Şeffaf (Transparent): proxy, istemcinin IP adresini X-Forwarded-For gibi başlıklarda iletir. Hedef sunucu isteğin bir proxy'den geldiğini bilir ve özgün istemci IP'sini tespit edebilir.

Anonimliği kontrol etmek için, istek başlıklarını geri döndüren bir URL'ye istek gönderin ve yanıtı inceleyin. Karşılaştırma için kendi genel IP adresinizi de (proxy'siz) bilmeniz gerekir.

import requests

def get_my_ip():
    """İstemcinin proxy'siz genel IP adresini döndürür."""
    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:
    """
    Bir proxy'nin anonimlik seviyesini kontrol eder.
    """
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    target_url = "http://httpbin.org/headers" # Tüm istek başlıklarını döndürür
    try:
        response = requests.get(target_url, proxies=proxies, timeout=timeout)
        response.raise_for_status()
        headers = response.json().get('headers', {})

        # X-Forwarded-For, Via, Proxy-Connection başlıklarını kontrol et
        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"

        # Origin IP (httpbin.org/ip'den) my_ip'den farklıysa ve tanımlayıcı başlık yoksa.
        # Bu, proxy üzerinden httpbin.org/ip'ye ayrı bir kontrol gerektirir.
        # Daha kesin bir kontrol için, proxy üzerinden httpbin.org/ip'den gelen 'origin' IP'yi
        # my_ip ile karşılaştırın ve ardından başlıkları kontrol edin.

        # Basitleştirilmiş kontrol: yaygın tanımlayıcı başlık yoksa Elite varsayılır (dikkatli doğrulama gerektirir)
        return "Elite"

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

# Kullanım örneği:
# 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))

Coğrafi konum

Coğrafi konum, proxy sunucusunun fiziksel konumunu (ülke, şehir, bölge) belirler. ipinfo.io/json gibi servisler bu bilgiyi, isteğin çıkış IP'sine göre JSON yanıtı olarak sağlar.

import requests

def get_proxy_geolocation(proxy_address: str, timeout: float = 10.0) -> dict:
    """
    Bir proxy için coğrafi konum verilerini alır.
    """
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    target_url = "https://ipinfo.io/json" # ipinfo.io coğrafi konum için iyidir
    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}

# Kullanım örneği:
# geo_data = get_proxy_geolocation("http://203.0.113.45:8080")
# print(geo_data)

Eşzamanlı bir checker oluşturma

Proxy'leri sırayla kontrol etmek, büyük listeler için verimsizdir. Çoklu iş parçacığı (multithreading) ile eşzamanlılık, birden fazla proxy'nin aynı anda kontrol edilmesini sağlar. Python'ın concurrent.futures.ThreadPoolExecutor yapısı bunu kolaylaştırır.

import concurrent.futures
import csv
import time

# check_http_proxy, check_anonymity, get_proxy_geolocation fonksiyonlarının yukarıda tanımlı olduğu varsayılır

def comprehensive_proxy_check(proxy_address: str, my_ip: str, timeout: float = 10.0) -> dict:
    """
    Tek bir proxy için kapsamlı kontrol yapar.
    """
    result = check_http_proxy(proxy_address, timeout) # Temel bağlanabilirlik ve gecikme
    if result["status"] == "Alive":
        # Anonimlik ve coğrafi konum kontrolüne yalnızca proxy çalışıyorsa devam et
        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):
    """
    Proxy'leri bir dosyadan okur, eşzamanlı olarak kontrol eder ve sonuçları CSV'ye yazar.
    """
    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)})

    # Sonuçları CSV'ye yaz
    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.")

# Kullanım örneği:
# Her satırda bir proxy olacak şekilde 'proxies.txt' adlı bir dosya oluşturun, örn.:
# http://user:[email protected]:8080
# socks5://user:[email protected]:1080
# http://203.0.113.3:80
# run_concurrent_checker('proxies.txt', 'proxy_results.csv', max_workers=20, timeout=15)

Sağlam hata yönetimi ve en iyi uygulamalar

• Belirli istisnalar: ayrıntılı geri bildirim için belirli requests.exceptions istisnalarını yakalayın (örn. Timeout, ConnectionError, HTTPError). Genel Exception yakalaması son çare olmalıdır.
• User-Agent: bir tarayıcıyı taklit etmek ve hedef servisler tarafından engellenmemek için isteklere User-Agent başlığı ekleyin.
  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'}
• Hedef URL seçimi: gerekli bilgileri (IP, başlıklar, coğrafi konum) sağlayan güvenilir, hızlı ve herkese açık servisleri seçin. IP engeli yememek için tek bir servisi aşırı yüklemekten kaçının.
• Giriş/çıkış: proxy listelerini metin dosyalarından okuyun (her satırda bir proxy) ve sonuçları analiz ve entegrasyon kolaylığı için CSV veya JSON gibi yapılandırılmış formatlara yazın.
• Proxy kimlik doğrulaması: proxy URL'lerinin kimlik doğrulama için doğru biçimde yazıldığından emin olun (örn. http://user:pass@ip:port).
• Yeniden denemeler: üretim düzeyinde checker'larda geçici hatalar için retry mantığı uygulamayı düşünün. requests kütüphanesi bu amaçla requests.adapters.HTTPAdapter ve urllib3.util.retry.Retry ile genişletilebilir.