اسکریپینگ با aiohttp و چرخش پروکسی‌ها

مقدمه

در این مقاله یاد می‌گیریم چگونه در اسکریپینگ با Python aiohttp از پروکسی‌ها استفاده و آن‌ها را بچرخانیم تا احتمال بلاک‌شدن کاهش یابد. در پایان شما:

• نحوه‌ی ارسال درخواست با پروکسی ساده را می‌دانید،
• روش‌های احراز هویت پروکسی را می‌شناسید،
• سه فرمت معمول پروکسی (لیست خام، دروازه/گیت‌وی، API-proxy) و نحوه‌ی پیاده‌سازی هر کدام را می‌توانید پیاده کنید،
• الگوهای عملی برای مدیریت خطا، تایم‌اوت و حذف پروکسی‌های خراب را یاد می‌گیرید.

استفاده‌ی ساده از پروکسی در aiohttp

برای هر درخواست کافیست پارامتر proxy را در session.get/post/... قرار دهید. این روش برای تمام متدهای HTTP که aiohttp پشتیبانی می‌کند (GET, POST, PUT, DELETE, PATCH, HEAD) کاربردی است.

import aiohttp
import asyncio

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get('http://httpbin.org/get', proxy='http://proxy.example.com:8080') as resp:
            print(resp.status)
            print(await resp.text())

asyncio.run(main())

توضیح کد:

• ورودی: URL هدف و آدرس پروکسی به صورت رشته.
• خروجی: وضعیت HTTP و متن پاسخ که از target برمی‌گردد.
• عملکرد: ایجاد یک ClientSession و ارسال درخواست با پارامتر proxy.

نکته: برای محافظت بیشتر همیشه از ClientTimeout استفاده کنید تا درخواست‌های آویزان را قطع کنید و منابع را آزاد نگه دارید.

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

برخی پروکسی‌ها نیاز به نام‌کاربری و رمزعبور دارند. دو روش معمول وجود دارد:

اضافه کردن نام‌کاربری و رمز به رشته‌ی پروکسی

ساده‌ترین راه افزودن 'username:password@' داخل URL پروکسی است:

# اضافه کردن cred در رشته‌ی پروکسی
proxy = 'http://USERNAME:PASSWORD@proxy.example.com:8080'

async with aiohttp.ClientSession() as session:
    async with session.get('http://httpbin.org/get', proxy=proxy) as resp:
        print(resp.status)
        print(await resp.text())

مزایا: آسان و سازگار با بسیاری از سرویس‌ها. معایب: ممکن است لاگ‌ها یا تاریخچه‌ی پردازش، اعتبارنامه‌ها را فاش کند؛ بهتر است از متغیرهای محیطی استفاده کنید و آن‌ها را در کد هاردکُد نکنید.

استفاده از proxy_auth با aiohttp.BasicAuth

روش امن‌تر استفاده از پارامتر proxy_auth با آبجکت aiohttp.BasicAuth است:

import aiohttp
import asyncio

async def main():
    proxy = 'http://proxy.example.com:8080'
    auth = aiohttp.BasicAuth('USERNAME', 'PASSWORD')

    async with aiohttp.ClientSession() as session:
        async with session.get('http://httpbin.org/get', proxy=proxy, proxy_auth=auth) as resp:
            print(resp.status)
            print(await resp.text())

asyncio.run(main())

توضیح: proxy_auth یک روش واضح‌تر برای نگهداری اعتبارنامه‌ها است و به مدیریت امن‌تر کمک می‌کند.

سه فرمت معمول پروکسی

سه الگوی رایج که امروز سرویس دهندگان پروکسی عرضه می‌کنند:

• لیست خام IP (یک فایل یا آرایه از آدرس‌ها که باید خودتان بچرخانید)
• Proxy Gateway (یک آدرس ثابت که خود سرویس‌دهنده داخلش گیت‌وی و چرخش انجام می‌دهد)
• Proxy API (ارسال URL به API و دریافت HTML یا نتیجه‌ی آماده)

امروزه بسیاری از ارائه‌دهندگان به جای فروش لیست خام، دسترسی از طریق گیت‌وی یا API می‌دهند تا پیچیدگی‌های مدیریت پروکسی را کاهش دهند.

ادغام #1: چرخش از طریق لیست پروکسی

ایده: یک لیست از پروکسی‌ها دارید و برای هر درخواست یک پروکسی متفاوت انتخاب می‌کنید. این مثال شامل مکانیزم ساده حذف پروکسی‌های مشکل‌دار هم هست.

import aiohttp
import asyncio
import random
from asyncio import Semaphore

proxy_list = [
    'http://User:Pass@85.237.57.198:20000',
    'http://User:Pass@85.237.57.198:21000',
    'http://User:Pass@85.237.57.198:22000',
    'http://User:Pass@85.237.57.198:23000',
]

# ساده‌ترین روش: انتخاب تصادفی
async def fetch(session, url, proxy):
    try:
        timeout = aiohttp.ClientTimeout(total=15)
        async with session.get(url, proxy=proxy, timeout=timeout) as resp:
            return resp.status, await resp.text()
    except Exception as e:
        return 'error', str(e)

async def main(urls):
    sem = Semaphore(10)  # کنترل concurrency
    failures = {p: 0 for p in proxy_list}
    alive = list(proxy_list)

    async with aiohttp.ClientSession() as session:
        async def worker(u):
            async with sem:
                if not alive:
                    raise RuntimeError('No proxies left')
                proxy = random.choice(alive)
                status, content = await fetch(session, u, proxy)
                if status == 'error' or (isinstance(status, int) and status >= 500):
                    failures[proxy] += 1
                    if failures[proxy] >= 3:
                        alive.remove(proxy)
                return proxy, status

        tasks = [asyncio.create_task(worker(u)) for u in urls]
        results = await asyncio.gather(*tasks)
        print(results)

# اجرا نمونه
# asyncio.run(main(['http://httpbin.org/get'] * 20))

توضیحات عملی:

• ورودی: لیست URLها و آرایه پروکسی‌ها.
• خروجی: وضعیت هر درخواست و پروکسی‌ای که استفاده شده.
• نکته: در مقیاس بزرگ بهتر است یک سرویس مانیتورینگ برای هر پروکسی بسازید که latency، خطاها و نرخ موفقیت را ثبت کند و پروکسی‌های بد را موقتاً حذف کند.

مزایا و چالش‌های لیست خام

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

ادغام #2: استفاده از Proxy Gateway

در این مدل به جای استفاده از لیست، یک آدرس پروکسی ثابت (gateway) دارید و سرویس‌دهنده خودش چرخش، انتخاب و پاکسازی IP را انجام می‌دهد. پیاده‌سازی بسیار ساده است:

import aiohttp
import asyncio

async def main():
    gateway = 'http://USERNAME:PASSWORD@zproxy.provider-gateway.io:22225'
    async with aiohttp.ClientSession() as session:
        async with session.get('http://httpbin.org/get', proxy=gateway) as resp:
            print(resp.status)
            print(await resp.text())

# asyncio.run(main())

چرا مفید است:

• پیچیدگی چرخش و مدیریت پروکسی از شما برداشته می‌شود.
• بیشتر مناسب residential و mobile proxies است.

معایب: هزینه‌ی بالاتر و کمتر بودن کنترل مستقیم روی IPها.

ادغام #3: استفاده از Proxy API Endpoint

در این حالت شما URL هدف را به یک API ارسال می‌کنید و API پاسخ HTML را بازمی‌گرداند. این روش برای زمانی که می‌خواهید کمترین پیچیدگی را در کلاینت داشته باشید عالی است.

import aiohttp
import asyncio
from urllib.parse import urlencode

async def main():
    api_endpoint = 'https://proxy-api.example.com/fetch'
    target_url = 'https://httpbin.org/ip'
    params = {'api_key': 'APIKEY', 'url': target_url}

    async with aiohttp.ClientSession() as session:
        # aiohttp خودش params را به صورت امن urlencode می‌کند
        async with session.get(api_endpoint, params=params) as resp:
            print(resp.status)
            print(await resp.text())

# asyncio.run(main())

نکات اجرایی:

• در بسیاری از APIها لازم است URL هدف را به صورت urlencode ارسال کنید؛ وقتی از پارامتر params استفاده می‌کنید aiohttp این کار را برایتان انجام می‌دهد، ولی اگر URL را داخل یک پارامتر رشته‌ای می‌سازید می‌توانید از urllib.parse.urlencode استفاده کنید.
• Proxy API سطح بالاتری از مدیریت را ارائه می‌دهد (چرخش هدر، مدیریت کوکی، ری‌تری) اما معمولاً هزینه‌ی بیشتری دارد.

توصیه‌های عملی، performance و امنیت

• از تایم‌اوت مناسب استفاده کنید (ClientTimeout) تا منابع بلاتکلیف آزاد شوند.
• برای همزمانی بالا از Semaphore برای محدود کردن همزمانی استفاده کنید تا فشار روی پروکسی‌ها و هدف معقول بماند.
• اعتبارنامه‌ها (API keys، user/pass) را در متغیرهای محیطی یا مدیر راز نگهداری کنید؛ هرگز در کد اصلی هاردکُد نکنید.
• لاگ‌ها را طوری تنظیم کنید که اطلاعات حساس (مثل رمزها) ذخیره نشود.
• الگوریتم چرخش: انتخاب تصادفی ساده است، اما برای عملکرد بهتر می‌توانید امتیازدهی بر اساس نرخ موفقیت و latency پیاده کنید و از round-robin یا weighted-random استفاده کنید.
• برای پایداری، پیاده‌سازی ری‌تری هوشمند (با backoff) و حذف پروکسی‌های با شکست مکرر لازم است.
• قوانین و اخلاق: قبل از اسکریپ کردن هر سایت قوانین استفاده و robots.txt را بررسی کنید و مسئولیت قانونی و اخلاقی را رعایت کنید.

جمع‌بندی

استفاده از پروکسی در اسکریپینگ با aiohttp می‌تواند از بلاک‌شدن جلوگیری کند و به شما امکان اجرای موازی گسترده را بدهد. برای پروژه‌های کوچک لیست پروکسی یا gateway کافی است، اما برای مقیاس بالا Proxy API یا gateway معمولاً هزینه‌ی اجرایی و نگهداری را کاهش می‌دهد. همیشه احراز هویت، مدیریت خطا، تایم‌اوت و روش حذف پروکسی‌های ناسالم را پیاده کنید تا یک اسکرِیپر پایدار و قابل اعتماد داشته باشید.