Python requests और aiohttp में प्रॉक्सी सेटअप: कोड उदाहरणों के साथ पूर्ण गाइड

पार्सर्स विकसित करते समय, डेटा संग्रह की स्वचालन या वेब सेवाओं का परीक्षण करते समय Python से प्रॉक्सी सर्वर का उपयोग करना अक्सर आवश्यक होता है। requests और aiohttp प्रॉक्सी के साथ काम करने के लिए लचीले तंत्र प्रदान करते हैं, लेकिन उनकी सेटिंग में महत्वपूर्ण बारीकियाँ होती हैं। इस गाइड में हम सिंक्रोनस और असिंक्रोनस दृष्टिकोणों पर चर्चा करेंगे, HTTP और SOCKS5 प्रॉक्सी के लिए उदाहरण दिखाएंगे, IP रोटेशन और त्रुटि हैंडलिंग पर विचार करेंगे।

requests में प्रॉक्सी की बुनियादी सेटिंग

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 पैरामीटर को स्पष्ट रूप से पास करना अनुशंसित है।

requests में प्रमाणीकरण और 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 का समर्थन करता है (कुछ अनुप्रयोगों के लिए आवश्यक)।

requests में प्रॉक्सी का रोटेशन

बड़े डेटा सेट के पार्सिंग के दौरान एक ही 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())

aiohttp में SOCKS5:

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 के माध्यम से बहुत अधिक समानांतर कनेक्शन लक्ष्य साइट या प्रॉक्सी प्रदाता द्वारा ब्लॉक कर सकते हैं।

त्रुटियों और टाइमआउट की हैंडलिंग

प्रॉक्सी के साथ काम करना उच्च मात्रा में त्रुटियों के साथ जुड़ा होता है: टाइमआउट, कनेक्शन टूटना, प्रॉक्सी सर्वरों का अस्वीकृत होना। त्रुटियों की सही हैंडलिंग पार्सर की स्थिरता के लिए कुंजी है।

प्रॉक्सी के साथ काम करते समय सामान्य त्रुटियाँ:

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. कनेक्शनों को पुन: उपयोग करने के लिए सत्र का उपयोग:

# requests: सत्र TCP कनेक्शनों को पुन: उपयोग करता है
session = requests.Session()
session.proxies = {'http': proxy, 'https': proxy}

for url in urls:
    response = session.get(url)  # requests.get() से तेज़

# aiohttp: असिंक्रोनसिटी के लिए सत्र आवश्यक है
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:
    # अनुरोध 5 मिनट के लिए DNS कैश का उपयोग करेंगे
    async with session.get(url, proxy=proxy) as response:
        data = await response.json()

6. कैप्चा और ब्लॉकों की हैंडलिंग:

प्रॉक्सी के लिए requests और aiohttp की तुलना

requests और aiohttp के बीच चयन कार्य और डेटा की मात्रा पर निर्भर करता है। मुख्य भिन्नताओं पर विचार करें।

कब 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 requests की तुलना में 10-20 गुना बेहतर प्रदर्शन दिखाता है, क्योंकि यह अनुरोधों को समानांतर में संसाधित करता है।

निष्कर्ष

Python में requests और aiohttp के माध्यम से प्रॉक्सी सेट करना पार्सर्स विकसित करने, डेटा संग्रह को स्वचालित करने और भौगोलिक प्रतिबंधों को बायपास करने के लिए एक मौलिक कौशल है। requests पुस्तकालय सरल स्क्रिप्ट और प्रोटोटाइपिंग के लिए उपयुक्त है, क्योंकि इसका सिंक्रोनस API स्पष्ट है, जबकि aiohttp हजारों अनुरोधों को असिंक्रोनस आर्किटेक्चर के माध्यम से संसाधित करते समय उच्च प्रदर्शन प्रदान करता है।

Python में प्रॉक्सी के साथ प्रभावी काम करने के लिए मुख्य बिंदु: त्रुटियों और टाइमआउट की सही हैंडलिंग, लोड को वितरित करने के लिए IP पते का रोटेशन लागू करना, कनेक्शनों को पुन: उपयोग करने के लिए सत्र का उपयोग करना, यथार्थवादी हेडर और User-Agent सेट करना, प्रॉक्सी सर्वरों के प्रदर्शन की निगरानी करना। SOCKS5 प्रॉक्सी के लिए अतिरिक्त पुस्तकालयों की आवश्यकता होती है - requests[socks] या aiohttp-socks।

पार्सिंग के लिए प्रॉक्सी के प्रकार का चयन करते समय कार्य की विशिष्टता पर विचार करें: उच्च लोड वाले पार्सर्स के लिए तेज डेटा सेंटर प्रॉक्सी उपयुक्त हैं, कठोर एंटी-बॉट सिस्टम को बायपास करने और सामाजिक नेटवर्क के साथ काम करने के लिए रेसिडेंशियल प्रॉक्सी की सिफारिश की जाती है, और अधिकतम गुमनामी और मोबाइल ट्रैफ़िक की नकल के लिए मोबाइल प्रॉक्सी सबसे अच्छे होते हैं।

यदि आप उच्च प्रदर्शन वाले पार्सर्स विकसित करने या कई स्रोतों से डेटा संग्रह को स्वचालित करने की योजना बना रहे हैं, तो रेसिडेंशियल प्रॉक्सी का प्रयास करने की सिफारिश की जाती है - वे उच्च स्तर की गुमनामी, ब्लॉकों का न्यूनतम जोखिम और अधिकांश सुरक्षित वेब सेवाओं के साथ स्थिर कार्य प्रदान करते हैं। तकनीकी कार्यों के लिए उच्च गति के साथ, डेटा सेंटर प्रॉक्सी भी उपयुक्त हैं, जिनकी देरी कम और बैंडविड्थ उच्च होती है।