أتمتة تغيير البروكسي عبر API: كيفية إعداد التدوير للاستخراج والتحكيم

التبديل اليدوي للبروكسي عند العمل مع مئات الطلبات هو إهدار للوقت والمال. تتيح دوران API التبديل التلقائي لعناوين IP عند الحظر، وتوزيع الحمل وتوسيع نطاق الاستخراج أو إدارة الحسابات المتعددة. في هذا الدليل، سنتناول كيفية إعداد التبديل التلقائي للبروكسي لمهام مختلفة: من استخراج بيانات المتاجر الإلكترونية إلى إدارة الحسابات في Facebook Ads.

المحتوى مناسب للمطورين الذين يكتبون أدوات استخراج بلغة Python أو Node.js، وكذلك للمسوقين بالعمولة الذين يستخدمون أدوات جاهزة مع تكامل API.

لماذا نحتاج إلى أتمتة تبديل البروكسي عبر API

يحل الدوران التلقائي لعناوين IP عبر API عدة مشاكل حرجة يواجهها المتخصصون في مجالات مختلفة:

المزايا الرئيسية لأتمتة API مقارنة بالتبديل اليدوي:

• السرعة: يحدث تبديل IP في أجزاء من الثانية برمجياً، دون تدخل بشري
• قابلية التوسع: يمكن إدارة آلاف البروكسيات في وقت واحد من خلال واجهة موحدة
• مقاومة الأعطال: استبدال تلقائي للبروكسيات غير العاملة دون إيقاف العملية
• المرونة: تكوين قواعد الدوران لمهمة محددة: حسب الوقت، عدد الطلبات، الموقع الجغرافي
• التوفير: الاستخدام الأمثل للبيانات من خلال موازنة الحمل

سيناريو استخدام نموذجي: أنت تستخرج أسعار المنافسين على Wildberries. بدون API، تحتاج إلى تتبع الحظر يدوياً، والدخول إلى لوحة مزود البروكسي، ونسخ البيانات الجديدة، ولصقها في السكريبت. مع API، يحدث كل هذا تلقائياً: يحصل السكريبت على خطأ 429 (Too Many Requests)، يرسل طلباً إلى API خدمة البروكسي، يحصل على IP جديد ويستمر في العمل.

أنواع دوران البروكسي: الجلسات الثابتة مقابل التبديل التلقائي

قبل إعداد الأتمتة، من المهم فهم الفرق بين أنواع دوران عناوين IP. يعتمد اختيار الاستراتيجية على مهمتك.

الجلسات الثابتة (Sticky Sessions)

عند استخدام الجلسات الثابتة، يتم تثبيت عنوان IP واحد لجلستك لفترة زمنية محددة (عادة من 5 إلى 30 دقيقة). يحدث التبديل فقط بعد انتهاء وقت الجلسة أو بناءً على طلب API الخاص بك.

متى تستخدم:

• العمل مع حسابات وسائل التواصل الاجتماعي (Instagram، Facebook) - التبديل المتكرر لـ IP يثير الشكوك
• ملء النماذج متعددة الصفحات حيث تحتاج إلى الحفاظ على الجلسة
• اختبار الإعلانات من منطقة معينة خلال الجلسة
• استخراج المواقع مع تسجيل الدخول، حيث سيؤدي تبديل IP إلى تسجيل الخروج

مثال على طلب API لإنشاء جلسة ثابتة (عادة ما يستخدم تنسيق تسجيل دخول خاص):

// التنسيق: username-session-SESSIONID:password
// SESSIONID — أي سلسلة نصية، نفس القيمة = نفس IP

proxy = "username-session-abc123:password@gate.proxycove.com:8000"

// جميع الطلبات مع session-abc123 ستحصل على نفس IP خلال الجلسة
// للحصول على IP جديد استخدم SESSIONID مختلف: session-xyz789

الدوران التلقائي لكل طلب

يتغير عنوان IP مع كل اتصال جديد بخادم البروكسي. هذا هو السلوك القياسي لـ البروكسيات السكنية دون تحديد معاملات الجلسة.

متى تستخدم:

• الاستخراج الجماعي بدون تسجيل دخول (الأسعار، جهات الاتصال، الإعلانات)
• تجاوز حدود المعدل العدوانية على واجهات API العامة
• جمع البيانات من المواقع التي تحظر IP بعد 10-20 طلب
• التحقق من توفر المحتوى من مناطق مختلفة

مثال استخدام في Python (كل طلب = IP جديد):

import requests

proxy = {
    "http": "http://username:password@gate.proxycove.com:8000",
    "https": "http://username:password@gate.proxycove.com:8000"
}

# كل طلب سيحصل على IP جديد
for i in range(10):
    response = requests.get("https://api.ipify.org", proxies=proxy)
    print(f"الطلب {i+1}, IP: {response.text}")

الدوران حسب المؤقت

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

متى تستخدم:

• تحسين استهلاك البيانات - التبديل فقط عند الضرورة
• العمل مع المواقع التي تتبع الأنماط (التبديل المتكرر جداً = حظر)
• التوازن بين عدم الكشف عن الهوية واستقرار الجلسة

أساسيات العمل مع API لخدمات البروكسي

توفر معظم مزودي البروكسي الحديثين طريقتين للإدارة: من خلال لوحة الويب ومن خلال API. يوفر API وصولاً برمجياً إلى الوظائف: الحصول على قائمة البروكسيات، إنشاء جلسات جديدة، التحقق من الرصيد، إحصائيات الاستخدام.

طرق API النموذجية لخدمات البروكسي

على الرغم من أن كل مزود لديه وثائقه الخاصة، عادة ما تتضمن الطرق القياسية:

• GET /api/v1/proxy/list — الحصول على قائمة البروكسيات المتاحة مع التصفية حسب البلد والنوع
• POST /api/v1/proxy/rotate — تبديل IP قسرياً في الجلسة النشطة
• GET /api/v1/account/balance — التحقق من رصيد البيانات أو المال في الحساب
• GET /api/v1/stats — إحصائيات الاستخدام: حجم البيانات، عدد الطلبات، الأخطاء
• POST /api/v1/session/create — إنشاء جلسة ثابتة جديدة مع معاملات (البلد، المدينة، المدة)

عادة ما تتم المصادقة من خلال مفتاح API في رأس الطلب:

curl -X GET "https://api.provider.com/v1/proxy/list?country=US" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json"

عادة ما يأتي الرد بتنسيق JSON:

{
  "status": "success",
  "data": {
    "proxies": [
      {
        "ip": "123.45.67.89",
        "port": 8000,
        "country": "US",
        "city": "New York",
        "protocol": "http",
        "username": "user123",
        "password": "pass456"
      }
    ],
    "total": 150,
    "available": 147
  }
}

إدارة الجلسات عبر API

للمهام التي تتطلب التحكم في عمر IP (إدارة حسابات متعددة، العمل مع الحسابات)، يتم استخدام إنشاء جلسات مسماة عبر API. يتيح ذلك إدارة عشرات ومئات عناوين IP المعزولة برمجياً.

مثال على إنشاء جلسة مع معاملات:

POST /api/v1/session/create
{
  "country": "US",
  "state": "California",
  "session_duration": 600,  // 10 دقائق
  "session_id": "facebook_account_001"
}

// الرد:
{
  "status": "success",
  "session": {
    "id": "facebook_account_001",
    "proxy": "gate.provider.com:8000",
    "username": "user-session-facebook_account_001",
    "password": "your_password",
    "ip": "45.67.89.123",
    "expires_at": "2024-01-15T15:30:00Z"
  }
}

الآن يمكنك استخدام هذا البروكسي في السكريبت الخاص بك أو متصفح مكافحة الكشف، وسيبقى IP دون تغيير لمدة 10 دقائق. لتمديد الجلسة، يتم إرسال طلب متكرر بنفس session_id.

أمثلة الأتمتة بلغة Python: requests و Selenium و Scrapy

Python هي اللغة الأكثر شعبية للاستخراج والأتمتة. دعونا ننظر في أمثلة لتكامل دوران API للبروكسي مع الأدوات الرئيسية.

التبديل التلقائي للبروكسي في requests

تُستخدم مكتبة requests لطلبات HTTP البسيطة. للدوران التلقائي، سننشئ فئة غلاف تبدل البروكسي عند الأخطاء:

import requests
import random
import time

class RotatingProxySession:
    def __init__(self, proxy_list):
        """
        proxy_list: قائمة قواميس ببيانات البروكسي
        [{"http": "http://user:pass@ip:port", "https": "..."}]
        """
        self.proxy_list = proxy_list
        self.current_proxy = None
        self.session = requests.Session()
        self.rotate()
    
    def rotate(self):
        """اختيار بروكسي عشوائي من القائمة"""
        self.current_proxy = random.choice(self.proxy_list)
        self.session.proxies.update(self.current_proxy)
        print(f"تم التبديل إلى بروكسي: {self.current_proxy['http']}")
    
    def get(self, url, max_retries=3, **kwargs):
        """طلب GET مع دوران تلقائي عند الأخطاء"""
        for attempt in range(max_retries):
            try:
                response = self.session.get(url, timeout=10, **kwargs)
                
                # إذا حصلنا على حظر - تبديل البروكسي
                if response.status_code in [403, 429, 503]:
                    print(f"حصلنا على {response.status_code}, نبدل البروكسي...")
                    self.rotate()
                    time.sleep(2)
                    continue
                
                return response
                
            except requests.exceptions.ProxyError:
                print(f"البروكسي لا يعمل، محاولة {attempt+1}/{max_retries}")
                self.rotate()
                time.sleep(2)
            
            except requests.exceptions.Timeout:
                print("انتهت المهلة، نبدل البروكسي...")
                self.rotate()
                time.sleep(2)
        
        raise Exception(f"فشل تنفيذ الطلب بعد {max_retries} محاولات")

# الاستخدام:
proxies = [
    {"http": "http://user1:pass@gate1.com:8000", "https": "http://user1:pass@gate1.com:8000"},
    {"http": "http://user2:pass@gate2.com:8000", "https": "http://user2:pass@gate2.com:8000"},
]

session = RotatingProxySession(proxies)

# استخراج Wildberries
for page in range(1, 50):
    url = f"https://www.wildberries.ru/catalog/page={page}"
    response = session.get(url)
    print(f"الصفحة {page}: {response.status_code}")

التكامل مع Selenium لأتمتة المتصفح

يُستخدم Selenium لاستخراج المواقع مع JavaScript وأتمتة الإجراءات في المتصفح. لتبديل البروكسي، تحتاج إلى إعادة إنشاء السائق، حيث يتم تعيين إعدادات البروكسي عند التهيئة:

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
import time

class SeleniumRotatingProxy:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
        self.driver = None
        self.current_proxy_index = 0
    
    def create_driver(self):
        """إنشاء سائق جديد مع البروكسي الحالي"""
        if self.driver:
            self.driver.quit()
        
        proxy = self.proxy_list[self.current_proxy_index]
        
        chrome_options = Options()
        chrome_options.add_argument(f'--proxy-server={proxy}')
        chrome_options.add_argument('--headless')  # بدون واجهة رسومية
        
        self.driver = webdriver.Chrome(options=chrome_options)
        print(f"تم إنشاء سائق مع بروكسي: {proxy}")
    
    def rotate(self):
        """التبديل إلى البروكسي التالي"""
        self.current_proxy_index = (self.current_proxy_index + 1) % len(self.proxy_list)
        self.create_driver()
    
    def get_with_retry(self, url, max_retries=3):
        """فتح URL مع تبديل تلقائي للبروكسي عند الأخطاء"""
        for attempt in range(max_retries):
            try:
                if not self.driver:
                    self.create_driver()
                
                self.driver.get(url)
                
                # التحقق من الحظر (مثلاً، البحث عن كابتشا)
                if "captcha" in self.driver.page_source.lower():
                    print("تم اكتشاف كابتشا، نبدل البروكسي...")
                    self.rotate()
                    time.sleep(3)
                    continue
                
                return self.driver.page_source
                
            except Exception as e:
                print(f"خطأ: {e}, نبدل البروكسي (محاولة {attempt+1})")
                self.rotate()
                time.sleep(3)
        
        raise Exception("فشل تحميل الصفحة")

# الاستخدام:
proxies = [
    "http://user:pass@gate1.com:8000",
    "http://user:pass@gate2.com:8000",
]

bot = SeleniumRotatingProxy(proxies)

# استخراج Ozon
for i in range(10):
    html = bot.get_with_retry(f"https://www.ozon.ru/category/page-{i}")
    print(f"حصلنا على HTML للصفحة {i}, الطول: {len(html)}")

bot.driver.quit()

Scrapy مع middleware لدوران البروكسي

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

# middlewares.py

import random
from scrapy.exceptions import IgnoreRequest

class RotatingProxyMiddleware:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
    
    @classmethod
    def from_crawler(cls, crawler):
        # الحصول على قائمة البروكسيات من الإعدادات
        proxy_list = crawler.settings.getlist('ROTATING_PROXY_LIST')
        return cls(proxy_list)
    
    def process_request(self, request, spider):
        # تعيين بروكسي عشوائي لكل طلب
        proxy = random.choice(self.proxy_list)
        request.meta['proxy'] = proxy
        spider.logger.info(f'نستخدم بروكسي: {proxy}')
    
    def process_exception(self, request, exception, spider):
        # عند خطأ البروكسي - إعادة المحاولة مع آخر
        proxy = random.choice(self.proxy_list)
        spider.logger.warning(f'خطأ في البروكسي، نتبدل إلى: {proxy}')
        request.meta['proxy'] = proxy
        return request  # إعادة الطلب

# settings.py

DOWNLOADER_MIDDLEWARES = {
    'myproject.middlewares.RotatingProxyMiddleware': 350,
}

ROTATING_PROXY_LIST = [
    'http://user:pass@gate1.com:8000',
    'http://user:pass@gate2.com:8000',
    'http://user:pass@gate3.com:8000',
]

# إعادة الطلبات عند الأخطاء
RETRY_TIMES = 5
RETRY_HTTP_CODES = [403, 429, 500, 502, 503]

الآن سيحصل كل طلب Scrapy تلقائياً على بروكسي عشوائي من القائمة، وعند الأخطاء سيتم إعادة المحاولة مع IP آخر.

الأتمتة بلغة Node.js: axios و Puppeteer و Playwright

Node.js شائع لإنشاء أدوات استخراج وبوتات بفضل عدم التزامن والتكامل الجيد مع أدوات المتصفح. دعونا ننظر في أمثلة دوران البروكسي في المكتبات الرئيسية.

Axios مع دوران تلقائي

Axios هي مكتبة لطلبات HTTP. سننشئ فئة مع مجموعة بروكسيات واستبدال تلقائي عند الأخطاء:

const axios = require('axios');
const HttpsProxyAgent = require('https-proxy-agent');

class RotatingProxyClient {
  constructor(proxyList) {
    this.proxyList = proxyList;
    this.currentIndex = 0;
  }

  getProxy() {
    const proxy = this.proxyList[this.currentIndex];
    this.currentIndex = (this.currentIndex + 1) % this.proxyList.length;
    return proxy;
  }

  async request(url, options = {}, maxRetries = 3) {
    for (let i = 0; i < maxRetries; i++) {
      const proxy = this.getProxy();
      const agent = new HttpsProxyAgent(proxy);

      try {
        const response = await axios.get(url, {
          ...options,
          httpsAgent: agent,
          timeout: 10000
        });

        // إذا حصلنا على حظر - المحاولة التالية
        if ([403, 429, 503].includes(response.status)) {
          console.log(`الحالة ${response.status}, نبدل البروكسي...`);
          continue;
        }

        return response.data;

      } catch (error) {
        console.log(`خطأ مع بروكسي ${proxy}: ${error.message}`);
        if (i === maxRetries - 1) throw error;
      }
    }
  }
}

// الاستخدام:
const proxies = [
  'http://user:pass@gate1.com:8000',
  'http://user:pass@gate2.com:8000',
];

const client = new RotatingProxyClient(proxies);

(async () => {
  for (let page = 1; page <= 20; page++) {
    const data = await client.request(`https://api.example.com/products?page=${page}`);
    console.log(`الصفحة ${page}: تم الحصول على ${data.length} منتجات`);
  }
})();

Puppeteer مع دوران البروكسي

يدير Puppeteer متصفح Chrome. يتم تعيين البروكسي عند تشغيل المتصفح، لذلك للتبديل تحتاج إلى إعادة إنشاء النسخة:

const puppeteer = require('puppeteer');

class PuppeteerRotatingProxy {
  constructor(proxyList) {
    this.proxyList = proxyList;
    this.currentIndex = 0;
    this.browser = null;
  }

  async createBrowser() {
    if (this.browser) await this.browser.close();

    const proxy = this.proxyList[this.currentIndex];
    console.log(`نشغل المتصفح مع بروكسي: ${proxy}`);

    this.browser = await puppeteer.launch({
      headless: true,
      args: [`--proxy-server=${proxy}`]
    });
  }

  rotate() {
    this.currentIndex = (this.currentIndex + 1) % this.proxyList.length;
  }

  async scrape(url, maxRetries = 3) {
    for (let i = 0; i < maxRetries; i++) {
      try {
        if (!this.browser) await this.createBrowser();

        const page = await this.browser.newPage();
        
        // مصادقة البروكسي (إذا لزم الأمر)
        await page.authenticate({
          username: 'your_username',
          password: 'your_password'
        });

        await page.goto(url, { waitUntil: 'networkidle2', timeout: 30000 });

        // التحقق من الكابتشا
        const content = await page.content();
        if (content.includes('captcha')) {
          console.log('تم اكتشاف كابتشا، نبدل البروكسي...');
          this.rotate();
          await this.createBrowser();
          continue;
        }

        return content;

      } catch (error) {
        console.log(`خطأ: ${error.message}, محاولة ${i+1}`);
        this.rotate();
        await this.createBrowser();
      }
    }
    throw new Error('فشل تحميل الصفحة');
  }
}

// الاستخدام:
const proxies = ['gate1.com:8000', 'gate2.com:8000'];
const scraper = new PuppeteerRotatingProxy(proxies);

(async () => {
  const html = await scraper.scrape('https://www.avito.ru/moskva');
  console.log(`حصلنا على HTML بطول: ${html.length}`);
  await scraper.browser.close();
})();

Playwright مع دعم الدوران

Playwright هو بديل حديث لـ Puppeteer مع أداء أفضل. إعداد البروكسي مماثل:

const { chromium } = require('playwright');

async function scrapeWithRotation(urls, proxyList) {
  let proxyIndex = 0;

  for (const url of urls) {
    const proxy = proxyList[proxyIndex];
    
    const browser = await chromium.launch({
      headless: true,
      proxy: {
        server: proxy,
        username: 'your_user',
        password: 'your_pass'
      }
    });

    const page = await browser.newPage();
    
    try {
      await page.goto(url, { timeout: 30000 });
      const title = await page.title();
      console.log(`${url} → ${title} (بروكسي: ${proxy})`);
    } catch (error) {
      console.log(`خطأ في ${url}: ${error.message}`);
    }

    await browser.close();
    
    // البروكسي التالي للـ URL التالي
    proxyIndex = (proxyIndex + 1) % proxyList.length;
  }
}

const urls = [
  'https://www.wildberries.ru',
  'https://www.ozon.ru',
  'https://www.avito.ru'
];

const proxies = [
  'http://gate1.com:8000',
  'http://gate2.com:8000'
];

scrapeWithRotation(urls, proxies);

التكامل مع متصفحات مكافحة الكشف: Dolphin Anty و AdsPower

بالنسبة للمسوقين بالعمولة ومتخصصي SMM الذين يعملون مع حسابات متعددة، فإن التعيين اليدوي للبروكسي لكل ملف شخصي في Dolphin Anty أو AdsPower يستغرق ساعات. يتيح API لهذه المتصفحات أتمتة إنشاء الملفات الشخصية وربط البروكسيات.

أتمتة Dolphin Anty عبر API

يوفر Dolphin Anty واجهة API محلية (عادة على http://localhost:3001/v1.0)، والتي من خلالها يمكنك إنشاء ملفات شخصية، وتعيين بروكسيات، وتشغيل المتصفحات برمجياً.

مثال على سكريبت Python للإنشاء الجماعي للملفات الشخصية مع بروكسيات فريدة:

import requests
import json

DOLPHIN_API = "http://localhost:3001/v1.0"
API_TOKEN = "your_dolphin_api_token"

# قائمة البروكسيات من مزودك (تم الحصول عليها عبر API الخاص بهم)
proxies = [
    {"host": "gate1.com", "port": 8000, "login": "user1", "password": "pass1"},
    {"host": "gate2.com", "port": 8000, "login": "user2", "password": "pass2"},
]

def create_profile_with_proxy(name, proxy):
    """إنشاء ملف شخصي في Dolphin مع ربط البروكسي"""
    
    payload = {
        "name": name,
        "tags": ["Facebook Ads", "Auto-created"],
        "proxy": {
            "type": "http",  # أو socks5
            "host": proxy["host"],
            "port": proxy["port"],
            "login": proxy["login"],
            "password": proxy["password"]
        },
        "fingerprint": {
            "os": "win",
            "webRTC": {
                "mode": "altered",
                "fillBasedOnIp": True
            },
            "canvas": {
                "mode": "noise"
            }
        }
    }
    
    headers = {
        "Authorization": f"Bearer {API_TOKEN}",
        "Content-Type": "application/json"
    }
    
    response = requests.post(
        f"{DOLPHIN_API}/browser_profiles",
        headers=headers,
        data=json.dumps(payload)
    )
    
    if response.status_code == 200:
        profile = response.json()
        print(f"✓ تم إنشاء ملف شخصي: {name}, ID: {profile['id']}")
        return profile['id']
    else:
        print(f"✗ خطأ في إنشاء {name}: {response.text}")
        return None

# إنشاء 50 ملف شخصي مع دوران البروكسيات
for i in range(50):
    proxy = proxies[i % len(proxies)]  # دوران دوري
    profile_name = f"FB_Account_{i+1:03d}"
    create_profile_with_proxy(profile_name, proxy)

الآن لديك 50 ملف شخصي في Dolphin Anty، كل منها ببصمة متصفح فريدة وبروكسي. لتشغيل الملف الشخصي برمجياً:

def start_profile(profile_id):
    """تشغيل ملف شخصي للمتصفح"""
    response = requests.get(
        f"{DOLPHIN_API}/browser_profiles/{profile_id}/start",
        headers={"Authorization": f"Bearer {API_TOKEN}"}
    )
    
    if response.status_code == 200:
        data = response.json()
        print(f"تم تشغيل الملف الشخصي، منفذ WebDriver: {data['automation']['port']}")
        return data['automation']['port']
    else:
        print(f"خطأ في التشغيل: {response.text}")

# تشغيل الملف الشخصي والتحكم عبر Selenium
port = start_profile("profile_id_here")

from selenium import webdriver
driver = webdriver.Remote(
    command_executor=f'http://127.0.0.1:{port}',
    options=webdriver.ChromeOptions()
)
driver.get("https://facebook.com")

أتمتة AdsPower

يوفر AdsPower أيضاً API محلي. المنطق مماثل لـ Dolphin، ولكن مع نقاط نهاية مختلفة:

import requests

ADSPOWER_API = "http://local.adspower.net:50325/api/v1"

def create_adspower_profile(name, proxy):
    payload = {
        "name": name,
        "group_id": "0",  # معرف مجموعة الملفات الشخصية
        "domain_name": "facebook.com",
        "open_urls": ["https://facebook.com"],
        "repeat_config": ["0"],
        "username": proxy["login"],
        "password": proxy["password"],
        "proxy_type": "http",
        "proxy_host": proxy["host"],
        "proxy_port": proxy["port"],
        "proxy_user": proxy["login"],
        "proxy_password": proxy["password"]
    }
    
    response = requests.post(
        f"{ADSPOWER_API}/user/create",
        json=payload
    )
    
    if response.json()["code"] == 0:
        user_id = response.json()["data"]["id"]
        print(f"✓ تم إنشاء ملف شخصي AdsPower: {name}, ID: {user_id}")
        return user_id
    else:
        print(f"✗ خطأ: {response.json()['msg']}")

# إنشاء الملفات الشخصية
for i, proxy in enumerate(proxies):
    create_adspower_profile(f"TikTok_Account_{i+1}", proxy)

هذه الأتمتة حاسمة عند العمل مع عشرات الحسابات. بدلاً من النسخ اليدوي لبيانات البروكسي في كل ملف شخصي، تقوم بتشغيل سكريبت وتحصل على بنية تحتية جاهزة في دقائق.

معالجة الأخطاء والاحتياطي التلقائي

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

أنواع الأخطاء واستراتيجيات المعالجة

تنفيذ نظام إعادة محاولة ذكي

بدلاً من إعادة الطلب البسيطة، سننشئ نظاماً بتأخير أسي وقائمة سوداء للبروكسيات غير العاملة:

import requests
import time
from collections import defaultdict

class SmartProxyRotator:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
        self.blacklist = set()  # IP التي لا تعمل
        self.error_count = defaultdict(int)  # عداد الأخطاء حسب IP
        self.max_errors = 3  # بعد 3 أخطاء - إلى القائمة السوداء
    
    def get_working_proxy(self):
        """الحصول على بروكسي ليس في القائمة السوداء"""
        available = [p for p in self.proxy_list if p not in self.blacklist]
        if not available:
            # جميع البروكسيات محظورة - مسح القائمة السوداء
            print("⚠ جميع البروكسيات محظورة، نعيد تعيين القائمة السوداء")
            self.blacklist.clear()
            self.error_count.clear()
            available = self.proxy_list
        return available[0]
    
    def mark_error(self, proxy):
        """تسجيل خطأ البروكسي"""
        self.error_count[proxy] += 1
        if self.error_count[proxy] >= self.max_errors:
            self.blacklist.add(proxy)
            print(f"✗ تمت إضافة البروكسي {proxy} إلى القائمة السوداء")
    
    def request_with_retry(self, url, max_retries=5):
        """طلب مع إعادة محاولات ذكية"""
        for attempt in range(max_retries):
            proxy = self.get_working_proxy()
            
            try:
                # تأخير أسي: 1s, 2s, 4s, 8s...
                if attempt > 0:
                    delay = 2 ** attempt
                    print(f"انتظار {delay}ث قبل المحاولة {attempt+1}")
                    time.sleep(delay)
                
                response = requests.get(
                    url,
                    proxies={"http": proxy, "https": proxy},
                    timeout=15
                )
                
                # نجاح - إعادة تعيين عداد الأخطاء
                if response.status_code == 200:
                    self.error_count[proxy] = 0
                    return response
                
                # حظر - تبديل البروكسي
                elif response.status_code in [403, 429]:
                    print(f"الحالة {response.status_code}, نبدل البروكسي")
                    self.mark_error(proxy)
                    continue
                
            except requests.exceptions.ProxyError:
                print(f"ProxyError مع {proxy}")
                self.mark_error(proxy)
                
            except requests.exceptions.Timeout:
                print(f"Timeout مع {proxy}")
                self.mark_error(proxy)
        
        raise Exception(f"فشل تنفيذ الطلب بعد {max_retries} محاولات")

# الاستخدام:
proxies = [
    "http://user:pass@gate1.com:8000",
    "http://user:pass@gate2.com:8000",
    "http://user:pass@gate3.com:8000",
]

rotator = SmartProxyRotator(proxies)

for i in range(100):
    try:
        response = rotator.request_with_retry(f"https://api.example.com/data?page={i}")
        print(f"✓ الصفحة {i}: {len(response.text)} بايت")
    except Exception as e:
        print(f"✗ خطأ حرج في الصفحة {i}: {e}")

المراقبة والتنبيهات

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

import logging
from datetime import datetime

class ProxyMonitor:
    def __init__(self):
        self.stats = {
            "total_requests": 0,
            "successful": 0,
            "failed": 0,
            "proxy_errors": defaultdict(int),
            "start_time": datetime.now()
        }
        
        # إعداد التسجيل
        logging.basicConfig(
            filename='proxy_rotation.log',
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
    
    def log_request(self, proxy, success, error=None):
        self.stats["total_requests"] += 1
        
        if success:
            self.stats["successful"] += 1
            logging.info(f"✓ نجاح مع {proxy}")
        else:
            self.stats["failed"] += 1
            self.stats["proxy_errors"][proxy] += 1
            logging.error(f"✗ خطأ مع {proxy}: {error}")
    
    def get_report(self):
        uptime = datetime.now() - self.stats["start_time"]
        success_rate = (self.stats["successful"] / self.stats["total_requests"] * 100) if self.stats["total_requests"] > 0 else 0
        
        return f"""
=== تقرير دوران البروكسي ===
وقت التشغيل: {uptime}
إجمالي الطلبات: {self.stats["total_requests"]}
ناجحة: {self.stats["successful"]} ({success_rate:.1f}%)
أخطاء: {self.stats["failed"]}

بروكسيات مشكلة:
{self._format_errors()}
        """
    
    def _format_errors(self):
        sorted_errors = sorted(
            self.stats["proxy_errors"].items(),
            key=lambda x: x[1],
            reverse=True
        )
        return "\n".join([f"  {proxy}: {count} أخطاء" for proxy, count in sorted_errors[:5]])

# التكامل مع rotator
monitor = ProxyMonitor()

# في حلقة الطلبات:
try:
    response = rotator.request_with_retry(url)
    monitor.log_request(current_proxy, success=True)
except Exception as e:
    monitor.log_request(current_proxy, success=False, error=str(e))

أفضل الممارسات وتحسين استهلاك البيانات

لتحقيق أقصى استفادة من أتمتة دوران البروكسي، اتبع هذه التوصيات:

• استخدم الجلسات الثابتة عندما يكون ذلك ممكناً: لتقليل استهلاك البيانات، حافظ على نفس IP لعدة طلبات إذا كان الموقع المستهدف يسمح بذلك
• قم بتنفيذ التخزين المؤقت: احفظ الاستجابات المتكررة محلياً لتجنب الطلبات غير الضرورية
• راقب معدل النجاح: إذا انخفض إلى أقل من 70%، قد تحتاج إلى تغيير مزود البروكسي أو تعديل الاستراتيجية
• استخدم تأخيرات عشوائية: بين الطلبات لتقليد السلوك البشري وتجنب الكشف
• قسّم المهام حسب الجغرافيا: استخدم بروكسيات من نفس البلد الذي يستهدفه الموقع لتحسين السرعة والموثوقية

الخلاصة

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

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