إعداد بروكسي Python requests، توصيل بروكسي SOCKS5 وHTTP

ما الذي يجب إعداده؟

تحتاج إلى Python 3.8 أو أحدث ومكتبة requests. لاستخدام SOCKS5، تحتاج إلى تثبيت حزمة دعم SOCKS. تثبيت requests[socks] يثبت PySocks تلقائيًا، مما يتيح استخدام عناوين socks5 و socks5h.

يمكنك الاطلاع على معلومات البروكسي في شاشة "بروكسياتي" في لوحة التحكم. في هذا المثال، نستخدم IP 198.51.100.23، منفذ HTTP 8000، منفذ SOCKS5 1080، اسم المستخدم spx_user، كلمة المرور spx_pass. استبدل هذه القيم بالقيم الفعلية.

pip install "requests[socks]"

كيفية الاتصال بـ HTTP بروكسي؟

يتم إرسال الطلبات عبر الوكيل عند إضافة العنوان إلى قاموس proxies في requests. من المهم إضافة نفس عنوان وكيل HTTP لكل من http و https. بدون مفتاح https، لن تمر طلبات HTTPS عبر الوكيل.

يتم تمرير طلبات مواقع HTTPS عبر البروكسي باستخدام طريقة CONNECT، حيث يتم نقل المحتوى مشفرًا من البداية إلى النهاية. يقوم الكود التالي بطلب عنوان IP للتحقق من ظهور IP البروكسي في الرد.

import requests

proxy = "http://spx_user:spx_pass@198.51.100.23:8000"
proxies = {"http": proxy, "https": proxy}

r = requests.get("https://api.ipify.org?format=json", proxies=proxies, timeout=20)
print(r.status_code, r.json())  # the "ip" field should show the proxy IP

كيف يتم الاتصال ببروكسي SOCKS5؟

يوصى باستخدام socks5h كبادئة للعنوان. socks5h يحل أسماء النطاقات من جهة البروكسي، مما يخفي طلبات DNS لخادمك ويحل الأسماء بناءً على بلد البروكسي. إذا استخدمت socks5، يتم حل الاسم على خادمك أولاً ثم يُرسل IP فقط عبر البروكسي.

يجب استخدام منفذ SOCKS5. إذا وضعت عنوان SOCKS5 في منفذ HTTP، فلن يتم الاتصال، لذا يرجى التمييز بين المنفذين في كل سطر.

import requests

proxy = "socks5h://spx_user:spx_pass@198.51.100.23:1080"
proxies = {"http": proxy, "https": proxy}

r = requests.get("https://api.ipify.org?format=json", proxies=proxies, timeout=20)
print(r.json())

كيفية تدوير عدة بروكسي لإرسال الطلبات؟

عند جمع العديد من الصفحات، تكون الطريقة الأساسية هي استلام القائمة واستخدام البروكسي التالي لكل طلب. يوضح الكود أدناه كيفية استلام قائمة SOCKS5 عبر API واستخدامها بالتناوب، وإعادة إرسال الطلبات الفاشلة إلى بروكسي آخر حتى 3 مرات. كما تمت إضافة فترة انتظار قصيرة لتجنب إرسال الطلبات بسرعة كبيرة إلى نفس النطاق.

من الأفضل إعطاء البروكسي الفاشل راحة قصيرة بدلاً من التخلص منه فورًا. غالبًا ما تكون الأخطاء المؤقتة بسبب حالة الموقع المستهدف.

import itertools
import time
import requests

API = "https://socks5proxy.io/api/v1"
KEY = "YOUR_API_KEY"

lines = requests.get(
    API + "/proxies/download",
    params={"format": "user:pass@ip:port", "protocol": "socks5"},
    headers={"Authorization": "Bearer " + KEY},
    timeout=30,
).text.split()
pool = itertools.cycle(lines)

def fetch(url, tries=3):
    for _ in range(tries):
        p = "socks5h://" + next(pool)
        try:
            r = requests.get(url, proxies={"http": p, "https": p}, timeout=20)
            if r.status_code == 200:
                return r.text
        except requests.RequestException:
            pass
        time.sleep(1)
    return None

for page in range(1, 6):
    html = fetch("https://example.com/list?page=" + str(page))
    print(page, "ok" if html else "failed")
    time.sleep(0.5)

كيفية الحفاظ على الجلسة وملفات تعريف الارتباط؟

لعمليات مثل تسجيل الدخول أو التنقل بين الصفحات حيث يجب استمرار الكوكيز، ثبت وكيل واحد لكل كائن Session في requests. ضع عنوان الوكيل في خاصية proxies للكائن Session وسيتم إرسال جميع الطلبات من هذه الجلسة عبر نفس الIP. عند انتهاء العملية، أغلق الجلسة وأنشئ جلسة جديدة بالوكيل التالي.

إذا قمت بتعيين رأس User-Agent وAccept-Language لكل جلسة، فإن المظهر الذي يراه موقع الهدف يكون ثابتًا.

import requests

s = requests.Session()
s.proxies = {
    "http": "http://spx_user:spx_pass@198.51.100.23:8000",
    "https": "http://spx_user:spx_pass@198.51.100.23:8000",
}
s.headers.update({"User-Agent": "Mozilla/5.0", "Accept-Language": "en-US,en;q=0.9"})

s.get("https://example.com/", timeout=20)          # cookies are stored in the session
r = s.get("https://example.com/account", timeout=20)  # same IP, same cookies
print(r.status_code)
s.close()

كيف يتم استخدام تفويض IP للاتصال؟

قم بتسجيل IPv4 للخادم الذي ينفذ هذا الكود في لوحة التحكم، ثم يمكنك استخدام العنوان بدون اسم مستخدم وكلمة مرور. هذا يسهل نشر نفس الكود على عدة خوادم دون الحاجة إلى تخزين كلمات المرور في الأكواد أو ملفات الإعدادات. يمكنك أيضًا اختيار تنسيق ip:port عند استلام القائمة.

import requests

proxy = "http://198.51.100.23:8000"  # no user or password, server IP is authorized in the dashboard
r = requests.get("https://api.ipify.org", proxies={"http": proxy, "https": proxy}, timeout=20)
print(r.text)

كيفية إرسال عدة طلبات في نفس الوقت بسرعة؟

تنتظر requests انتهاء كل طلب قبل المتابعة، لذا لجمع صفحات كثيرة بسرعة، تحتاج إلى تشغيل مهام متعددة في وقت واحد. أسهل طريقة هي استخدام ThreadPoolExecutor من concurrent.futures. حافظ على عدد المهام أقل من عدد الوكيلات لتجنب التحميل الزائد على IP واحد.

يقسم الكود التالي قائمة العناوين إلى 20 عاملاً، ويختار البروكسي التالي من القائمة لكل مهمة. يمكن استخدام itertools.cycle مع عدة خيوط دون مشاكل في استخراج القيمة التالية، ولكن إذا كنت بحاجة إلى ترتيب دقيق، استخدم القفل معه.

import itertools
from concurrent.futures import ThreadPoolExecutor
import requests

lines = open("proxies.txt").read().split()  # user:pass@ip:port per line
pool = itertools.cycle(lines)

def get(url):
    p = "http://" + next(pool)
    try:
        return url, requests.get(url, proxies={"http": p, "https": p}, timeout=20).status_code
    except requests.RequestException as e:
        return url, type(e).__name__

urls = ["https://example.com/item/" + str(i) for i in range(200)]
with ThreadPoolExecutor(max_workers=20) as ex:
    for url, result in ex.map(get, urls):
        print(url, result)

هل يمكن استخدام مكتبات غير متزامنة مثل httpx أو aiohttp؟

نعم. يمكنك إضافة عنوان البروكسي إلى الوسيطة proxy عند إنشاء عميل في httpx، وللاستخدام مع SOCKS5، يمكنك تثبيت httpx[socks] لاستخدام عنوان socks5. العميل غير المتزامن AsyncClient يدعم نفس الوسائط.

aiohttp يتطلب إضافة عنوان HTTP بروكسي في معامل proxy واستخدام BasicAuth للتحقق. لاستخدام SOCKS5، يجب استخدام حزمة aiohttp-socks.

تتيح الطريقة غير المتزامنة معالجة آلاف الطلبات بموارد محدودة، ولكن بدون تحديد حد للطلبات المتزامنة، قد يتسبب ذلك في حمل زائد على الموقع والبروكسي. استخدم asyncio.Semaphore لتحديد عدد الطلبات التي تتم معالجتها في نفس الوقت.

import httpx

with httpx.Client(proxy="http://spx_user:spx_pass@198.51.100.23:8000", timeout=20) as client:
    print(client.get("https://api.ipify.org").text)

كيف يتم حل الأخطاء الشائعة؟

خطأ Missing dependencies for SOCKS support يعني أن requests[socks] غير مثبتة. يرجى إعادة تثبيتها في نفس البيئة الافتراضية.

407 Proxy Authentication Required يعني أن اسم المستخدم أو كلمة المرور غير صحيحة، أو أنك تستخدم تفويض الـ IP ولم يتم تسجيل IP الخادم. إذا كانت كلمة المرور تحتوي على أحرف خاصة، يجب تحويلها باستخدام urllib.parse.quote.

في معظم الحالات، يكون سبب انتهاء مهلة الاتصال هو عدم تطابق المنفذ. تأكد من استخدام منفذ HTTP لعنوان HTTP ومنفذ SOCKS5 لعنوان socks5h. إضافة قيمة timeout يمنع الطلبات التي لا تستجيب من إيقاف العملية بأكملها.

• خطأ SOCKS: pip install "requests[socks]"
• 407: تحقق من اسم المستخدم/كلمة المرور أو الـ IP المسجل
• انتهى الوقت: التمييز بين منفذ HTTP وSOCKS5
• إضافة timeout لجميع الطلبات