إعداد البروكسي في إطار عمل سكرابي: دليل كامل مع أمثلة على الكود

Scrapy هو أحد أقوى أطر عمل Python للتجريف على الويب، ولكن بدون إعداد الوكيل الصحيح، ستتلقى أدوات التجريف لديك حظراً بعد بضع دقائق من العمل. في هذا الدليل، سأعرض جميع طرق دمج الوكيل في Scrapy: من الإعدادات الأساسية إلى الطرق المتقدمة لتدوير عناوين IP مع معالجة الأخطاء تلقائياً.

يعتمد المحتوى على تجربة حقيقية في تجريف مواقع التجارة الإلكترونية الكبيرة والمواقع المحمية. ستحصل على أمثلة جاهزة من الشيفرة التي يمكنك استخدامها مباشرة في مشاريعك.

لماذا يحصل Scrapy على حظر بدون وكيل

تستخدم المواقع الحديثة حماية متعددة المستويات ضد التجريف. حتى إذا قمت بإعداد User-Agent وتأخيرات بين الطلبات، فإن عنوان IP الخاص بك يكشف عن الأتمتة بعدة علامات:

• تكرار الطلبات: يقوم عنوان IP واحد بعمل أكثر من 100 طلب في الدقيقة - علامة واضحة على الروبوت
• أنماط السلوك: تصفح متسلسل للصفحات بدون انتقالات عشوائية
• عدم وجود JavaScript: لا يقوم Scrapy بتنفيذ JavaScript، مما يسهل اكتشافه
• الموقع الجغرافي: الوصول من مركز بيانات بدلاً من الشبكة المنزلية

النتيجة - حظر حسب IP لعدة ساعات أو أيام. تستخدم الأسواق (Amazon، Wildberries، Ozon) والشبكات الاجتماعية والمواقع التي تستخدم Cloudflare حماية عدوانية بشكل خاص. تحل الوكلاء هذه المشكلة عن طريق توزيع الطلبات بين العديد من عناوين IP.

الإعداد الأساسي للوكيل في Scrapy

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

الطريقة 1: عبر meta في الطلب

import scrapy

class MySpider(scrapy.Spider):
    name = 'example'
    start_urls = ['https://example.com']
    
    def start_requests(self):
        proxy = 'http://username:password@proxy.example.com:8080'
        
        for url in self.start_urls:
            yield scrapy.Request(
                url=url,
                callback=self.parse,
                meta={'proxy': proxy}
            )
    
    def parse(self, response):
        # منطق التجريف الخاص بك
        self.log(f'Scraped {response.url} via {response.meta["proxy"]}')

يعتمد تنسيق الوكيل على البروتوكول وطريقة المصادقة:

• http://proxy.example.com:8080 - بدون مصادقة
• http://user:pass@proxy.example.com:8080 - مع اسم المستخدم/كلمة المرور
• socks5://user:pass@proxy.example.com:1080 - وكيل SOCKS5

الطريقة 2: الإعداد العالمي في settings.py

# settings.py

# وكيل HTTP لجميع الطلبات
HTTPPROXY_ENABLED = True
HTTPPROXY_AUTH_ENCODING = 'utf-8'

# الإعداد عبر متغيرات البيئة
HTTP_PROXY = 'http://username:password@proxy.example.com:8080'
HTTPS_PROXY = 'http://username:password@proxy.example.com:8080'

هذه الطريقة مريحة للاختبار السريع، لكنها غير مناسبة للإنتاج: لا يوجد تدوير IP، وعند تعطل الوكيل يتوقف كل أداة التجريف، ولا يمكن استخدام وكلاء مختلفين لمواقع مختلفة.

إنشاء Middleware وكيل مخصص

للتجريف في الإنتاج، تحتاج إلى middleware خاص يدير مجموعة الوكلاء، ويعالج الأخطاء، ويقوم بتدوير IP تلقائياً. إليك تنفيذ أساسي:

# middlewares.py

import random
from scrapy import signals
from scrapy.exceptions import NotConfigured

class RandomProxyMiddleware:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
    
    @classmethod
    def from_crawler(cls, crawler):
        # تحميل قائمة الوكلاء من الإعدادات
        proxy_list = crawler.settings.getlist('PROXY_LIST')
        
        if not proxy_list:
            raise NotConfigured('PROXY_LIST not configured')
        
        return cls(proxy_list)
    
    def process_request(self, request, spider):
        # اختيار وكيل عشوائي من المجموعة
        proxy = random.choice(self.proxy_list)
        request.meta['proxy'] = proxy
        
        spider.logger.info(f'Using proxy: {proxy}')
    
    def process_exception(self, request, exception, spider):
        # عند حدوث خطأ، نجرب وكيل آخر
        proxy = random.choice(self.proxy_list)
        request.meta['proxy'] = proxy
        
        spider.logger.warning(
            f'Proxy error, switching to: {proxy}'
        )
        
        return request

الآن نقوم بإعداد استخدام middleware في settings.py:

# settings.py

# قائمة الوكلاء (يمكن تحميلها من ملف أو API)
PROXY_LIST = [
    'http://user1:pass1@proxy1.example.com:8080',
    'http://user2:pass2@proxy2.example.com:8080',
    'http://user3:pass3@proxy3.example.com:8080',
    # ... أضف 50+ وكيل لتدوير فعال
]

# توصيل middleware
DOWNLOADER_MIDDLEWARES = {
    'myproject.middlewares.RandomProxyMiddleware': 350,
    'scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware': 400,
}

# محاولات إعادة الطلب عند حدوث الأخطاء
RETRY_TIMES = 3
RETRY_HTTP_CODES = [500, 502, 503, 504, 408, 429]

تدوير الوكلاء: ثلاث طرق فعالة

الاختيار العشوائي للوكلاء (كما في المثال أعلاه) هو أبسط طريقة، ولكنها ليست الأكثر فعالية. دعونا نستعرض ثلاث استراتيجيات للتدوير لمواقف مختلفة.

الطريقة 1: Round-robin (تدوير متسلسل)

يتم اختيار الوكلاء بشكل دائري. مناسب لتوزيع الحمل بشكل متساوٍ:

class RoundRobinProxyMiddleware:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
        self.current_index = 0
    
    @classmethod
    def from_crawler(cls, crawler):
        proxy_list = crawler.settings.getlist('PROXY_LIST')
        return cls(proxy_list)
    
    def process_request(self, request, spider):
        # أخذ الوكيل التالي بشكل دائري
        proxy = self.proxy_list[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.proxy_list)
        
        request.meta['proxy'] = proxy

الطريقة 2: تدوير ذكي مع قائمة سوداء

تتبع الوكلاء المproblematic واستبعادهم مؤقتاً من التدوير:

import time
from collections import defaultdict

class SmartProxyMiddleware:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
        self.proxy_errors = defaultdict(int)
        self.blacklist = set()
        self.blacklist_timeout = 300  # 5 دقائق
        self.blacklist_time = {}
    
    @classmethod
    def from_crawler(cls, crawler):
        proxy_list = crawler.settings.getlist('PROXY_LIST')
        return cls(proxy_list)
    
    def get_working_proxies(self):
        # إزالة الوكلاء من القائمة السوداء الذين انتهت مدة حظرهم
        current_time = time.time()
        expired = [
            proxy for proxy, ban_time in self.blacklist_time.items()
            if current_time - ban_time > self.blacklist_timeout
        ]
        
        for proxy in expired:
            self.blacklist.discard(proxy)
            self.proxy_errors[proxy] = 0
        
        # إرجاع الوكلاء العاملين
        return [p for p in self.proxy_list if p not in self.blacklist]
    
    def process_request(self, request, spider):
        working_proxies = self.get_working_proxies()
        
        if not working_proxies:
            spider.logger.error('All proxies are blacklisted!')
            return
        
        proxy = random.choice(working_proxies)
        request.meta['proxy'] = proxy
    
    def process_response(self, request, response, spider):
        # إذا حصلنا على حظر - نضيف إلى القائمة السوداء
        if response.status in [403, 429, 503]:
            proxy = request.meta.get('proxy')
            self.proxy_errors[proxy] += 1
            
            if self.proxy_errors[proxy] >= 3:
                self.blacklist.add(proxy)
                self.blacklist_time[proxy] = time.time()
                spider.logger.warning(
                    f'Proxy {proxy} blacklisted for {self.blacklist_timeout}s'
                )
        
        return response

الطريقة 3: تدوير عبر واجهة برمجة التطبيقات لمزود الخدمة

العديد من مزودي الوكلاء (بما في ذلك الوكلاء السكنيين) يقدمون نقطة نهاية متغيرة - عنوان URL واحد، الذي يغير IP تلقائياً مع كل طلب:

# settings.py

# نقطة نهاية واحدة مع تدوير تلقائي
ROTATING_PROXY = 'http://username:password@rotating.proxy.com:8080'

# Middleware بسيطة
class RotatingProxyMiddleware:
    def __init__(self, proxy):
        self.proxy = proxy
    
    @classmethod
    def from_crawler(cls, crawler):
        proxy = crawler.settings.get('ROTATING_PROXY')
        return cls(proxy)
    
    def process_request(self, request, spider):
        # عنوان واحد، ولكن كل طلب يأتي من IP جديد
        request.meta['proxy'] = self.proxy

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

المصادقة: اسم المستخدم/كلمة المرور مقابل قائمة IP البيضاء

يقدم مزودو الوكلاء طريقتين للمصادقة. يؤثر الاختيار على سرعة الاتصال وسهولة الإعداد.

مصادقة User:Pass

يتم تمرير اسم المستخدم وكلمة المرور في عنوان URL للوكيل. يقوم Scrapy تلقائياً بتحويلها إلى رأس HTTP Proxy-Authorization:

proxy = 'http://username:password@proxy.example.com:8080'
request.meta['proxy'] = proxy

# سيضيف Scrapy تلقائياً الرأس:
# Proxy-Authorization: Basic dXNlcm5hbWU6cGFzc3dvcmQ=

الإيجابيات: يعمل من أي IP، سهل تغيير الوكيل في الشيفرة.
السلبيات: تأخير بسيط على كل طلب (~50-100ms)، بيانات الاعتماد في شكل مكشوف في الشيفرة.

مصادقة قائمة IP البيضاء

تقوم بإضافة IP الخاص بالخادم الخاص بك إلى القائمة البيضاء لدى المزود، ولا تحتاج إلى مصادقة:

proxy = 'http://proxy.example.com:8080'  # بدون اسم مستخدم/كلمة مرور
request.meta['proxy'] = proxy

الإيجابيات: أسرع بـ 50-100ms، أكثر أماناً (لا توجد بيانات اعتماد في الشيفرة).
السلبيات: يعمل فقط من IPs معينة، تحتاج إلى تحديث القائمة البيضاء عند تغيير الخادم.

معالجة الأخطاء وتغيير IP تلقائياً

حتى مع الوكلاء الجيدين، ستحدث أخطاء: مهلات، رفض الاتصال، حظر. تعتبر معالجة الأخطاء الصحيحة حاسمة لعمل أداة التجريف بشكل مستقر.

معالجة رموز HTTP

class ProxyMiddleware:
    def process_response(self, request, response, spider):
        # الرموز التي تحتاج إلى تغيير الوكيل وإعادة المحاولة
        ban_codes = [403, 407, 429, 503]
        
        if response.status in ban_codes:
            proxy = request.meta.get('proxy')
            spider.logger.warning(
                f'Got {response.status} from {proxy}, retrying...'
            )
            
            # وضع علامة لإعادة المحاولة مع وكيل جديد
            request.meta['dont_retry'] = False
            request.meta['proxy'] = self.get_new_proxy()
            
            return request
        
        return response

معالجة استثناءات الشبكة

from twisted.internet.error import TimeoutError, ConnectionRefusedError
from scrapy.exceptions import IgnoreRequest

class ProxyMiddleware:
    def process_exception(self, request, exception, spider):
        # أخطاء الاتصال بالوكيل
        proxy_errors = (
            TimeoutError,
            ConnectionRefusedError,
            ConnectionLost,
        )
        
        if isinstance(exception, proxy_errors):
            proxy = request.meta.get('proxy')
            spider.logger.error(
                f'Proxy {proxy} connection failed: {exception}'
            )
            
            # تغيير الوكيل وتجربة مرة أخرى
            request.meta['proxy'] = self.get_new_proxy()
            return request
        
        # بالنسبة للأخطاء الأخرى، نستخدم المعالجة القياسية
        return None

الكشف عن الحظر من خلال المحتوى

بعض المواقع تعيد HTTP 200، ولكن تظهر CAPTCHA أو صفحة حظر:

class ProxyMiddleware:
    def process_response(self, request, response, spider):
        # علامات الحظر في المحتوى
        ban_indicators = [
            'captcha',
            'access denied',
            'blocked',
            'unusual traffic',
            'robot check',
        ]
        
        body_text = response.text.lower()
        
        if any(indicator in body_text for indicator in ban_indicators):
            spider.logger.warning(
                f'Ban page detected from {request.meta.get("proxy")}'
            )
            
            # تغيير الوكيل وإعادة المحاولة
            request.meta['proxy'] = self.get_new_proxy()
            return request
        
        return response

أي نوع من الوكلاء تختار لـ Scrapy

يعتمد اختيار نوع الوكيل على الموقع المستهدف، والميزانية، والسرعة المطلوبة للتجريف. إليك مقارنة بين الخيارات الرئيسية:

توصيات للاختيار

لتجريف الأسواق (Amazon، Wildberries، Ozon، AliExpress) - استخدم فقط الوكلاء السكنيين. هذه المواقع تحظر مراكز البيانات بشكل عدواني. تحتاج إلى تدوير واستهداف جغرافي (على سبيل المثال، IPs روسية لـ Wildberries).

لتجريف مواقع الأخبار، المدونات، المنتديات - يمكن استخدام وكلاء مراكز البيانات. الحماية قليلة، الأهمية هي السرعة وتكلفة المرور المنخفضة.

لتجريف المواقع التي تستخدم Cloudflare - الوكلاء السكنيون إلزاميون. تكتشف مراكز بيانات Cloudflare تقريباً على الفور. أضف مكتبة cloudscraper إلى Scrapy لتجاوز تحديات JavaScript.

لتجريف Google Search، أدوات SEO - استخدم الوكلاء السكنيين مع الاستهداف الجغرافي. تعرض Google نتائج مختلفة لدول ومدن مختلفة.

تقنيات متقدمة: الجلسات وعناوين IP الثابتة

عند تجريف بعض المواقع، تحتاج إلى الحفاظ على IP واحد طوال الجلسة (المصادقة، سلة التسوق، النماذج متعددة الخطوات). إليك كيفية تنفيذ الجلسات الثابتة في Scrapy.

IP ثابت لنطاق واحد

from urllib.parse import urlparse

class StickyProxyMiddleware:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
        # قاموس: نطاق -> وكيل
        self.domain_proxy_map = {}
    
    @classmethod
    def from_crawler(cls, crawler):
        proxy_list = crawler.settings.getlist('PROXY_LIST')
        return cls(proxy_list)
    
    def process_request(self, request, spider):
        # استخراج النطاق من URL
        domain = urlparse(request.url).netloc
        
        # إذا كان لهذا النطاق وكيل بالفعل - استخدمه
        if domain in self.domain_proxy_map:
            proxy = self.domain_proxy_map[domain]
        else:
            # خلاف ذلك، اختر جديداً وتذكره
            proxy = random.choice(self.proxy_list)
            self.domain_proxy_map[domain] = proxy
            spider.logger.info(f'Assigned {proxy} to {domain}')
        
        request.meta['proxy'] = proxy

IP ثابت مع مهلة للجلسة

خيار أكثر تقدماً: يتم ربط الوكيل بالنطاق لفترة معينة (على سبيل المثال، 10 دقائق)، ثم يتم تغييره:

import time
from urllib.parse import urlparse

class SessionProxyMiddleware:
    def __init__(self, proxy_list, session_timeout=600):
        self.proxy_list = proxy_list
        self.session_timeout = session_timeout  # 10 دقائق
        # قاموس: نطاق -> (وكيل، وقت الإنشاء)
        self.sessions = {}
    
    @classmethod
    def from_crawler(cls, crawler):
        proxy_list = crawler.settings.getlist('PROXY_LIST')
        timeout = crawler.settings.getint('PROXY_SESSION_TIMEOUT', 600)
        return cls(proxy_list, timeout)
    
    def get_proxy_for_domain(self, domain):
        current_time = time.time()
        
        # تحقق مما إذا كانت هناك جلسة نشطة
        if domain in self.sessions:
            proxy, created_at = self.sessions[domain]
            
            # إذا لم تنتهِ الجلسة - استخدم نفس الوكيل
            if current_time - created_at < self.session_timeout:
                return proxy
        
        # إنشاء جلسة جديدة مع وكيل جديد
        new_proxy = random.choice(self.proxy_list)
        self.sessions[domain] = (new_proxy, current_time)
        
        return new_proxy
    
    def process_request(self, request, spider):
        domain = urlparse(request.url).netloc
        proxy = self.get_proxy_for_domain(domain)
        request.meta['proxy'] = proxy

التكامل مع Middleware الكوكيز

للحصول على جلسات كاملة، يجب مزامنة الوكيل والكوكيز. يحتفظ Scrapy بالكوكيز بشكل منفصل لكل نطاق، ولكن عند تغيير الوكيل، يجب مسح الكوكيز:

# settings.py

# تفعيل middleware الكوكيز
COOKIES_ENABLED = True
COOKIES_DEBUG = False

# Middleware لمزامنة الوكيل والكوكيز
class ProxyCookieMiddleware:
    def process_request(self, request, spider):
        # الحصول على الوكيل الحالي
        current_proxy = request.meta.get('proxy')
        
        # إذا تم تغيير الوكيل - امسح الكوكيز
        previous_proxy = request.meta.get('previous_proxy')
        
        if previous_proxy and previous_proxy != current_proxy:
            # امسح الكوكيز لهذا النطاق
            jar = spider.crawler.engine.downloader.middleware.middlewares[0].jars
            domain = urlparse(request.url).netloc
            
            if domain in jar:
                jar[domain].clear()
                spider.logger.info(f'Cleared cookies for {domain}')
        
        request.meta['previous_proxy'] = current_proxy

الخاتمة

الإعداد الصحيح للوكيل في Scrapy هو أساس التجريف المستقر بدون حظر. لقد استعرضنا جميع الجوانب الرئيسية: من الدمج الأساسي إلى تقنيات متقدمة للتدوير وإدارة الجلسات.

الاستنتاجات الرئيسية:

• لإنتاج، استخدم middleware مخصص مع تدوير ذكي وقائمة سوداء لعناوين IP المproblematic
• عالج جميع أنواع الأخطاء: رموز HTTP، استثناءات الشبكة، الحظر من خلال المحتوى
• اختر نوع الوكيل المناسب للمهمة: مراكز البيانات لمواقع بسيطة، الوكلاء السكنيون للمواقع المحمية
• للمواقع التي تتطلب مصادقة، استخدم الجلسات الثابتة مع ربط الوكيل بالنطاق
• ابدأ بمجموعة من 10-50 وكيل، وزدها عند زيادة الحمل

إذا كنت تخطط لتجريف مواقع محمية (أسواق، شبكات اجتماعية، مواقع تستخدم Cloudflare)، أوصي باستخدام الوكلاء السكنيين - حيث يوفرون أقصى درجات الخصوصية وأقل خطر للحظر. للتجريف عالي السرعة، اختر مزودين مع نقطة نهاية متغيرة ومجموعة من 10,000 عنوان IP.

جميع أمثلة الشيفرة في هذه المقالة تم اختبارها على Scrapy 2.x وجاهزة للاستخدام في الإنتاج. قم بتكييفها مع احتياجاتك وزيادة حجمها مع نمو المشروع.