إذا كنت تقوم بالاستخلاص (scraping) أو الزحف (crawling) أو أتمتة أي شيء على نطاق واسع في Python، فعاجلاً أم آجلاً سيلاحظ الموقع المستهدف أن كل طلب يأتي من عنوان IP واحد ويقوم بإيقافك. تدوير البروكسيات في Python يوزّع حركة مرورك عبر العديد من عناوين IP بحيث لا يتسبب أي عنوان بمفرده في تجاوز حد المعدل أو الحصول على حظر. يوضّح هذا الدليل الكود الدقيق لـ requests وhttpx وaiohttp، بالإضافة إلى كيفية الاختيار بين بوابة تدوير وبين تجمّع IP مُدار.
هناك قراران شاملان يشكّلان كل ما يلي. الأول: الجلسات الدوارة مقابل الثابتة — عنوان IP جديد لكل طلب، أو عنوان IP واحد محتفظ به لتدفّق متعدد الخطوات. الثاني: عناوين IP السكنية مقابل مراكز البيانات — راجع مصفوفة قرار نوع البروكسي. اضبط هذين الأمرين بشكل صحيح ويصبح الكود هو الجزء السهل.
1. تدوير البوابة (موصى به). تتصل بنقطة نهاية واحدة ويمنحك المزوّد عنوان IP خروج جديداً عند كل اتصال جديد. لا قائمة لإدارتها، ولا عناوين IP ميتة لتنقيتها. يعمل تجمّع السكني الديناميكي من JIBAO بهذه الطريقة: نفس المضيف والمنفذ، وعنوان IP جديد لكل طلب.
2. قائمة IP ذاتية الإدارة. تحتفظ بقائمة من إدخالات host:port وتختار واحداً لكل طلب. تحكّم أكبر، لكنك تتحمّل فحص السلامة وإعادة المحاولات وتتبّع الحظر. مفيد مع عناوين IP مراكز البيانات المخصصة عندما تحتاج إلى عناوين معروفة ومستقرة.
أبسط حالة — توجيه طلب واحد عبر بوابة تقوم بالتدوير نيابةً عنك. استخدم socks5h:// (الحرف h في النهاية يعني أن DNS يُحَلّ على جانب البروكسي، ما يتجنّب تسريب عمليات البحث):
import requests
# JIBAO rotating residential gateway: a new exit IP per connection
PROXY = "socks5h://USERNAME:[email protected]:10001"
proxies = {"http": PROXY, "https": PROXY}
for _ in range(5):
r = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=30)
print(r.json()["origin"]) # a different IP each loopهل تحتاج إلى أن يتحدث requests بروتوكول SOCKS5؟ ثبّت الإضافة: pip install "requests[socks]". إذا كنت تفضّل نقطة نهاية بروكسي HTTP، فبدّل المخطط إلى http:// — والباقي مطابق تماماً.
import random
import requests
PROXY_POOL = [
"socks5h://USERNAME-country-us:[email protected]:10001",
"socks5h://USERNAME-country-uk:[email protected]:10001",
"socks5h://USERNAME-country-de:[email protected]:10001",
]
def fetch(url):
proxy = random.choice(PROXY_POOL)
return requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)
resp = fetch("https://example.com")
print(resp.status_code)تتعطّل تدفّقات تسجيل الدخول وعربات التسوق والنتائج المقسّمة على صفحات إذا تغيّر عنوان IP في منتصف المهمة. ثبّت عنوان IP بإضافة رمز جلسة إلى اسم المستخدم؛ أعد استخدام نفس الرمز للإبقاء على نفس عنوان IP الخروج طوال مدّته:
import requests
SESSION_ID = "task-0001"
PROXY = f"socks5h://USERNAME-session-{SESSION_ID}:[email protected]:10001"
s = requests.Session()
s.proxies = {"http": PROXY, "https": PROXY}
s.post("https://example.com/login", data={"u": "me", "p": "secret"})
s.get("https://example.com/dashboard") # same IP, session intactيمنحك httpx عميلاً حديثاً مع HTTP/2 وغير متزامن (async) أصلي. لاحظ أن الإصدارات الأحدث تأخذ وسيطة proxy= واحدة:
import httpx
PROXY = "socks5h://USERNAME:[email protected]:10001"
with httpx.Client(proxy=PROXY, timeout=30) as client:
for _ in range(5):
print(client.get("https://httpbin.org/ip").json()["origin"])يحتاج دعم SOCKS في httpx إلى pip install "httpx[socks]". بالنسبة إلى HTTP/2، أضف http2=True إلى العميل — فهذا يجعل حركة مرورك تبدو أكثر شبهاً بمتصفّح حقيقي، ما يتناغم جيداً مع مطابقة بصمة TLS.
للحصول على إنتاجية عالية، أطلق العديد من الطلبات دفعة واحدة. يحصل كل اتصال عبر البوابة على عنوان IP خروج خاص به، لذا يجتمع التزامن والتدوير معاً:
import asyncio
import aiohttp
from aiohttp_socks import ProxyConnector
PROXY = "socks5://USERNAME:[email protected]:10001"
async def fetch(url):
connector = ProxyConnector.from_url(PROXY)
async with aiohttp.ClientSession(connector=connector) as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as r:
return await r.text()
async def main():
urls = ["https://httpbin.org/ip"] * 20
results = await asyncio.gather(*(fetch(u) for u in urls))
print(f"fetched {len(results)} pages")
asyncio.run(main())ثبّت موصّل SOCKS بـ pip install aiohttp_socks. ابنِ موصّلاً جديداً لكل مهمة بحيث يفتح كل طلب اتصالاً جديداً عبر البروكسي.
لا يؤتي التدوير ثماره إلا إذا أعدت محاولة الإخفاقات على عنوان IP جديد. تراجع بشكل أسّي ودوّر رمز الجلسة في كل محاولة:
import requests
from time import sleep
def fetch_with_retry(url, max_retries=4):
for attempt in range(max_retries):
proxy = f"socks5h://USERNAME-session-r{attempt}:[email protected]:10001"
try:
r = requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)
if r.status_code in (403, 429):
raise requests.HTTPError(f"blocked: {r.status_code}")
r.raise_for_status()
return r.text
except requests.RequestException:
sleep(2 ** attempt) # 1s, 2s, 4s, 8s
raise RuntimeError(f"failed after {max_retries} tries: {url}")قبل عملية تشغيل كبيرة، تأكّد من أن عنوان IP يتغيّر فعلاً. اطرق نقطة نهاية صدى (echo) بضع مرات وتحقّق من اختلاف عناوين IP. وللتأكد أيضاً من أن بصمة TLS الخاصة بك لا تفضحك، شغّل الطلب مقابل مدقّق بصمة JA3/TLS المجاني من JIBAO — فعنوان IP دوّار ببصمة Python فاضحة لا يزال يُحظَر.
بمجرد أن يصبح التدوير متيناً، يكون الجدار التالي الذي تصطدم به عادةً هو البصمات، لا عناوين IP. إذا كانت عناوين IP السكنية النظيفة لا تزال تحصل على 403، فاقرأ كيفية تجاوز بصمات TLS باستخدام curl_cffi ووصفة تجاوز Cloudflare الكاملة.
احصل على 500MB من حركة المرور المجانية وبوابة سكنية دوّارة تمنحك عنوان IP جديداً عند كل طلب.
ابدأ التجربة المجانيةيحصل المستخدمون الجدد على 500MB عند التسجيل، بالإضافة إلى مكافأة على أول عملية شحن. العرض لفترة محدودة.