Proxy rotasyonu

Proxy rotasyonu; ağ isteklerinin kaynağını gizlemek, rate limit'leri aşmak ve anonimliği korumak için bir IP adresi (proxy) listesi üzerinde otomatik olarak sırayla geçiş yapma sürecidir. Bu teknik; web scraping, pazar araştırması ve reklam doğrulaması gibi yüksek hacimli istek gerektiren işlemler için temeldir. Bu tür işlemlerde tek bir IP adresinin sürekli kullanılması rate limit, CAPTCHA veya doğrudan yasaklarla sonuçlanır.

Proxy rotasyonu neden gerekli?

Proxy rotasyonunun temel amacı, ağ trafiğini birden fazla IP adresine dağıtmaktır. Web siteleri ve çevrimiçi servisler, tek bir IP adresinden veya dar bir IP aralığından gelen otomatik istekleri tespit edip engellemek için gelişmiş algılama mekanizmaları kullanır. Proxy'ler döndürüldüğünde her istek ya da istek dizisi farklı bir konumdan ve cihazdan geliyormuş gibi görünür; bu da hedef sistemlerin istemciyi tespit edip engellemesini belirgin şekilde zorlaştırır.

Proxy rotasyonunun başlıca faydaları:
* Rate limit'leri aşma: Birçok servis, bir IP'nin belirli bir zaman aralığında yapabileceği istek sayısını sınırlar. Rotasyon bu sınırların üzerine çıkmayı mümkün kılar.
* IP yasaklarını önleme: Tek bir IP'den gelen kesintisiz etkinlik yasaklamaya yol açabilir. IP'leri döndürmek bu riski azaltır.
* Anonimliği koruma: İstemcinin gerçek IP adresini gizleyerek gizliliği artırır.
* Coğrafi kısıtlı içeriğe erişim: Farklı coğrafi konumlardaki proxy'leri kullanarak belirli bölgelere kısıtlanmış içeriğe erişebilirsiniz.
* Yük dağıtımı: Yükü birden fazla ağ çıkış noktasına yayar.

"Proxy havuzu", her istek için içinden bir IP adresi seçilen, kullanılabilir proxy sunucularının bütünüdür. Etkili bir proxy rotasyonu, bu havuzun yönetimine ve uygun bir seçim algoritmasının uygulanmasına dayanır.

Proxy rotasyon algoritmaları

Proxy rotasyonu için, kullanım senaryosuna göre kendine özgü avantaj ve dezavantajları olan çeşitli algoritmalar vardır.

Basit sıralı rotasyon

Bu algoritma proxy havuzunu sabit bir sırayla dolaşır. Her istek listedeki bir sonraki proxy'yi kullanır; listenin sonuna gelindiğinde başa dönülür.

Özellikler:
* Öngörülebilir: Proxy sırası bellidir.
* Eşit dağılım: Zaman içinde tüm proxy'lerin eşit oranda kullanılmasını sağlar.
* Basitlik: Uygulaması kolaydır.

Sınırlamalar:
* Engellenen bir proxy, elle çıkarılana veya atlanana kadar sırayı durdurabilir.
* Sıralı IP kullanım örüntülerini izleyebilen gelişmiş algılamaya karşı daha az etkilidir.

Python uygulama örneği:

import itertools

class SequentialProxyRotator:
    def __init__(self, proxies):
        self.proxies = proxies
        self.proxy_cycle = itertools.cycle(self.proxies)

    def get_next_proxy(self):
        return next(self.proxy_cycle)

# Example Usage:
proxy_list = [
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000",
]
rotator = SequentialProxyRotator(proxy_list)

print(rotator.get_next_proxy()) # http://user:[email protected]:8000
print(rotator.get_next_proxy()) # http://user:[email protected]:8000
print(rotator.get_next_proxy()) # http://user:[email protected]:8000
print(rotator.get_next_proxy()) # http://user:[email protected]:8000 (cycles back)

Rastgele rotasyon

Bu yaklaşımda her istek için aktif havuzdan rastgele bir proxy seçilir.

Özellikler:
* Öngörülemez: Hedef sistemlerin bir örüntü tespit etmesini zorlaştırır.
* Basit: Standart kütüphane fonksiyonlarıyla kolayca uygulanır.

Sınırlamalar:
* Bazı proxy'ler diğerlerinden daha sık kullanılabilir; bu da belirli IP'lerin daha hızlı tespit edilmesine veya tükenmesine yol açabilir.
* Bozuk bir proxy havuzdan çıkarılmazsa tekrar tekrar seçilebilir.

Python uygulama örneği:

import random

class RandomProxyRotator:
    def __init__(self, proxies):
        self.proxies = proxies

    def get_next_proxy(self):
        return random.choice(self.proxies)

# Example Usage:
proxy_list = [
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000",
]
rotator = RandomProxyRotator(proxy_list)

print(rotator.get_next_proxy()) # Randomly selected
print(rotator.get_next_proxy()) # Randomly selected

Zamanlı rotasyon (LRU — Least Recently Used varyantı)

Bu algoritma, bir sonrakine geçmeden önce bir proxy'yi belirli bir süre boyunca veya sabit sayıda istek için kullanır. Bir varyantı, her proxy'nin son kullanım zamanını izler ve yakın zamanda kullanılmayanlara öncelik verir.

Özellikler:
* Kontrollü kullanım: Bir proxy'nin kısa bir zaman aralığında aşırı kullanılmamasını sağlar.
* Daha küçük iz: Her IP'nin etkinliğini zamana yayar.

Sınırlamalar:
* Her proxy için durum yönetimi gerektirir (son kullanım zamanı, istek sayısı).
* Aktif ve pasif proxy'leri yönetme ihtiyacıyla birlikte karmaşıklık artar.

Python uygulama örneği (kavramsal LRU):

import time
from collections import deque

class TimedProxyRotator:
    def __init__(self, proxies, rotation_interval_seconds=60):
        self.proxies = deque(proxies)
        self.rotation_interval = rotation_interval_seconds
        self.current_proxy = None
        self.last_switch_time = 0

    def get_next_proxy(self):
        if self.current_proxy is None or (time.time() - self.last_switch_time) > self.rotation_interval:
            # Rotate proxy
            if self.current_proxy:
                self.proxies.append(self.current_proxy) # Put current back to end
            self.current_proxy = self.proxies.popleft()
            self.last_switch_time = time.time()
        return self.current_proxy

# Example Usage:
proxy_list = [
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000",
]
rotator = TimedProxyRotator(proxy_list, rotation_interval_seconds=10)

print(f"Initial: {rotator.get_next_proxy()}")
time.sleep(2)
print(f"Still same: {rotator.get_next_proxy()}")
# Simulate waiting for rotation_interval
# time.sleep(10)
# print(f"Rotated: {rotator.get_next_proxy()}")

Sağlık odaklı/uyarlanabilir rotasyon

Bu gelişmiş algoritma her proxy'nin performansını ve güvenilirliğini izler. Sürekli başarısız olan, yavaş çalışan veya yasaklanan proxy'ler aktif havuzdan geçici ya da kalıcı olarak çıkarılır. Yeni veya toparlanan proxy'ler havuza geri eklenir.

Özellikler:
* Yüksek güvenilirlik: Çalışan proxy'lere öncelik vererek istek hatalarını en aza indirir.
* Dinamik havuz: Proxy sağlığındaki değişimlere uyum sağlar.
* En iyi performans: Daha hızlı ve daha güvenilir proxy'leri kullanır.

Sınırlamalar:
* Uygulaması belirgin biçimde daha karmaşıktır.
* Sürekli izleme ve sağlam bir sağlık kontrol mekanizması gerektirir.
* Havuz büyük olduğunda sağlık kontrollerinin ek yükü ciddi boyuta ulaşabilir.

Python uygulama örneği (kavramsal çatı):

import time
import requests

class Proxy:
    def __init__(self, address):
        self.address = address
        self.is_healthy = True
        self.failure_count = 0
        self.last_used = 0
        self.response_times = []

    def mark_unhealthy(self):
        self.is_healthy = False
        self.failure_count += 1
        # Implement logic to temporarily disable or remove after N failures

    def mark_healthy(self):
        self.is_healthy = True
        self.failure_count = 0 # Reset on success

    def record_usage(self):
        self.last_used = time.time()

    def add_response_time(self, r_time):
        self.response_times.append(r_time)
        if len(self.response_times) > 10: # Keep last 10
            self.response_times.pop(0)

    @property
    def avg_response_time(self):
        return sum(self.response_times) / len(self.response_times) if self.response_times else float('inf')

class HealthAwareProxyRotator:
    def __init__(self, proxy_addresses, max_failures=3):
        self.proxies = {addr: Proxy(addr) for addr in proxy_addresses}
        self.max_failures = max_failures
        self.active_proxies = deque([p for p in self.proxies.values() if p.is_healthy])
        self.inactive_proxies = []

    def get_next_proxy(self):
        if not self.active_proxies:
            self.attempt_reactivate_proxies()
            if not self.active_proxies:
                raise Exception("No healthy proxies available.")

        # Simple sequential or random from active, for demonstration
        proxy_obj = self.active_proxies.popleft()
        self.active_proxies.append(proxy_obj) # Put back for sequential-like rotation
        proxy_obj.record_usage()
        return proxy_obj.address

    def report_status(self, proxy_address, success, response_time=None):
        proxy_obj = self.proxies.get(proxy_address)
        if not proxy_obj:
            return

        if success:
            proxy_obj.mark_healthy()
            if response_time is not None:
                proxy_obj.add_response_time(response_time)
        else:
            proxy_obj.mark_unhealthy()
            if proxy_obj.failure_count >= self.max_failures and proxy_obj in self.active_proxies:
                self.active_proxies.remove(proxy_obj)
                self.inactive_proxies.append(proxy_obj)
                print(f"Proxy {proxy_address} moved to inactive due to {proxy_obj.failure_count} failures.")

    def attempt_reactivate_proxies(self):
        # Implement periodic health checks for inactive proxies
        # For simplicity, just move all inactive proxies back to active if they exist
        if self.inactive_proxies:
            print("Attempting to reactivate inactive proxies...")
            for proxy_obj in list(self.inactive_proxies): # Iterate copy to allow modification
                # In a real system, you'd perform a health check here
                # For this example, assume they become healthy after some time
                proxy_obj.mark_healthy()
                self.active_proxies.append(proxy_obj)
                self.inactive_proxies.remove(proxy_obj)
            print(f"Reactivated {len(self.active_proxies)} proxies.")

# Example Usage:
proxy_list = [
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000", # Assume this one fails
    "http://user:[email protected]:8000",
]
rotator = HealthAwareProxyRotator(proxy_list)

# Simulate requests
p1 = rotator.get_next_proxy()
print(f"Using {p1}")
rotator.report_status(p1, True, 0.5)

p2 = rotator.get_next_proxy()
print(f"Using {p2}")
rotator.report_status(p2, False) # Fail
rotator.report_status(p2, False) # Fail
rotator.report_status(p2, False) # Fail - Should be moved to inactive

p3 = rotator.get_next_proxy()
print(f"Using {p3}")
rotator.report_status(p3, True, 0.7)

p_next = rotator.get_next_proxy() # Should now skip p2
print(f"Next active: {p_next}")

# If all active proxies fail, it would attempt reactivation
# rotator.report_status(p1, False, 0.5)
# rotator.report_status(p1, False, 0.5)
# rotator.report_status(p1, False, 0.5)
# rotator.report_status(p3, False, 0.5)
# rotator.report_status(p3, False, 0.5)
# rotator.report_status(p3, False, 0.5)
# print(rotator.get_next_proxy()) # This would trigger reactivation attempt

Ağırlıklı rastgele rotasyon

Bu algoritma her proxy'ye bir ağırlık atar; bu ağırlık proxy'nin seçilme olasılığını etkiler. Ağırlıklar geçmiş başarı oranı, yanıt süresi veya coğrafi konum gibi etkenlere dayanabilir.

Özellikler:
* Öncelikli kullanım: Yüksek performanslı veya belirli proxy'leri öne çıkarır.
* Esnek: Ağırlıklar dinamik olarak ayarlanabilir.

Sınırlamalar:
* Ağırlıkları belirleyip güncelleyecek bir mekanizma gerektirir.
* Ağırlıklar doğru tutulmazsa etkisi azalır.

Python uygulama örneği:

import random

class WeightedRandomProxyRotator:
    def __init__(self, proxies_with_weights):
        # proxies_with_weights is a list of tuples: [("proxy_addr", weight), ...]
        self.proxy_addresses = [pw[0] for pw in proxies_with_weights]
        self.weights = [pw[1] for pw in proxies_with_weights]

    def get_next_proxy(self):
        return random.choices(self.proxy_addresses, weights=self.weights, k=1)[0]

# Example Usage:
weighted_proxy_list = [
    ("http://user:[email protected]:8000", 5), # High weight, used more often
    ("http://user:[email protected]:8000", 1), # Low weight
    ("http://user:[email protected]:8000", 3),
]
rotator = WeightedRandomProxyRotator(weighted_proxy_list)

# print(rotator.get_next_proxy()) # Will likely print 1.1.1.1 more often

Algoritma karşılaştırması

Pratik uygulama ayrıntıları

Proxy havuzu yönetimi

Sağlam bir proxy rotasyon sistemi, proxy havuzunun etkin yönetimini gerektirir.
* Veri yapısı: collections modülündeki deque, verimli append ve popleft işlemleri sayesinde sıralı veya LRU benzeri rotasyon için uygundur. Sağlık odaklı sistemlerde, proxy adreslerini özel Proxy nesnelerine eşleyen bir sözlük (örnekte gösterildiği gibi) meta verileri saklamak için etkilidir.
* Proxy ekleme/çıkarma: Sistem, proxy listesini kesintiye uğramadan dinamik biçimde güncelleyebilmelidir.
* İlk doğrulama: Proxy'ler aktif havuza eklenmeden önce yapılan ilk sağlık kontrolü, çalışırlıklarını ve performanslarını doğrular.

İstek kütüphaneleriyle entegrasyon

HTTP istek kütüphanelerinin çoğu proxy destekler. Python tarafında yaygın olarak requests kütüphanesi kullanılır.

import requests

def make_request_with_proxy(url, proxy_address):
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    try:
        start_time = time.time()
        response = requests.get(url, proxies=proxies, timeout=10)
        end_time = time.time()
        response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
        return response.text, (end_time - start_time)
    except requests.exceptions.RequestException as e:
        print(f"Request failed with proxy {proxy_address}: {e}")
        return None, None

# Example using a rotator:
# rotator = HealthAwareProxyRotator(proxy_list)
# selected_proxy = rotator.get_next_proxy()
# content, r_time = make_request_with_proxy("http://example.com", selected_proxy)
# if content:
#     rotator.report_status(selected_proxy, True, r_time)
# else:
#     rotator.report_status(selected_proxy, False)

Hata yönetimi ve yeniden denemeler

Bir proxy başarısız olduğunda sistem şunları yapmalıdır:
1. Proxy'yi sağlıksız olarak işaretle: Havuzdaki durumunu güncelle.
2. Yeni bir proxy ile yeniden dene: Havuzdan başka bir proxy seçip isteği tekrar gönder.
3. Yeniden deneme sınırı koy: Kalıcı hatalarda sonsuz döngüleri önle.

def robust_request(url, rotator, max_retries=3):
    for _ in range(max_retries):
        try:
            proxy_address = rotator.get_next_proxy()
            print(f"Attempting {url} with {proxy_address}")
            content, r_time = make_request_with_proxy(url, proxy_address)
            if content:
                rotator.report_status(proxy_address, True, r_time)
                return content
            else:
                rotator.report_status(proxy_address, False)
        except Exception as e:
            print(f"Error during request attempt: {e}")
            # The get_next_proxy itself might raise an exception if no healthy proxies
            pass
    raise Exception(f"Failed to fetch {url} after {max_retries} retries.")

# Example usage with the HealthAwareRotator and a dummy URL
# try:
#     final_content = robust_request("http://dummy-url-that-might-fail.com", rotator)
#     print("Request successful.")
# except Exception as e:
#     print(f"Final failure: {e}")

Eşzamanlılık konuları

Çok iş parçacıklı (multi-threaded) veya asenkron ortamlarda proxy rotatörünün durumu (proxy havuzu, geçerli proxy, sağlık metrikleri) thread-safe olmalıdır.
* Kilitler: Paylaşılan proxy havuzu veri yapılarını güncellerken kritik bölümleri korumak için threading.Lock kullanın.
* Kuyruklar: queue.Queue proxy'leri yönetebilir; böylece birden fazla işçi proxy'leri güvenle get ve put edebilir.

import threading
import queue
import time

class ThreadSafeProxyRotator:
    def __init__(self, proxies):
        self.proxy_queue = queue.Queue()
        for p in proxies:
            self.proxy_queue.put(p)
        self.lock = threading.Lock()
        self.in_use = set() # Track proxies currently in use by a thread

    def get_proxy(self):
        with self.lock:
            if self.proxy_queue.empty() and not self.in_use:
                raise Exception("No proxies available in pool.")
            elif self.proxy_queue.empty(): # All proxies are currently in use
                # Implement waiting or re-adding used proxies here for a real system
                # For simplicity, just raise an error for now
                raise Exception("All proxies are currently in use.")

            proxy = self.proxy_queue.get()
            self.in_use.add(proxy)
            return proxy

    def return_proxy(self, proxy, is_healthy=True):
        with self.lock:
            if proxy in self.in_use:
                self.in_use.remove(proxy)
                if is_healthy:
                    self.proxy_queue.put(proxy) # Return to pool
                else:
                    print(f"Proxy {proxy} marked unhealthy and not returned to pool.")
            else:
                print(f"Attempted to return unknown or already returned proxy: {proxy}")

# Example of a worker thread
# def worker(rotator, thread_id):
#     try:
#         proxy = rotator.get_proxy()
#         print(f"Thread {thread_id} using {proxy}")
#         time.sleep(random.uniform(1, 3)) # Simulate work
#         success = random.choice([True, False]) # Simulate success/failure
#         rotator.return_proxy(proxy, success)
#         print(f"Thread {thread_id} finished with {proxy}, success: {success}")
#     except Exception as e:
#         print(f"Thread {thread_id} error: {e}")
#
# proxy_list = ["proxy1", "proxy2", "proxy3"]
# ts_rotator = ThreadSafeProxyRotator(proxy_list)
# threads = []
# for i in range(5):
#     t = threading.Thread(target=worker, args=(ts_rotator, i))
#     threads.append(t)
#     t.start()
#
# for t in threads:
#     t.join()