AWS Lambda هي منصة بدون خادم تتيح لك تشغيل الكود دون إدارة الخوادم. ومع ذلك، عند العمل على جمع بيانات المواقع، أو واجهات برمجة التطبيقات للأسواق، أو أتمتة المهام، غالبًا ما تظهر مشكلة: تستخدم وظائف Lambda عناوين IP من AWS، والتي يمكن اكتشافها بسهولة وحظرها. في هذا الدليل، سنستعرض كيفية دمج البروكسي في Lambda، وإعداد تدوير IP، وتجنب الأخطاء الشائعة.
المقال موجه للمطورين الذين يقومون بأتمتة المهام عبر AWS Lambda: جمع البيانات من المواقع المحمية، ومراقبة أسعار المنافسين، والعمل مع واجهات برمجة التطبيقات لوسائل التواصل الاجتماعي أو الأسواق. ستحصل على أمثلة جاهزة من الكود في Python وNode.js، والتي يمكنك استخدامها مباشرة بعد القراءة.
لماذا تحتاج البروكسي في AWS Lambda
تستخدم AWS Lambda بشكل افتراضي عناوين IP من مجموعة Amazon Web Services. هذه العناوين موجودة في قوائم عامة ويمكن تحديدها بسهولة بواسطة أنظمة الحماية من الروبوتات. إليك السيناريوهات الرئيسية التي تصبح فيها البروكسي ضرورية:
حالة حقيقية: قام مطور بإعداد Lambda لمراقبة أسعار Wildberries كل 15 دقيقة. بعد يومين، بدأ السوق في إرجاع خطأ 403 محظور - تم إدراج IP الخاص بـ AWS في القائمة السوداء. بعد توصيل البروكسي السكنية، يعمل جمع البيانات بشكل مستقر منذ 6 أشهر.
الأسباب الرئيسية لاستخدام البروكسي في Lambda:
- جمع البيانات من المواقع المحمية: العديد من المواقع تحظر الطلبات من عناوين IP لمراكز البيانات AWS. تسمح البروكسي بإخفاء Lambda كأنها مستخدمون عاديون.
- قيود الجغرافيا: إذا كنت بحاجة إلى الحصول على بيانات من موقع متاح فقط من بلد معين (على سبيل المثال، الأسعار الإقليمية على Ozon)، فإن البروكسي ذات الموقع الجغرافي المطلوب تحل المشكلة.
- تجاوز قيود المعدل: تحدد واجهات برمجة التطبيقات للعديد من الخدمات عدد الطلبات من IP واحد. يسمح تدوير البروكسي بتوزيع الحمل.
- اختبار A/B للإعلانات: التحقق من عرض الإعلانات من مناطق مختلفة لتحليل المنافسين.
- مراقبة الأسواق: تتبع مواقع المنتجات، وأسعار المنافسين على Wildberries، Ozon، Avito دون حظر.
غالبًا ما يتم تشغيل وظائف Lambda وفقًا لجدول زمني (عبر أحداث CloudWatch) أو المحفزات، مما يجعلها أداة مثالية للأتمتة. ومع ذلك، بدون البروكسي، تواجه هذه المهام حظرًا سريعًا من الموارد المستهدفة.
أي نوع من البروكسي تختار لـ Lambda
يعتمد اختيار نوع البروكسي على المهمة التي تقوم بها وظيفة Lambda الخاصة بك. دعنا نستعرض ثلاثة أنواع رئيسية وتطبيقاتها في بنية بدون خادم:
| نوع البروكسي | السرعة | الخصوصية | أفضل السيناريوهات لـ Lambda |
|---|---|---|---|
| بروكسي مراكز البيانات | عالية جدًا (50-200 مللي ثانية) | متوسطة | جمع البيانات من واجهات برمجة التطبيقات بدون حماية صارمة، فحوصات جماعية لتوافر المواقع، مراقبة SEO |
| بروكسي سكنية | متوسطة (300-800 مللي ثانية) | عالية جدًا | جمع البيانات من المواقع المحمية (الأسواق، وسائل التواصل الاجتماعي)، تجاوز Cloudflare، العمل مع واجهة برمجة تطبيقات Instagram/Facebook |
| بروكسي موبايل | متوسطة (400-1000 مللي ثانية) | قصوى | العمل مع واجهات برمجة تطبيقات الموبايل (TikTok، Instagram)، اختبار إعلانات الموبايل، تجاوز أقوى أنظمة الحماية |
توصيات للاختيار:
- لجمع البيانات من Wildberries، Ozon، Avito: استخدم بروكسي سكنية بموقع جغرافي روسي. هذه المنصات تحظر بنشاط عناوين IP لمراكز البيانات.
- لرصد واجهات برمجة التطبيقات بدون حماية صارمة: بروكسي مراكز البيانات كافية، فهي أرخص وأسرع.
- للاستخدام مع واجهات برمجة تطبيقات Instagram، Facebook، TikTok: فقط بروكسي موبايل أو سكنية - هذه المنصات تكشف وتحظر مراكز البيانات.
- لتجاوز Cloudflare، PerimeterX: بروكسي سكنية مع تدوير، ويفضل أن تكون مع جلسات لاصقة (حفظ IP لمدة 5-30 دقيقة).
مهم: وظائف Lambda لها حد زمني للتنفيذ (حد أقصى 15 دقيقة). عند استخدام بروكسي بطيئة (سكنية/موبايل)، ضع في اعتبارك التأخيرات - إذا استغرق الطلب عبر البروكسي ثانيتين، فستتمكن من إجراء حوالي 450 طلبًا كحد أقصى خلال 15 دقيقة.
إعداد البروكسي في Lambda على Python (requests، urllib3)
Python هو أكثر اللغات شعبية لوظائف Lambda، خاصة لمهام جمع البيانات والأتمتة. دعنا نستعرض إعداد البروكسي باستخدام مكتبة requests، التي تُستخدم في 90% من الحالات.
الإعداد الأساسي لبروكسي HTTP
أبسط طريقة لتوصيل البروكسي هي تمرير المعامل proxies إلى طريقة requests.get():
import requests
import os
def lambda_handler(event, context):
# الحصول على بيانات اعتماد البروكسي من متغيرات البيئة
proxy_host = os.environ['PROXY_HOST'] # على سبيل المثال: proxy.example.com
proxy_port = os.environ['PROXY_PORT'] # على سبيل المثال: 8080
proxy_user = os.environ['PROXY_USER']
proxy_pass = os.environ['PROXY_PASS']
# تشكيل 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 # مهم! قم بتعيين مهلة
)
return {
'statusCode': 200,
'body': response.text
}
except requests.exceptions.ProxyError as e:
print(f"خطأ البروكسي: {e}")
return {
'statusCode': 500,
'body': 'فشل الاتصال بالبروكسي'
}
except requests.exceptions.Timeout as e:
print(f"خطأ المهلة: {e}")
return {
'statusCode': 504,
'body': 'مهلة الطلب'
}
النقاط الرئيسية في هذا الكود:
- متغيرات البيئة: لا تخزن بيانات اعتماد البروكسي مباشرة في الكود أبدًا! استخدم متغيرات البيئة في إعدادات Lambda.
- المهلة: تأكد من تعيين مهلة (10-30 ثانية). بدونها، قد تتعطل Lambda حتى انتهاء الحد الأقصى لوقت التنفيذ.
- معالجة الأخطاء: قد تكون البروكسي غير متاحة أو بطيئة - دائمًا قم بمعالجة الاستثناءات
ProxyErrorوTimeout. - HTTP وHTTPS: حدد كلا البروتوكولين في القاموس
proxies، حتى لو كنت تستخدم HTTPS فقط.
إعداد بروكسي SOCKS5
توفر بروكسي SOCKS5 مستوى أعلى من الخصوصية وتعمل على مستوى TCP، مما يجعلها غير مرئية لبعض أنظمة الحماية. للعمل مع SOCKS5 في requests، تحتاج إلى مكتبة requests[socks]:
import requests
import os
def lambda_handler(event, context):
proxy_host = os.environ['PROXY_HOST']
proxy_port = os.environ['PROXY_PORT']
proxy_user = os.environ['PROXY_USER']
proxy_pass = os.environ['PROXY_PASS']
# بروكسي SOCKS5 مع التفويض
proxy_url = f"socks5://{proxy_user}:{proxy_pass}@{proxy_host}:{proxy_port}"
proxies = {
'http': proxy_url,
'https': proxy_url
}
try:
response = requests.get(
'https://www.wildberries.ru/catalog/12345/detail.aspx',
proxies=proxies,
timeout=15
)
# جمع البيانات
return {
'statusCode': 200,
'body': response.text
}
except Exception as e:
print(f"خطأ: {e}")
return {
'statusCode': 500,
'body': str(e)
}
مهم للنشر في Lambda: عند استخدام SOCKS5، أضف إلى requirements.txt:
requests[socks]
PySocks
التحقق من IP عبر البروكسي
قبل تشغيل المنطق الرئيسي، من المفيد التحقق من أن البروكسي يعمل ويعيد IP المطلوب:
def check_proxy_ip(proxies):
"""يتحقق من IP الذي تراه العالم الخارجي عبر البروكسي"""
try:
response = requests.get(
'https://api.ipify.org?format=json',
proxies=proxies,
timeout=10
)
ip_data = response.json()
print(f"IP الحالي عبر البروكسي: {ip_data['ip']}")
return ip_data['ip']
except Exception as e:
print(f"فشل التحقق من البروكسي: {e}")
return None
def lambda_handler(event, context):
# ... إعداد البروكسي ...
# تحقق من IP قبل العمل الرئيسي
current_ip = check_proxy_ip(proxies)
if not current_ip:
return {
'statusCode': 500,
'body': 'فشل التحقق من البروكسي'
}
# المنطق الرئيسي لجمع البيانات
# ...
إعداد البروكسي في Lambda على Node.js (axios، got)
Node.js هو ثاني أكثر اللغات شعبية لوظائف Lambda، خاصة عندما تكون هناك حاجة لأداء عالٍ عند العمل مع واجهات برمجة التطبيقات. دعنا نستعرض إعداد البروكسي باستخدام مكتبات axios وgot.
الإعداد باستخدام axios
Axios هي مكتبة HTTP الأكثر شعبية لـ Node.js. للعمل مع البروكسي، ستحتاج إلى حزمة إضافية https-proxy-agent:
const axios = require('axios');
const HttpsProxyAgent = require('https-proxy-agent');
exports.handler = async (event) => {
// الحصول على بيانات الاعتماد من متغيرات البيئة
const proxyHost = process.env.PROXY_HOST;
const proxyPort = process.env.PROXY_PORT;
const proxyUser = process.env.PROXY_USER;
const proxyPass = process.env.PROXY_PASS;
// تشكيل URL البروكسي
const proxyUrl = `http://${proxyUser}:${proxyPass}@${proxyHost}:${proxyPort}`;
// إنشاء وكيل للبروكسي
const agent = new HttpsProxyAgent(proxyUrl);
try {
const response = await axios.get('https://api.example.com/data', {
httpsAgent: agent,
timeout: 10000 // 10 ثوانٍ
});
return {
statusCode: 200,
body: JSON.stringify(response.data)
};
} catch (error) {
console.error('فشل الطلب:', error.message);
return {
statusCode: 500,
body: JSON.stringify({
error: error.message
})
};
}
};
تثبيت التبعيات: أضف إلى package.json:
{
"dependencies": {
"axios": "^1.6.0",
"https-proxy-agent": "^7.0.0"
}
}
إعداد SOCKS5 باستخدام axios
لاستخدام بروكسي SOCKS5، استخدم الحزمة socks-proxy-agent:
const axios = require('axios');
const { SocksProxyAgent } = require('socks-proxy-agent');
exports.handler = async (event) => {
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 axios.get('https://www.ozon.ru/api/products', {
httpAgent: agent,
httpsAgent: agent,
timeout: 15000
});
return {
statusCode: 200,
body: JSON.stringify(response.data)
};
} catch (error) {
console.error('خطأ:', error.message);
return {
statusCode: 500,
body: JSON.stringify({ error: error.message })
};
}
};
بديل: مكتبة got
Got هي مكتبة HTTP حديثة تدعم البروكسي بشكل أصلي (لا تتطلب وكلاء منفصلين):
const got = require('got');
exports.handler = async (event) => {
const proxyUrl = `http://${process.env.PROXY_USER}:${process.env.PROXY_PASS}@${process.env.PROXY_HOST}:${process.env.PROXY_PORT}`;
try {
const response = await got('https://api.example.com/data', {
agent: {
http: new (require('http-proxy-agent'))(proxyUrl),
https: new (require('https-proxy-agent'))(proxyUrl)
},
timeout: {
request: 10000
},
responseType: 'json'
});
return {
statusCode: 200,
body: JSON.stringify(response.body)
};
} catch (error) {
console.error('خطأ:', error.message);
return {
statusCode: 500,
body: JSON.stringify({ error: error.message })
};
}
};
تدوير البروكسي في Lambda: كيفية تغيير IP تلقائيًا
يعتبر تدوير البروكسي أمرًا حيويًا للمهام التي تحتاج إلى إجراء العديد من الطلبات دون حظر. هناك نهجان رئيسيان: استخدام خدمات البروكسي مع تدوير تلقائي أو إدارة مجموعة البروكسي يدويًا.
التدوير التلقائي عبر المزود
تقدم معظم مزودي البروكسي السكنية (بما في ذلك ProxyCove) نقطة نهاية مع تدوير تلقائي - يتم تغيير IP تلقائيًا مع كل طلب أو كل N دقيقة:
import requests
import os
def lambda_handler(event, context):
# بروكسي مع تدوير تلقائي
# التنسيق: rotating.proxy.com:port
# كل طلب = IP جديد
proxy_url = f"http://{os.environ['PROXY_USER']}:{os.environ['PROXY_PASS']}@rotating.proxycove.com:8080"
proxies = {
'http': proxy_url,
'https': proxy_url
}
results = []
# إجراء 10 طلبات - كل منها مع IP جديد
for i in range(10):
try:
response = requests.get(
f'https://api.wildberries.ru/products/{i}',
proxies=proxies,
timeout=10
)
results.append({
'product_id': i,
'status': response.status_code,
'data': response.json()
})
except Exception as e:
results.append({
'product_id': i,
'error': str(e)
})
return {
'statusCode': 200,
'body': json.dumps(results)
}
تدوير يدوي من مجموعة البروكسي
إذا كان لديك قائمة بالبروكسي، يمكنك تنفيذ التدوير يدويًا. هذا مفيد عندما تحتاج إلى التحكم في البروكسي المستخدم لكل طلب:
import requests
import random
import json
def lambda_handler(event, context):
# قائمة البروكسي (يمكن تخزينها في DynamoDB أو S3)
proxy_pool = [
{
'host': 'proxy1.example.com',
'port': '8080',
'user': 'user1',
'pass': 'pass1'
},
{
'host': 'proxy2.example.com',
'port': '8080',
'user': 'user2',
'pass': 'pass2'
},
{
'host': 'proxy3.example.com',
'port': '8080',
'user': 'user3',
'pass': 'pass3'
}
]
results = []
for i in range(10):
# اختيار بروكسي عشوائي من المجموعة
proxy = random.choice(proxy_pool)
proxy_url = f"http://{proxy['user']}:{proxy['pass']}@{proxy['host']}:{proxy['port']}"
proxies = {
'http': proxy_url,
'https': proxy_url
}
try:
response = requests.get(
f'https://api.example.com/item/{i}',
proxies=proxies,
timeout=10
)
results.append({
'item': i,
'proxy_used': proxy['host'],
'status': response.status_code
})
except Exception as e:
results.append({
'item': i,
'proxy_used': proxy['host'],
'error': str(e)
})
return {
'statusCode': 200,
'body': json.dumps(results)
}
جلسات لاصقة للحفاظ على IP
تتطلب بعض المهام الحفاظ على IP واحد طوال الجلسة (على سبيل المثال، تسجيل الدخول إلى موقع). تقدم مزودي البروكسي جلسات لاصقة عبر معامل في URL:
import requests
import uuid
def lambda_handler(event, context):
# توليد session_id فريد
session_id = str(uuid.uuid4())
# بروكسي مع جلسة لاصقة (IP محفوظ لمدة 10 دقائق)
proxy_url = f"http://{os.environ['PROXY_USER']}-session-{session_id}:{os.environ['PROXY_PASS']}@sticky.proxycove.com:8080"
proxies = {
'http': proxy_url,
'https': proxy_url
}
# جميع الطلبات في هذه Lambda ستنفذ مع نفس IP
# 1. تسجيل الدخول
login_response = requests.post(
'https://example.com/login',
data={'user': 'test', 'pass': 'test'},
proxies=proxies
)
# 2. الحصول على البيانات (يستخدم نفس IP)
data_response = requests.get(
'https://example.com/dashboard',
proxies=proxies,
cookies=login_response.cookies
)
return {
'statusCode': 200,
'body': data_response.text
}
تخزين بيانات اعتماد البروكسي عبر متغيرات البيئة
لا تخزن أبدًا بيانات اعتماد البروكسي (اسم المستخدم، كلمة المرور، المضيف) مباشرة في كود وظيفة Lambda. تقدم AWS عدة طرق آمنة لتخزين البيانات الحساسة:
1. متغيرات البيئة (الطريقة الأساسية)
في وحدة تحكم AWS Lambda → الإعدادات → متغيرات البيئة، أضف:
PROXY_HOST= proxy.example.comPROXY_PORT= 8080PROXY_USER= your_usernamePROXY_PASS= your_password
تقوم AWS بتشفير متغيرات البيئة تلقائيًا في حالة السكون. الوصول إليها في الكود:
# Python
import os
proxy_host = os.environ['PROXY_HOST']
// Node.js
const proxyHost = process.env.PROXY_HOST;
2. AWS Secrets Manager (موصى به للإنتاج)
لأقصى درجات الأمان، استخدم AWS Secrets Manager - حيث يوفر تدويرًا تلقائيًا للأسرار وتحكمًا تفصيليًا في الوصول:
import boto3
import json
from botocore.exceptions import ClientError
def get_proxy_credentials():
secret_name = "proxy-credentials"
region_name = "us-east-1"
session = boto3.session.Session()
client = session.client(
service_name='secretsmanager',
region_name=region_name
)
try:
get_secret_value_response = client.get_secret_value(
SecretId=secret_name
)
secret = json.loads(get_secret_value_response['SecretString'])
return secret
except ClientError as e:
print(f"خطأ في استرجاع السر: {e}")
raise e
def lambda_handler(event, context):
# الحصول على بيانات الاعتماد من Secrets Manager
creds = get_proxy_credentials()
proxy_url = f"http://{creds['user']}:{creds['password']}@{creds['host']}:{creds['port']}"
# استخدام البروكسي
# ...
مهم: لا تنس إضافة حقوق IAM لوظيفة Lambda للوصول إلى Secrets Manager:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"secretsmanager:GetSecretValue"
],
"Resource": "arn:aws:secretsmanager:us-east-1:123456789:secret:proxy-credentials-*"
}
]
}
الأخطاء الشائعة وحلولها
عند العمل مع البروكسي في Lambda، غالبًا ما يواجه المطورون نفس المشكلات. دعنا نستعرض أكثرها شيوعًا وطرق حلها:
الخطأ: ProxyError / مهلة الاتصال
الأعراض: requests.exceptions.ProxyError: HTTPConnectionPool(host='proxy.example.com', port=8080): تجاوز الحد الأقصى من المحاولات
الأسباب:
- بيانات اعتماد البروكسي غير صحيحة (اسم المستخدم/كلمة المرور)
- خادم البروكسي غير متاح أو محمّل
- جدار الحماية يحظر الاتصالات الصادرة من Lambda
- مهلة قصيرة جدًا
الحل:
# 1. تحقق من بيانات الاعتماد
print(f"استخدام البروكسي: {proxy_host}:{proxy_port}")
print(f"المستخدم: {proxy_user}")
# 2. زيادة المهلة
response = requests.get(url, proxies=proxies, timeout=30)
# 3. إضافة منطق إعادة المحاولة
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
session = requests.Session()
retry = Retry(
total=3,
backoff_factor=1,
status_forcelist=[500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry)
session.mount('http://', adapter)
session.mount('https://', adapter)
response = session.get(url, proxies=proxies, timeout=30)
الخطأ: فشل التحقق من شهادة SSL
الأعراض: SSLError: [SSL: CERTIFICATE_VERIFY_FAILED]
السبب: تستخدم بعض البروكسي (خاصة الرخيصة) شهادات SSL موقعة ذاتيًا.
الحل (استخدم بحذر!):
# تعطيل التحقق من SSL (فقط للاختبار!)
response = requests.get(
url,
proxies=proxies,
verify=False # لا تستخدم في الإنتاج!
)
# الأفضل: حدد مسار شهادة CA
response = requests.get(
url,
proxies=proxies,
verify='/path/to/ca-bundle.crt'
)
مهم: يؤدي تعطيل التحقق من SSL (verify=False) إلى جعل الاتصال عرضة لهجمات man-in-the-middle. استخدمه فقط لأغراض التصحيح في بيئة التطوير!
الخطأ: مهلة Lambda (انتهت المهمة بعد X ثوانٍ)
الأعراض: تنتهي وظيفة Lambda بخطأ مهلة، دون انتظار استجابة من البروكسي.
السبب: بروكسي بطيئة (خاصة السكنية/المحمولة) + عدد كبير من الطلبات.
الحل:
- زيادة مهلة وظيفة Lambda: الإعدادات → الإعدادات العامة → المهلة (حد أقصى 15 دقيقة)
- تقليل عدد الطلبات في تنفيذ واحد
- استخدام الطلبات غير المتزامنة (asyncio في Python، Promise.all في Node.js)
- التحول إلى بروكسي أسرع للمهام غير الحرجة
# Python: طلبات غير متزامنة لتسريع
import asyncio
import aiohttp
async def fetch_url(session, url, proxy):
async with session.get(url, proxy=proxy, timeout=10) as response:
return await response.text()
async def lambda_handler_async(event, context):
proxy_url = f"http://{os.environ['PROXY_USER']}:{os.environ['PROXY_PASS']}@{os.environ['PROXY_HOST']}:{os.environ['PROXY_PORT']}"
urls = [f'https://api.example.com/item/{i}' for i in range(50)]
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url, proxy_url) for url in urls]
results = await asyncio.gather(*tasks)
return {
'statusCode': 200,
'body': json.dumps({'count': len(results)})
}
def lambda_handler(event, context):
return asyncio.run(lambda_handler_async(event, context))
الخطأ: 407 مطلوب مصادقة البروكسي
الأعراض: خطأ HTTP 407 عند محاولة استخدام البروكسي.
السبب: تنسيق غير صحيح لتمرير بيانات الاعتماد أو يتطلب البروكسي مصادقة IP بدلاً من اسم المستخدم/كلمة المرور.
الحل:
# تحقق من تنسيق URL البروكسي
# صحيح:
proxy_url = f"http://{user}:{password}@{host}:{port}"
# غير صحيح (تم حذف البروتوكول):
proxy_url = f"{user}:{password}@{host}:{port}" # ❌
# إذا كان البروكسي يتطلب مصادقة IP:
# 1. اعرف IP الخارجي لـ Lambda الخاص بك (قد يتغير!)
# 2. أضف هذا IP إلى القائمة البيضاء لمزود البروكسي
# 3. استخدم البروكسي بدون user:pass
# الحصول على IP الخارجي لـ Lambda:
response = requests.get('https://api.ipify.org?format=json')
lambda_ip = response.json()['ip']
print(f"IP الخارجي لـ Lambda: {lambda_ip}")
تحسين أداء Lambda مع البروكسي
يضيف استخدام البروكسي تأخيرًا إلى كل طلب. إليك طرق مثبتة لتقليل التأثير على الأداء:
1. تجميع الاتصالات
أعد استخدام اتصالات TCP بدلاً من إنشاء واحدة جديدة لكل طلب:
# Python: استخدم Session بدلاً من requests.get()
import requests
# أنشئ جلسة مرة واحدة (يمكنك إخراجها خارج المعالج)
session = requests.Session()
session.proxies = {
'http': proxy_url,
'https': proxy_url
}
def lambda_handler(event, context):
# جميع الطلبات تعيد استخدام الاتصالات
for i in range(100):
response = session.get(f'https://api.example.com/item/{i}')
# معالجة الاستجابة...
2. الطلبات المتوازية
إذا كنت بحاجة إلى إجراء العديد من الطلبات المستقلة، قم بتنفيذها بشكل متوازي:
// Node.js: الطلبات المتوازية مع Promise.all
const axios = require('axios');
const HttpsProxyAgent = require('https-proxy-agent');
const agent = new HttpsProxyAgent(proxyUrl);
exports.handler = async (event) => {
const urls = Array.from({length: 50}, (_, i) =>
`https://api.example.com/item/${i}`
);
// جميع الطلبات تُنفذ بشكل متوازي
const promises = urls.map(url =>
axios.get(url, {
httpsAgent: agent,
timeout: 10000
})
);
try {
const results = await Promise.all(promises);
return {
statusCode: 200,
body: JSON.stringify({
count: results.length,
data: results.map(r => r.data)
})
};
} catch (error) {
console.error('خطأ:', error.message);
return {
statusCode: 500,
body: JSON.stringify({ error: error.message })
};
}
};
3. تخزين النتائج في الذاكرة المؤقتة
إذا كانت البيانات تتغير نادرًا، قم بتخزين النتائج في DynamoDB أو S3:
import boto3
import json
import time
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('proxy-cache')
def get_cached_or_fetch(url, proxies, cache_ttl=3600):
"""يعود بالبيانات من الذاكرة المؤقتة أو يجري طلبًا عبر البروكسي"""
# تحقق من الذاكرة المؤقتة
try:
response = table.get_item(Key={'url': url})
if 'Item' in response:
item = response['Item']
if time.time() - item['timestamp'] < cache_ttl:
print(f"ضرب في الذاكرة المؤقتة لـ {url}")
return item['data']
except Exception as e:
print(f"خطأ في الذاكرة المؤقتة: {e}")
# الذاكرة المؤقتة فارغة أو منتهية الصلاحية - اجري الطلب
print(f"فشل في الذاكرة المؤقتة لـ {url}، يجري التحميل...")
response = requests.get(url, proxies=proxies, timeout=10)
data = response.text
# حفظ في الذاكرة المؤقتة
try:
table.put_item(Item={
'url': url,
'data': data,
'timestamp': int(time.time())
})
except Exception as e:
print(f"خطأ في حفظ الذاكرة المؤقتة: {e}")
return data
4. اختيار النوع الصحيح من البروكسي
مقارنة سرعة أنواع البروكسي المختلفة في ظروف حقيقية:
| نوع البروكسي | متوسط التأخير | طلبات/دقيقة (Lambda 1GB RAM) | توصية |
|---|---|---|---|
| مراكز البيانات | 50-200 مللي ثانية | 300-600 | جمع بيانات API بكثافة |
| سكنية | 300-800 مللي ثانية | 100-200 | المواقع المحمية |
| محمولة | 500-1500 مللي ثانية | 50-100 | تجاوز الحماية الصارمة |
توصيات:
- استخدم بروكسي سكنية للمواقع المحمية: مثل Ozon وWildberries.
- استخدم بروكسي مراكز البيانات لمهام جمع البيانات غير المحمية.
- استخدم بروكسي موبايل للاختبارات الإعلانية.