دمج البروكسي مع وظائف جوجل السحابية: إعداد للتجريف والأتمتة

Google Cloud Functions هي منصة بدون خادم لتشغيل التعليمات البرمجية دون إدارة الخوادم. عند العمل مع جمع البيانات، وأتمتة طلبات API، أو جمع البيانات، غالبًا ما يتطلب توجيه حركة المرور عبر البروكسي لتجاوز الحظر، وتدوير IP، واستهداف جغرافي. في هذا الدليل، سنستعرض إعداد البروكسي في Cloud Functions باستخدام Python وNode.js مع أمثلة عملية.

لماذا استخدام البروكسي في Cloud Functions

تعمل Google Cloud Functions في بيئة معزولة مع عناوين IP مشتركة لمراكز بيانات Google. عند إجراء طلبات متكررة إلى واجهات برمجة التطبيقات الخارجية أو المواقع الإلكترونية، تظهر مشكلات:

• حظر IP — العديد من الخدمات (Google، Facebook، الأسواق) تتعرف على حركة المرور من مراكز البيانات وتطبق قيود السرعة أو الحظر الكامل.
• قيود جغرافية — للوصول إلى المحتوى المتاح فقط في دول معينة (على سبيل المثال، جمع أسعار الإقليمية على Wildberries أو Ozon).
• قيود السرعة — يمكن أن يقوم عنوان IP واحد بإجراء عدد محدود من الطلبات في الدقيقة. تسمح البروكسي بتوزيع الحمل.
• الخصوصية — إخفاء المصدر الحقيقي للطلبات عند العمل مع البيانات الحساسة أو التجسس على المنافسين.

السيناريوهات الشائعة لاستخدام البروكسي في Cloud Functions:

• جمع البيانات من الأسواق (Wildberries، Ozon، Amazon) لمراقبة أسعار المنافسين
• جمع البيانات من وسائل التواصل الاجتماعي (Instagram، TikTok) عبر واجهة برمجة التطبيقات أو جمع البيانات من الويب
• أتمتة فحص الإعلانات في مناطق مختلفة
• طلبات جماعية إلى محركات البحث (Google، Yandex) لتحليل SEO
• اختبار وظائف تحديد المواقع الجغرافية للتطبيقات

ما هي أنواع البروكسي المناسبة لـ Cloud Functions

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

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

• لجمع البيانات من الأسواق (Wildberries، Ozon، Amazon) — بروكسي سكنية مع تدوير حسب الطلب، بحيث يتم إرسال كل طلب من IP جديد.
• لطلبات API (Google Maps API، OpenWeatherMap) — بروكسي مركز البيانات بسرعة عالية، إذا لم تكن هناك قيود صارمة على IP.
• لوسائل التواصل الاجتماعي (Instagram، TikTok) — بروكسي موبايل، لأنها تحتوي على IP لمشغلي الهواتف المحمولة ونادرًا ما يتم حظرها.
• لجمع البيانات من SEO (Google، Yandex) — بروكسي سكنية مرتبطة جغرافيًا بالمنطقة المطلوبة.

إعداد البروكسي في Python (requests، aiohttp)

Python هو أكثر اللغات شعبية في Cloud Functions عند العمل مع جمع البيانات وأتمتة. دعونا نستعرض دمج البروكسي مع المكتبات requests (الطلبات المتزامنة) وaiohttp (الطلبات غير المتزامنة).

مثال مع مكتبة requests (بروكسي HTTP)

import requests
import os

def parse_with_proxy(request):
    # الحصول على بيانات البروكسي من متغيرات البيئة
    proxy_host = os.environ.get('PROXY_HOST', 'proxy.example.com')
    proxy_port = os.environ.get('PROXY_PORT', '8080')
    proxy_user = os.environ.get('PROXY_USER', 'username')
    proxy_pass = os.environ.get('PROXY_PASS', 'password')
    
    # تشكيل عنوان URL للبروكسي مع المصادقة
    proxy_url = f"http://{proxy_user}:{proxy_pass}@{proxy_host}:{proxy_port}"
    
    proxies = {
        'http': proxy_url,
        'https': proxy_url
    }
    
    try:
        # إجراء الطلب عبر البروكسي مع مهلة
        response = requests.get(
            'https://api.example.com/data',
            proxies=proxies,
            timeout=10,
            headers={'User-Agent': 'Mozilla/5.0'}
        )
        
        # التحقق من حالة الاستجابة
        response.raise_for_status()
        
        return {
            'statusCode': 200,
            'body': response.json(),
            'ip_used': response.headers.get('X-Forwarded-For', 'unknown')
        }
        
    except requests.exceptions.ProxyError as e:
        return {'statusCode': 502, 'error': f'خطأ في البروكسي: {str(e)}'}
    except requests.exceptions.Timeout:
        return {'statusCode': 504, 'error': 'مهلة الطلب'}
    except requests.exceptions.RequestException as e:
        return {'statusCode': 500, 'error': f'فشل الطلب: {str(e)}'}

نقاط مهمة:

• متغيرات البيئة — احتفظ ببيانات البروكسي (المضيف، المنفذ، اسم المستخدم، كلمة المرور) في Secret Manager أو متغيرات البيئة في Cloud Functions، وليس في الكود.
• المهل الزمنية — تأكد من تعيين timeout، حتى لا تتعطل الوظيفة عند حدوث مشكلات مع البروكسي.
• User-Agent — أضف رأس User-Agent، حتى تبدو الطلبات وكأنها من متصفح حقيقي.
• معالجة الأخطاء — تعامل بشكل منفصل مع ProxyError (مشكلات البروكسي) وTimeout (بروكسي بطيء).

مثال مع aiohttp (طلبات غير متزامنة)

للمهام ذات الحمل العالي (مثل جمع بيانات أكثر من 1000 صفحة)، استخدم الطلبات غير المتزامنة مع aiohttp:

import aiohttp
import asyncio
import os

async def fetch_with_proxy(url, proxy_url):
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get(
                url,
                proxy=proxy_url,
                timeout=aiohttp.ClientTimeout(total=10),
                headers={'User-Agent': 'Mozilla/5.0'}
            ) as response:
                return await response.text()
        except aiohttp.ClientProxyConnectionError:
            return {'error': 'فشل الاتصال بالبروكسي'}
        except asyncio.TimeoutError:
            return {'error': 'مهلة الطلب'}

def parse_multiple_urls(request):
    proxy_url = f"http://{os.environ['PROXY_USER']}:{os.environ['PROXY_PASS']}@{os.environ['PROXY_HOST']}:{os.environ['PROXY_PORT']}"
    
    urls = [
        'https://example.com/page1',
        'https://example.com/page2',
        'https://example.com/page3'
    ]
    
    # تشغيل الطلبات غير المتزامنة بالتوازي
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    tasks = [fetch_with_proxy(url, proxy_url) for url in urls]
    results = loop.run_until_complete(asyncio.gather(*tasks))
    
    return {'statusCode': 200, 'results': results}

يسمح النهج غير المتزامن بإجراء 10-100 طلبات متوازية عبر البروكسي، وهو أمر حاسم لجمع كميات كبيرة من البيانات ضمن الوقت المحدد لتنفيذ Cloud Functions (حتى 9 دقائق).

العمل مع بروكسي SOCKS5

بعض مزودي البروكسي يقدمون بروكسي SOCKS5 للعمل بشكل أكثر موثوقية مع حركة مرور UDP أو لتجاوز الحظر. لاستخدام SOCKS5 في Python، استخدم مكتبة requests[socks]:

# أضف إلى requirements.txt:
# requests[socks]

import requests

def use_socks5_proxy(request):
    proxy_url = f"socks5://{os.environ['PROXY_USER']}:{os.environ['PROXY_PASS']}@{os.environ['PROXY_HOST']}:{os.environ['PROXY_PORT']}"
    
    proxies = {
        'http': proxy_url,
        'https': proxy_url
    }
    
    response = requests.get(
        'https://api.ipify.org?format=json',
        proxies=proxies,
        timeout=10
    )
    
    return {'statusCode': 200, 'ip': response.json()}

إعداد البروكسي في Node.js (axios، node-fetch)

Node.js هي اللغة الثانية الأكثر شعبية في Cloud Functions. دعونا نستعرض دمج البروكسي مع المكتبات axios وnode-fetch.

مثال مع axios

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

exports.parseWithProxy = async (req, res) => {
  const proxyUrl = `http://${process.env.PROXY_USER}:${process.env.PROXY_PASS}@${process.env.PROXY_HOST}:${process.env.PROXY_PORT}`;
  
  const agent = new HttpsProxyAgent(proxyUrl);
  
  try {
    const response = await axios.get('https://api.example.com/data', {
      httpsAgent: agent,
      timeout: 10000,
      headers: {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
      }
    });
    
    res.status(200).json({
      success: true,
      data: response.data,
      proxyUsed: proxyUrl.split('@')[1] // إرجاع المضيف:المنفذ بدون كلمة المرور
    });
    
  } catch (error) {
    if (error.code === 'ECONNREFUSED') {
      res.status(502).json({ error: 'تم رفض اتصال البروكسي' });
    } else if (error.code === 'ETIMEDOUT') {
      res.status(504).json({ error: 'مهلة البروكسي' });
    } else {
      res.status(500).json({ error: error.message });
    }
  }
};

التبعيات لـ package.json:

{
  "dependencies": {
    "axios": "^1.6.0",
    "https-proxy-agent": "^7.0.2"
  }
}

مثال مع node-fetch و SOCKS5

const fetch = require('node-fetch');
const { SocksProxyAgent } = require('socks-proxy-agent');

exports.fetchWithSocks5 = async (req, res) => {
  const proxyUrl = `socks5://${process.env.PROXY_USER}:${process.env.PROXY_PASS}@${process.env.PROXY_HOST}:${process.env.PROXY_PORT}`;
  
  const agent = new SocksProxyAgent(proxyUrl);
  
  try {
    const response = await fetch('https://api.ipify.org?format=json', {
      agent,
      timeout: 10000
    });
    
    const data = await response.json();
    
    res.status(200).json({
      success: true,
      yourIP: data.ip
    });
    
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
};

التبعيات لـ SOCKS5:

{
  "dependencies": {
    "node-fetch": "^2.7.0",
    "socks-proxy-agent": "^8.0.2"
  }
}

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

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

1. المصادقة باستخدام اسم المستخدم وكلمة المرور

الطريقة الأكثر شيوعًا هي تمرير بيانات الاعتماد في عنوان URL للبروكسي:

http://username:password@proxy.example.com:8080

المزايا: سهولة الإعداد، لا تتطلب IP ثابت للمصدر.

العيوب: يتم تمرير بيانات الاعتماد في كل طلب، مما يضيف بعض الحمل.

2. المصادقة باستخدام قائمة IP البيضاء

بعض المزودين يسمحون بإضافة عناوين IP لـ Cloud Functions إلى قائمة IP البيضاء. المشكلة: تستخدم Cloud Functions عناوين IP ديناميكية من مجموعة Google Cloud.

الحل: استخدم Cloud NAT لتوجيه حركة المرور الصادرة عبر IP خارجي ثابت:

1. قم بإنشاء شبكة VPC و subnet في Google Cloud
2. قم بإعداد Cloud NAT مع حجز IP ثابت
3. قم بتوصيل Cloud Functions بـ VPC Connector
4. أضف IP الثابت إلى قائمة IP البيضاء لمزود البروكسي

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

proxies = {
    'http': 'http://proxy.example.com:8080',
    'https': 'http://proxy.example.com:8080'
}

التوصية: في معظم الحالات، استخدم المصادقة باستخدام اسم المستخدم/كلمة المرور — فهي أسهل ولا تتطلب تكاليف إضافية على Cloud NAT (من 0.044 دولار/ساعة + حركة المرور).

تدوير IP وإدارة مجموعة البروكسي

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

1. التدوير من جانب المزود (بروكسي متغيرة)

العديد من المزودين يقدمون بروكسي متغيرة — نقطة نهاية واحدة، تتغير IP تلقائيًا مع كل طلب أو حسب المؤقت:

# نقطة نهاية واحدة، تتغير IP تلقائيًا
proxy_url = "http://username:password@rotating.proxy.com:8080"

# كل طلب يأتي من IP جديد
for i in range(100):
    response = requests.get('https://api.ipify.org', proxies={'http': proxy_url})
    print(f"طلب {i}: IP = {response.text}")

المزايا: لا حاجة لإدارة مجموعة البروكسي يدويًا، تكامل بسيط.

العيوب: لا يوجد تحكم في IP معين، قد يكون أغلى.

2. إدارة مجموعة البروكسي يدويًا

إذا كان لديك قائمة بالبروكسي الثابتة، نفذ التدوير على مستوى الكود:

import random
import requests

# مجموعة البروكسي (يمكن تحميلها من Secret Manager)
PROXY_POOL = [
    "http://user:pass@proxy1.example.com:8080",
    "http://user:pass@proxy2.example.com:8080",
    "http://user:pass@proxy3.example.com:8080",
]

def get_random_proxy():
    return random.choice(PROXY_POOL)

def parse_with_rotation(urls):
    results = []
    
    for url in urls:
        proxy = get_random_proxy()
        
        try:
            response = requests.get(
                url,
                proxies={'http': proxy, 'https': proxy},
                timeout=10
            )
            results.append({
                'url': url,
                'status': response.status_code,
                'proxy': proxy.split('@')[1]
            })
        except Exception as e:
            # إذا لم يعمل البروكسي، جرب آخر
            proxy = get_random_proxy()
            response = requests.get(url, proxies={'http': proxy, 'https': proxy})
            results.append({'url': url, 'status': response.status_code})
    
    return results

3. بروكسي قائم على الجلسة (جلسات ثابتة)

للمهام التي تحتاج إلى الحفاظ على IP واحد ضمن الجلسة (مثل تسجيل الدخول إلى موقع)، استخدم معرف الجلسة في عنوان URL للبروكسي:

# إضافة معرف الجلسة إلى اسم المستخدم
import uuid

session_id = str(uuid.uuid4())
proxy_url = f"http://username-session-{session_id}:password@proxy.example.com:8080"

# جميع الطلبات مع هذا session_id ستذهب عبر نفس IP
session = requests.Session()
session.proxies = {'http': proxy_url, 'https': proxy_url}

# تسجيل الدخول
session.post('https://example.com/login', data={'user': 'test', 'pass': '123'})

# الطلبات التالية في نفس الجلسة
session.get('https://example.com/dashboard')

معالجة الأخطاء والمهل الزمنية

عند العمل مع البروكسي في Cloud Functions، من الضروري معالجة الأخطاء بشكل صحيح لتجنب فقدان البيانات وعدم تجاوز حدود وقت التنفيذ.

أنواع الأخطاء وطرق المعالجة

مثال على معالجة الأخطاء الشاملة

import requests
import time
from requests.exceptions import ProxyError, Timeout, RequestException

def fetch_with_retry(url, proxy_pool, max_retries=3):
    """
    طلب مع إعادة المحاولة تلقائيًا وتغيير البروكسي عند حدوث أخطاء
    """
    for attempt in range(max_retries):
        proxy = random.choice(proxy_pool)
        
        try:
            response = requests.get(
                url,
                proxies={'http': proxy, 'https': proxy},
                timeout=10,
                headers={'User-Agent': 'Mozilla/5.0'}
            )
            
            # التحقق من رمز الحالة
            if response.status_code == 200:
                return {'success': True, 'data': response.text, 'proxy': proxy}
            elif response.status_code == 429:
                # قيود السرعة — انتظر وحاول مرة أخرى
                time.sleep(2 ** attempt)  # زيادة زمن الانتظار
                continue
            elif response.status_code == 403:
                # تم حظر IP — تغيير البروكسي
                continue
            else:
                return {'success': False, 'status': response.status_code}
                
        except ProxyError:
            # البروكسي لا يعمل — جرب التالي
            print(f"البروكسي {proxy} فشل، جرب آخر...")
            continue
        except Timeout:
            # مهلة — جرب مع بروكسي آخر
            print(f"مهلة مع {proxy}، إعادة المحاولة...")
            continue
        except RequestException as e:
            # أخطاء أخرى
            print(f"فشل الطلب: {e}")
            if attempt == max_retries - 1:
                return {'success': False, 'error': str(e)}
            continue
    
    return {'success': False, 'error': 'تم تجاوز الحد الأقصى لعدد المحاولات'}

تعيين المهل الزمنية في Cloud Functions

تحتوي Cloud Functions على حد زمني للتنفيذ (بشكل افتراضي 60 ثانية، والحد الأقصى 540 ثانية). ضع ذلك في الاعتبار عند تعيين المهل الزمنية للبروكسي:

• مهلة الاتصال — الوقت اللازم لإنشاء اتصال مع البروكسي (يوصى بـ 5 ثوانٍ)
• مهلة القراءة — الوقت اللازم للحصول على استجابة من الخادم المستهدف عبر البروكسي (يوصى بـ 10-15 ثانية)
• المهلة الإجمالية — الوقت الإجمالي لكل الطلب (يجب أن يكون أقل من مهلة الوظيفة)

# Python: مهلات منفصلة
response = requests.get(
    url,
    proxies=proxies,
    timeout=(5, 15)  # (مهلة الاتصال، مهلة القراءة)
)

# Node.js مع axios
const response = await axios.get(url, {
  httpsAgent: agent,
  timeout: 10000  // المهلة الإجمالية بالمللي ثانية
});

أفضل الممارسات وتحسين الأداء

توصيات للعمل بشكل فعال مع البروكسي في Cloud Functions:

1. استخدم متغيرات البيئة لبيانات الاعتماد

لا تخزن أبدًا أسماء المستخدمين وكلمات المرور للبروكسي في الكود. استخدم Secret Manager أو متغيرات البيئة:

# إنشاء سر في Google Cloud
gcloud secrets create proxy-credentials \
    --data-file=proxy-config.json

# منح الوصول إلى Cloud Functions
gcloud secrets add-iam-policy-binding proxy-credentials \
    --member=serviceAccount:PROJECT_ID@appspot.gserviceaccount.com \
    --role=roles/secretmanager.secretAccessor

# قراءة السر في الكود
from google.cloud import secretmanager
import json

def get_proxy_config():
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/{PROJECT_ID}/secrets/proxy-credentials/versions/latest"
    response = client.access_secret_version(request={"name": name})
    return json.loads(response.payload.data.decode('UTF-8'))

2. قم بتخزين نتائج الجمع في ذاكرة التخزين المؤقت

استخدم Cloud Storage أو Firestore لتخزين البيانات في الذاكرة المؤقتة، حتى لا تضطر إلى إجراء طلبات متكررة عبر البروكسي:

import hashlib
from google.cloud import storage

def fetch_with_cache(url, proxy):
    # إنشاء مفتاح الذاكرة المؤقتة بناءً على URL
    cache_key = hashlib.md5(url.encode()).hexdigest()
    
    # التحقق من الذاكرة المؤقتة في Cloud Storage
    bucket = storage.Client().bucket('my-cache-bucket')
    blob = bucket.blob(f"cache/{cache_key}.json")
    
    if blob.exists():
        # إرجاع البيانات المخزنة في الذاكرة المؤقتة
        return json.loads(blob.download_as_text())
    
    # إجراء الطلب عبر البروكسي
    response = requests.get(url, proxies={'http': proxy})
    data = response.json()
    
    # حفظ في الذاكرة المؤقتة
    blob.upload_from_string(json.dumps(data))
    
    return data

3. المراقبة والتسجيل

تتبع أداء البروكسي وتكرار الأخطاء عبر Cloud Logging:

import logging
import time

def fetch_with_logging(url, proxy):
    start_time = time.time()
    
    try:
        response = requests.get(url, proxies={'http': proxy}, timeout=10)
        duration = time.time() - start_time
        
        logging.info({
            'url': url,
            'proxy': proxy.split('@')[1],
            'status': response.status_code,
            'duration': duration,
            'success': True
        })
        
        return response
        
    except Exception as e:
        duration = time.time() - start_time
        
        logging.error({
            'url': url,
            'proxy': proxy.split('@')[1],
            'error': str(e),
            'duration': duration,
            'success': False
        })
        
        raise

4. تحسين بدء التشغيل البارد

تحتوي Cloud Functions على تأخير عند بدء التشغيل البارد. قلل من التبعيات واستخدم الإصدارات الدنيا من المكتبات:

# requirements.txt — فقط المكتبات الضرورية
requests==2.31.0
# تجنب المكتبات الثقيلة مثل pandas، إذا لم تكن ضرورية

استخدم المتغيرات العالمية لإعادة استخدام الاتصالات:

# إنشاء جلسة مرة واحدة عند بدء التشغيل البارد
session = requests.Session()
session.proxies = {'http': PROXY_URL, 'https': PROXY_URL}

def parse_data(request):
    # إعادة استخدام الجلسة بين الاستدعاءات
    response = session.get('https://api.example.com/data')
    return response.json()

5. الربط الجغرافي للبروكسي

للمهام ذات الاستهداف الجغرافي (مثل جمع أسعار الإقليمية)، استخدم بروكسي مرتبط بدولة أو مدينة معينة:

# مثال مع بروكسي سكنية، حيث يمكنك تحديد الدولة في اسم المستخدم
proxy_url = f"http://username-country-ru:password@proxy.example.com:8080"

# أو استخدام نقاط نهاية مختلفة لدول مختلفة
PROXIES_BY_COUNTRY = {
    'RU': 'http://user:pass@ru.proxy.example.com:8080',
    'US': 'http://user:pass@us.proxy.example.com:8080',
    'DE': 'http://user:pass@de.proxy.example.com:8080'
}

def parse_by_country(country_code):
    proxy = PROXIES_BY_COUNTRY.get(country_code)
    response = requests.get('https://example.com', proxies={'http': proxy})
    return response.text

الخاتمة

يفتح دمج البروكسي مع Google Cloud Functions آفاقًا واسعة لجمع البيانات، وأتمتة العمل مع واجهات برمجة التطبيقات دون قيود على IP. النقاط الرئيسية التي يجب مراعاتها: معالجة الأخطاء بشكل صحيح مع منطق إعادة المحاولة، استخدام المهل الزمنية لمنع التعليق، تدوير IP لتجنب الحظر، وتخزين بيانات الاعتماد بشكل آمن في Secret Manager.

بالنسبة لمعظم مهام الجمع والأتمتة، سيكون الخيار الأمثل هو البروكسي السكنية — حيث توفر خصوصية عالية ونسبة منخفضة من الحظر بفضل استخدام IP لمستخدمين حقيقيين. للعمل مع وسائل التواصل الاجتماعي والتطبيقات المحمولة، نوصي بـ البروكسي الموبايل، التي تحتوي على IP لمشغلي الهواتف المحمولة ونادرًا ما يتم حظرها من قبل منصات مثل Instagram وTikTok.

مع الإعداد الصحيح لـ Cloud Functions مع البروكسي، ستحصل على حل قابل للتوسع وفعال من حيث التكلفة لمعالجة كميات كبيرة من البيانات دون الحاجة إلى إدارة البنية التحتية.