بازگشت به وبلاگ

تنظیم پروکسی در Python requests و aiohttp: راهنمای کامل با مثال‌های کد

راهنمای گام به گام برای ادغام پروکسی در درخواست‌های Python و aiohttp با مثال‌های کد برای درخواست‌های همزمان و ناهمزمان، چرخش IP و مدیریت خطاها.

📅۲۴ بهمن ۱۴۰۴
```html

هنگام توسعه پارسرها، خودکارسازی جمع‌آوری داده‌ها یا آزمایش خدمات وب از Python، معمولاً نیاز به استفاده از سرورهای پروکسی وجود دارد. کتابخانه‌های requests و aiohttp مکانیزم‌های انعطاف‌پذیری برای کار با پروکسی ارائه می‌دهند، اما تنظیم آن‌ها نکات مهمی دارد. در این راهنما به بررسی رویکردهای همزمان و غیرهمزمان می‌پردازیم، مثال‌هایی برای پروکسی HTTP و SOCKS5 ارائه می‌دهیم، چرخش IP و مدیریت خطاها را بررسی می‌کنیم.

تنظیمات پایه پروکسی در درخواست‌ها

کتابخانه requests استانداردی برای درخواست‌های HTTP در Python است. تنظیم پروکسی از طریق پارامتر proxies انجام می‌شود که دیکشنری‌ای با پروتکل‌ها و آدرس‌های سرورهای پروکسی را می‌پذیرد.

ساده‌ترین مثال با پروکسی HTTP:

import requests

# تنظیم پروکسی
proxies = {
    'http': 'http://123.45.67.89:8080',
    'https': 'http://123.45.67.89:8080'
}

# انجام درخواست از طریق پروکسی
response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())  # {'origin': '123.45.67.89'}

توجه داشته باشید: برای درخواست‌های HTTPS نیز پروتکل http:// در مقدار پروکسی مشخص می‌شود (نه https://). این به این دلیل است که اتصال به سرور پروکسی از طریق HTTP برقرار می‌شود و سپس از طریق متد CONNECT تونلی برای ترافیک HTTPS ایجاد می‌شود.

استفاده از متغیرهای محیطی:

کتابخانه requests به طور خودکار پروکسی را از متغیرهای محیطی HTTP_PROXY و HTTPS_PROXY می‌خواند:

import os
import requests

# تنظیم از طریق متغیرهای محیطی
os.environ['HTTP_PROXY'] = 'http://123.45.67.89:8080'
os.environ['HTTPS_PROXY'] = 'http://123.45.67.89:8080'

# پروکسی به طور خودکار اعمال می‌شود
response = requests.get('https://httpbin.org/ip')
print(response.json())

این رویکرد برای کانتینرها (Docker) یا زمانی که پروکسی‌ها در سطح سیستم تنظیم می‌شوند، مناسب است. با این حال، برای انعطاف‌پذیری، توصیه می‌شود که پارامتر proxies به صورت صریح منتقل شود.

احراز هویت و SOCKS5 در درخواست‌ها

بیشتر خدمات پروکسی تجاری نیاز به احراز هویت با نام کاربری و رمز عبور دارند. در requests این موضوع از طریق فرمت URL با اطلاعات کاربری پیاده‌سازی می‌شود.

پروکسی HTTP با احراز هویت:

import requests

# فرمت: http://username:password@host:port
proxies = {
    'http': 'http://user123:pass456@proxy.example.com:8080',
    'https': 'http://user123:pass456@proxy.example.com:8080'
}

response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())

تنظیم پروکسی SOCKS5:

برای کار با SOCKS5 به کتابخانه اضافی requests[socks] یا PySocks نیاز است. نصب:

pip install requests[socks]

مثال استفاده از SOCKS5:

import requests

# SOCKS5 بدون احراز هویت
proxies = {
    'http': 'socks5://123.45.67.89:1080',
    'https': 'socks5://123.45.67.89:1080'
}

# SOCKS5 با احراز هویت
proxies_auth = {
    'http': 'socks5://user:pass@123.45.67.89:1080',
    'https': 'socks5://user:pass@123.45.67.89:1080'
}

response = requests.get('https://httpbin.org/ip', proxies=proxies_auth)
print(response.json())

پروکسی SOCKS5 به ویژه هنگام کار با پروکسی‌های مسکونی مفید است، زیرا این پروتکل تونل‌سازی ترافیک را به طور قابل اعتمادتری فراهم می‌کند و از UDP پشتیبانی می‌کند (که برای برخی از برنامه‌ها ضروری است).

چرخش پروکسی در درخواست‌ها

هنگام پارس کردن حجم‌های بزرگ داده، استفاده از یک آدرس IP منجر به مسدود شدن می‌شود. چرخش پروکسی به معنای تغییر دوره‌ای IP برای توزیع بار و دور زدن محدودیت‌های نرخ است.

چرخش ساده از طریق لیست:

import requests
import itertools

# لیست سرورهای پروکسی
proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
    'http://user:pass@proxy3.example.com:8080',
]

# ایجاد یک ایتراکتور بی‌پایان
proxy_pool = itertools.cycle(proxy_list)

# انجام درخواست‌ها با چرخش
for i in range(10):
    proxy = next(proxy_pool)
    proxies = {'http': proxy, 'https': proxy}
    
    try:
        response = requests.get('https://httpbin.org/ip', proxies=proxies, timeout=5)
        print(f"درخواست {i+1}: IP = {response.json()['origin']}")
    except Exception as e:
        print(f"خطا با پروکسی {proxy}: {e}")

چرخش با نشست‌ها برای حفظ کوکی‌ها:

import requests
from itertools import cycle

class ProxyRotator:
    def __init__(self, proxy_list):
        self.proxy_pool = cycle(proxy_list)
        self.session = requests.Session()
    
    def get(self, url, **kwargs):
        proxy = next(self.proxy_pool)
        self.session.proxies = {'http': proxy, 'https': proxy}
        return self.session.get(url, **kwargs)

# استفاده
proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
]

rotator = ProxyRotator(proxy_list)

for i in range(5):
    response = rotator.get('https://httpbin.org/ip', timeout=5)
    print(f"درخواست {i+1}: {response.json()['origin']}")

چرخش تصادفی برای غیرقابل پیش‌بینی بودن:

import requests
import random

proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
    'http://user:pass@proxy3.example.com:8080',
]

def get_random_proxy():
    proxy = random.choice(proxy_list)
    return {'http': proxy, 'https': proxy}

# هر درخواست با پروکسی تصادفی
for i in range(5):
    response = requests.get('https://httpbin.org/ip', proxies=get_random_proxy(), timeout=5)
    print(f"درخواست {i+1}: {response.json()['origin']}")

چرخش تصادفی هنگام کار با سایت‌هایی که الگوهای درخواست را ردیابی می‌کنند، مؤثرتر است. تغییر متوالی IP ممکن است مشکوک به نظر برسد، در حالی که انتخاب تصادفی رفتار کاربران مختلف را شبیه‌سازی می‌کند.

تنظیم پروکسی در aiohttp

کتابخانه aiohttp برای درخواست‌های HTTP غیرهمزمان طراحی شده و برای پارسرهای با بار بالا حیاتی است. تنظیم پروکسی در اینجا متفاوت از requests است و از پارامتر proxy (در حالت مفرد) استفاده می‌شود.

مثال پایه با پروکسی HTTP:

import aiohttp
import asyncio

async def fetch_with_proxy():
    proxy = 'http://123.45.67.89:8080'
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', proxy=proxy) as response:
            data = await response.json()
            print(data)

# راه‌اندازی
asyncio.run(fetch_with_proxy())

پروکسی با احراز هویت:

در aiohttp، احراز هویت از طریق شیء aiohttp.BasicAuth یا مستقیماً در URL منتقل می‌شود:

import aiohttp
import asyncio

async def fetch_with_auth_proxy():
    # گزینه 1: اطلاعات کاربری در URL
    proxy = 'http://user123:pass456@proxy.example.com:8080'
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', proxy=proxy) as response:
            print(await response.json())

# گزینه 2: از طریق BasicAuth (برای برخی پروکسی‌ها)
async def fetch_with_basic_auth():
    proxy = 'http://proxy.example.com:8080'
    proxy_auth = aiohttp.BasicAuth('user123', 'pass456')
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', 
                                proxy=proxy, 
                                proxy_auth=proxy_auth) as response:
            print(await response.json())

asyncio.run(fetch_with_auth_proxy())

SOCKS5 در aiohttp:

برای SOCKS5 به کتابخانه aiohttp-socks نیاز است:

pip install aiohttp-socks
import asyncio
from aiohttp_socks import ProxyConnector
import aiohttp

async def fetch_with_socks5():
    connector = ProxyConnector.from_url('socks5://user:pass@123.45.67.89:1080')
    
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get('https://httpbin.org/ip') as response:
            print(await response.json())

asyncio.run(fetch_with_socks5())

هنگام کار با پروکسی‌های موبایل برای پارس کردن شبکه‌های اجتماعی یا بازارها، توصیه می‌شود که از aiohttp استفاده کنید — غیرهمزمانی اجازه می‌دهد که صدها درخواست به طور همزمان بدون مسدود کردن جریان اجرا پردازش شوند.

چرخش غیرهمزمان و استخر پروکسی

برای پارسرهای با بار بالا، چرخش مؤثر پروکسی با مدیریت خطاها و جایگزینی خودکار IPهای غیرعملیاتی حیاتی است. الگوهای پیشرفته‌ای برای aiohttp بررسی می‌کنیم.

کلاسی برای مدیریت استخر پروکسی:

import aiohttp
import asyncio
from itertools import cycle
from typing import List, Optional

class ProxyPool:
    def __init__(self, proxy_list: List[str]):
        self.proxy_list = proxy_list
        self.proxy_cycle = cycle(proxy_list)
        self.failed_proxies = set()
    
    def get_next_proxy(self) -> Optional[str]:
        """دریافت پروکسی کاری بعدی"""
        for _ in range(len(self.proxy_list)):
            proxy = next(self.proxy_cycle)
            if proxy not in self.failed_proxies:
                return proxy
        return None  # همه پروکسی‌ها در دسترس نیستند
    
    def mark_failed(self, proxy: str):
        """علامت‌گذاری پروکسی به عنوان غیرعملیاتی"""
        self.failed_proxies.add(proxy)
        print(f"پروکسی {proxy} به عنوان غیرقابل دسترس علامت‌گذاری شد")
    
    async def fetch(self, session: aiohttp.ClientSession, url: str, **kwargs):
        """انجام درخواست با تغییر خودکار پروکسی در صورت خطا"""
        max_retries = 3
        
        for attempt in range(max_retries):
            proxy = self.get_next_proxy()
            if not proxy:
                raise Exception("همه پروکسی‌ها در دسترس نیستند")
            
            try:
                async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=10), **kwargs) as response:
                    return await response.json()
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                print(f"خطا با پروکسی {proxy}: {e}")
                self.mark_failed(proxy)
                continue
        
        raise Exception(f"درخواست پس از {max_retries} تلاش انجام نشد")

# استفاده
async def main():
    proxy_list = [
        'http://user:pass@proxy1.example.com:8080',
        'http://user:pass@proxy2.example.com:8080',
        'http://user:pass@proxy3.example.com:8080',
    ]
    
    pool = ProxyPool(proxy_list)
    
    async with aiohttp.ClientSession() as session:
        # انجام 10 درخواست با چرخش خودکار
        tasks = [pool.fetch(session, 'https://httpbin.org/ip') for _ in range(10)]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                print(f"درخواست {i+1} با خطا مواجه شد: {result}")
            else:
                print(f"درخواست {i+1}: IP = {result.get('origin')}")

asyncio.run(main())

پردازش موازی با محدودیت همزمانی:

import aiohttp
import asyncio
from itertools import cycle

async def fetch_url(session, url, proxy, semaphore):
    async with semaphore:  # محدودیت درخواست‌های همزمان
        try:
            async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=10)) as response:
                data = await response.json()
                return {'url': url, 'ip': data.get('origin'), 'status': response.status}
        except Exception as e:
            return {'url': url, 'error': str(e)}

async def main():
    urls = [f'https://httpbin.org/ip' for _ in range(50)]  # 50 درخواست
    proxy_list = [
        'http://user:pass@proxy1.example.com:8080',
        'http://user:pass@proxy2.example.com:8080',
    ]
    proxy_cycle = cycle(proxy_list)
    
    # محدودیت: حداکثر 10 درخواست همزمان
    semaphore = asyncio.Semaphore(10)
    
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_url(session, url, next(proxy_cycle), semaphore)
            for url in urls
        ]
        results = await asyncio.gather(*tasks)
        
        # تجزیه و تحلیل نتایج
        successful = [r for r in results if 'ip' in r]
        failed = [r for r in results if 'error' in r]
        
        print(f"درخواست‌های موفق: {len(successful)}")
        print(f"درخواست‌های ناموفق: {len(failed)}")

asyncio.run(main())

استفاده از asyncio.Semaphore در هنگام کار با پروکسی حیاتی است — تعداد بسیار زیاد اتصالات همزمان از طریق یک IP ممکن است منجر به مسدود شدن از سوی سایت هدف یا ارائه‌دهنده پروکسی شود.

مدیریت خطاها و زمان‌های تایم‌اوت

کار با پروکسی با افزایش تعداد خطاها همراه است: زمان‌های تایم‌اوت، قطع اتصال، و عدم دسترسی سرورهای پروکسی. مدیریت صحیح خطاها کلید ثبات پارسر است.

خطاهای متداول در کار با پروکسی:

خطا علت راه‌حل
ProxyError سرور پروکسی در دسترس نیست به پروکسی دیگری سوئیچ کنید
ConnectTimeout پروکسی به موقع پاسخ نمی‌دهد زمان تایم‌اوت را افزایش دهید یا پروکسی را عوض کنید
ProxyAuthenticationRequired نام کاربری/رمز عبور نادرست اطلاعات کاربری را بررسی کنید
SSLError مشکلات با گواهی SSL بررسی SSL را غیرفعال کنید (توصیه نمی‌شود)
TooManyRedirects پروکسی یک حلقه ریدایرکت ایجاد می‌کند پروکسی را عوض کنید یا ریدایرکت‌ها را محدود کنید

مدیریت خطاها در requests:

import requests
from requests.exceptions import ProxyError, ConnectTimeout, RequestException

def fetch_with_retry(url, proxies, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.get(
                url, 
                proxies=proxies, 
                timeout=(5, 10),  # (زمان تایم‌اوت اتصال، زمان تایم‌اوت خواندن)
                allow_redirects=True,
                verify=True  # بررسی گواهی SSL
            )
            response.raise_for_status()  # در صورت 4xx/5xx استثنا ایجاد می‌کند
            return response.json()
            
        except ProxyError as e:
            print(f"تلاش {attempt + 1}: پروکسی در دسترس نیست - {e}")
        except ConnectTimeout as e:
            print(f"تلاش {attempt + 1}: زمان تایم‌اوت اتصال - {e}")
        except requests.exceptions.HTTPError as e:
            print(f"تلاش {attempt + 1}: خطای HTTP {e.response.status_code}")
            if e.response.status_code == 407:  # نیاز به احراز هویت پروکسی
                print("خطای احراز هویت پروکسی!")
                break  # در صورت خطای احراز هویت تکرار نکنید
        except RequestException as e:
            print(f"تلاش {attempt + 1}: خطای عمومی - {e}")
        
        if attempt < max_retries - 1:
            print(f"تکرار پس از 2 ثانیه...")
            import time
            time.sleep(2)
    
    raise Exception(f"درخواست پس از {max_retries} تلاش انجام نشد")

# استفاده
proxies = {'http': 'http://user:pass@proxy.example.com:8080', 'https': 'http://user:pass@proxy.example.com:8080'}
try:
    data = fetch_with_retry('https://httpbin.org/ip', proxies)
    print(data)
except Exception as e:
    print(f"خطای بحرانی: {e}")

مدیریت خطاها در aiohttp:

import aiohttp
import asyncio
from aiohttp import ClientError, ClientProxyConnectionError

async def fetch_with_retry(session, url, proxy, max_retries=3):
    for attempt in range(max_retries):
        try:
            timeout = aiohttp.ClientTimeout(total=10, connect=5)
            async with session.get(url, proxy=proxy, timeout=timeout) as response:
                response.raise_for_status()
                return await response.json()
                
        except ClientProxyConnectionError as e:
            print(f"تلاش {attempt + 1}: خطای اتصال به پروکسی - {e}")
        except asyncio.TimeoutError:
            print(f"تلاش {attempt + 1}: زمان تایم‌اوت")
        except aiohttp.ClientHttpProxyError as e:
            print(f"تلاش {attempt + 1}: خطای HTTP پروکسی - {e}")
            if e.status == 407:
                print("خطای احراز هویت پروکسی!")
                break
        except ClientError as e:
            print(f"تلاش {attempt + 1}: خطای عمومی کلاینت - {e}")
        
        if attempt < max_retries - 1:
            await asyncio.sleep(2)
    
    raise Exception(f"درخواست پس از {max_retries} تلاش انجام نشد")

async def main():
    proxy = 'http://user:pass@proxy.example.com:8080'
    async with aiohttp.ClientSession() as session:
        try:
            data = await fetch_with_retry(session, 'https://httpbin.org/ip', proxy)
            print(data)
        except Exception as e:
            print(f"خطای بحرانی: {e}")

asyncio.run(main())

تنظیم زمان‌های تایم‌اوت:

تنظیم صحیح زمان‌های تایم‌اوت برای ثبات حیاتی است. مقادیر پیشنهادی:

  • زمان تایم‌اوت اتصال: 5-10 ثانیه (زمان برای برقراری اتصال با پروکسی)
  • زمان تایم‌اوت خواندن: 10-30 ثانیه (زمان برای دریافت پاسخ از سایت هدف)
  • زمان تایم‌اوت کل: 30-60 ثانیه (زمان کل درخواست)

برای پروکسی‌های مسکونی کند، توصیه می‌شود زمان‌های تایم‌اوت را تا 20-30 ثانیه برای اتصال افزایش دهید، زیرا مسیریابی از طریق ارائه‌دهندگان واقعی ممکن است زمان بیشتری ببرد.

بهترین شیوه‌ها و بهینه‌سازی

کار مؤثر با پروکسی نیاز به رعایت مجموعه‌ای از قوانین برای حداقل‌سازی مسدودیت‌ها و حداکثر کردن عملکرد دارد.

1. استفاده از Session برای استفاده مجدد از اتصالات:

# requests: Session از اتصالات TCP استفاده مجدد می‌کند
session = requests.Session()
session.proxies = {'http': proxy, 'https': proxy}

for url in urls:
    response = session.get(url)  # سریع‌تر از requests.get()

# aiohttp: Session برای غیرهمزمانی الزامی است
async with aiohttp.ClientSession() as session:
    tasks = [session.get(url, proxy=proxy) for url in urls]
    await asyncio.gather(*tasks)

2. تنظیم User-Agent و هدرهای واقع‌گرایانه:

import requests

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
    'Accept-Language': 'en-US,en;q=0.5',
    'Accept-Encoding': 'gzip, deflate, br',
    'DNT': '1',
    'Connection': 'keep-alive',
    'Upgrade-Insecure-Requests': '1'
}

proxies = {'http': proxy, 'https': proxy}
response = requests.get('https://example.com', headers=headers, proxies=proxies)

3. محدود کردن نرخ (درخواست‌ها در ثانیه):

import time
import requests

class RateLimiter:
    def __init__(self, max_requests_per_second):
        self.max_requests = max_requests_per_second
        self.interval = 1.0 / max_requests_per_second
        self.last_request_time = 0
    
    def wait(self):
        elapsed = time.time() - self.last_request_time
        if elapsed < self.interval:
            time.sleep(self.interval - elapsed)
        self.last_request_time = time.time()

# استفاده: حداکثر 2 درخواست در ثانیه
limiter = RateLimiter(2)
proxies = {'http': proxy, 'https': proxy}

for url in urls:
    limiter.wait()
    response = requests.get(url, proxies=proxies)

4. ثبت‌نام و نظارت بر پروکسی:

import logging
from collections import defaultdict

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProxyMonitor:
    def __init__(self):
        self.stats = defaultdict(lambda: {'success': 0, 'failed': 0, 'total_time': 0})
    
    def log_request(self, proxy, success, response_time):
        stats = self.stats[proxy]
        if success:
            stats['success'] += 1
        else:
            stats['failed'] += 1
        stats['total_time'] += response_time
        
        # ثبت‌نام هر 10 درخواست
        total = stats['success'] + stats['failed']
        if total % 10 == 0:
            avg_time = stats['total_time'] / total
            success_rate = stats['success'] / total * 100
            logger.info(f"پروکسی {proxy}: {total} درخواست، موفقیت {success_rate:.1f}%, میانگین {avg_time:.2f}s")

monitor = ProxyMonitor()

# در کد درخواست
import time
start = time.time()
try:
    response = requests.get(url, proxies=proxies, timeout=10)
    monitor.log_request(proxy, True, time.time() - start)
except Exception as e:
    monitor.log_request(proxy, False, time.time() - start)
    logger.error(f"خطا با پروکسی {proxy}: {e}")

5. کش DNS برای تسریع:

# aiohttp با کش DNS
import aiohttp
from aiohttp.resolver import AsyncResolver

resolver = AsyncResolver(nameservers=['8.8.8.8', '8.8.4.4'])
connector = aiohttp.TCPConnector(resolver=resolver, ttl_dns_cache=300)

async with aiohttp.ClientSession(connector=connector) as session:
    # درخواست‌ها از کش DNS به مدت 5 دقیقه استفاده خواهند کرد
    async with session.get(url, proxy=proxy) as response:
        data = await response.json()

6. مدیریت کپچا و مسدودیت‌ها:

نکته: هنگام دریافت وضعیت 403، 429 یا کپچا، توصیه می‌شود:

  • پروکسی را به IP از زیرشبکه دیگر تغییر دهید
  • زمان تأخیر بین درخواست‌ها را افزایش دهید (تا 5-10 ثانیه)
  • User-Agent و سایر هدرها را تغییر دهید
  • از کوکی‌های جلسات موفق قبلی استفاده کنید

مقایسه درخواست‌ها و aiohttp برای پروکسی

انتخاب بین requests و aiohttp به وظیفه و حجم داده‌ها بستگی دارد. به بررسی تفاوت‌های کلیدی می‌پردازیم.

معیار requests aiohttp
همزمانی همزمان (مسدودکننده) غیرهمزمان (غیرمسدودکننده)
عملکرد ~10-50 درخواست/ثانیه ~100-1000 درخواست/ثانیه
سادگی کد برای مبتدیان ساده‌تر است نیاز به دانش async/await دارد
تنظیم پروکسی دیکشنری proxies پارامتر proxy
پشتیبانی SOCKS5 از طریق requests[socks] از طریق aiohttp-socks
استفاده از حافظه کمتر (یک رشته) بیشتر (چندین کار)
بهتر برای اسکریپت‌های ساده، <100 درخواست پارسرها، >1000 درخواست

کی وقتی از requests استفاده کنیم:

  • اسکریپت‌های ساده برای وظایف یک‌باره
  • پروتوتایپ و آزمایش
  • حجم کم درخواست‌ها (تا 100 در دقیقه)
  • زمانی که سادگی کد و خوانایی مهم است
  • ادغام با کتابخانه‌های همزمان

کی وقتی از aiohttp استفاده کنیم:

  • پارس کردن حجم‌های بزرگ داده (هزاران صفحه)
  • نظارت بر منابع متعدد در زمان واقعی
  • خدمات API با بار بالا
  • زمانی که سرعت پردازش حیاتی است
  • کار با WebSocket از طریق پروکسی

مقایسه عملی عملکرد:

# تست: 100 درخواست از طریق پروکسی

# requests (همزمان) - ~50 ثانیه
import requests
import time

start = time.time()
proxies = {'http': proxy, 'https': proxy}
for i in range(100):
    response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(f"requests: {time.time() - start:.2f} ثانیه")

# aiohttp (غیرهمزمان) - ~5 ثانیه
import aiohttp
import asyncio

async def fetch_all():
    async with aiohttp.ClientSession() as session:
        tasks = [
            session.get('https://httpbin.org/ip', proxy=proxy)
            for _ in range(100)
        ]
        await asyncio.gather(*tasks)

start = time.time()
asyncio.run(fetch_all())
print(f"aiohttp: {time.time() - start:.2f} ثانیه")

هنگام استفاده از پروکسی‌های دیتا سنتر برای پارس کردن با سرعت بالا، aiohttp برتری 10-20 برابری نسبت به requests نشان می‌دهد به دلیل پردازش موازی درخواست‌ها.

نتیجه‌گیری

تنظیم پروکسی در Python از طریق کتابخانه‌های requests و aiohttp یک مهارت بنیادی برای توسعه پارسرها، خودکارسازی جمع‌آوری داده‌ها و دور زدن محدودیت‌های جغرافیایی است. کتابخانه requests برای اسکریپت‌های ساده و پروتوتایپ مناسب است به دلیل API همزمان واضح، در حالی که aiohttp عملکرد بالایی را در پردازش هزاران درخواست از طریق معماری غیرهمزمان فراهم می‌کند.

نکات کلیدی برای کار مؤثر با پروکسی در Python: مدیریت صحیح خطاها و زمان‌های تایم‌اوت، پیاده‌سازی چرخش آدرس‌های IP برای توزیع بار، استفاده از Session برای استفاده مجدد از اتصالات، تنظیم هدرها و User-Agent‌های واقع‌گرایانه، نظارت بر عملکرد سرورهای پروکسی. برای پروکسی‌های SOCKS5 به کتابخانه‌های اضافی نیاز است — requests[socks] یا aiohttp-socks.

هنگام انتخاب نوع پروکسی برای پارس کردن، به ویژگی‌های وظیفه توجه کنید: برای پارسرهای با بار بالا که هزاران درخواست دارند، پروکسی‌های سریع دیتا سنتر مناسب هستند، برای دور زدن سیستم‌های ضد ربات سخت و کار با شبکه‌های اجتماعی، پروکسی‌های مسکونی با IP‌های واقعی کاربران خانگی توصیه می‌شوند، و برای وظایفی که نیاز به حداکثر ناشناسی و شبیه‌سازی ترافیک موبایل دارند، پروکسی‌های موبایل با IP‌های اپراتورهای تلفن همراه بهینه هستند.

اگر قصد دارید پارسرهای با عملکرد بالا توسعه دهید یا جمع‌آوری داده‌ها را از منابع متعدد خودکار کنید، توصیه می‌کنیم از پروکسی‌های مسکونی استفاده کنید — آن‌ها سطح بالایی از ناشناسی، حداقل خطر مسدودیت و عملکرد پایدار با اکثر خدمات وب محافظت شده را فراهم می‌کنند. برای وظایف فنی با سرعت پردازش بالا، همچنین پروکسی‌های دیتا سنتر با تأخیر کم و ظرفیت بالا مناسب هستند.

```