Google Cloud Functions یک پلتفرم بدون سرور برای اجرای کد بدون مدیریت سرورها است. هنگام کار با پارسینگ، اتوماسیون درخواستهای API یا جمعآوری دادهها، اغلب نیاز به مسیریابی ترافیک از طریق پروکسی برای دور زدن مسدودیتها، چرخش IP و هدفگذاری جغرافیایی وجود دارد. در این راهنما، تنظیم پروکسی در Cloud Functions با Python و Node.js را با نمونههای عملی بررسی خواهیم کرد.
چرا باید از پروکسی در Cloud Functions استفاده کنیم
Google Cloud Functions در یک محیط ایزوله با IPهای مشترک مراکز داده Google کار میکند. هنگام درخواستهای مکرر به APIهای خارجی یا وبسایتها، مشکلاتی به وجود میآید:
- مسدودیتهای IP — بسیاری از سرویسها (Google، Facebook، بازارها) ترافیک را از مراکز داده شناسایی کرده و محدودیت نرخ یا مسدودیت کامل اعمال میکنند.
- محدودیتهای جغرافیایی — برای دسترسی به محتوایی که فقط در کشورهای خاص در دسترس است (مثلاً پارسینگ قیمتهای منطقهای در Wildberries یا Ozon).
- محدودیت نرخ — یک آدرس IP میتواند تعداد محدودی درخواست در دقیقه ارسال کند. پروکسیها اجازه میدهند بار را توزیع کنیم.
- ناشناس بودن — پنهان کردن منبع واقعی درخواستها هنگام کار با دادههای حساس یا اطلاعات رقابتی.
سناریوهای معمول استفاده از پروکسی در Cloud Functions:
- پارسینگ بازارها (Wildberries، Ozon، Amazon) برای نظارت بر قیمتهای رقباء
- جمعآوری دادهها از شبکههای اجتماعی (Instagram، TikTok) از طریق API یا وباسکرپینگ
- اتوماسیون بررسی تبلیغات در مناطق مختلف
- درخواستهای انبوه به موتورهای جستجو (Google، Yandex) برای تحلیل SEO
- آزمایش عملکردهای جغرافیایی برنامهها
کدام نوع پروکسی برای Cloud Functions مناسب است
انتخاب نوع پروکسی بستگی به وظیفه، بودجه و الزامات ناشناس بودن دارد. در اینجا مقایسهای از گزینههای اصلی آورده شده است:
| نوع پروکسی | سرعت | ناشناس بودن | بهترین برای |
|---|---|---|---|
| پروکسی دیتاسنتر | بالا (50-200 میلیثانیه) | متوسط | پارسینگ وبسایتهای ساده، درخواستهای API، نظارت بر SEO |
| پروکسی مسکونی | متوسط (200-800 میلیثانیه) | بالا | پارسینگ شبکههای اجتماعی، بازارها، دور زدن سیستمهای ضد ربات |
| پروکسی موبایل | متوسط (300-1000 میلیثانیه) | بسیار بالا | Instagram، TikTok، برنامههای موبایل، Facebook API |
توصیهها برای انتخاب:
- برای پارسینگ بازارها (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 whitelist
دو روش اصلی برای احراز هویت هنگام کار با پروکسی وجود دارد:
1. احراز هویت با نام کاربری و رمز عبور
رایجترین روش — ارسال اطلاعات کاربری در URL پروکسی:
http://username:password@proxy.example.com:8080
مزایا: سادگی تنظیم، نیاز به IP ثابت منبع ندارد.
معایب: اطلاعات کاربری در هر درخواست ارسال میشود، بار اضافی کمی دارد.
2. احراز هویت با IP whitelist
برخی از ارائهدهندگان اجازه میدهند IPهای Cloud Functions را به whitelist اضافه کنید. مشکل: Cloud Functions از IPهای دینامیک از استخر Google Cloud استفاده میکند.
راهحل: از Cloud NAT برای مسیریابی ترافیک خروجی از طریق IP خارجی ثابت استفاده کنید:
- ایجاد شبکه VPC و زیرشبکه در Google Cloud
- تنظیم Cloud NAT با رزرو IP ثابت
- اتصال Cloud Functions به VPC Connector
- اضافه کردن IP ثابت به whitelist ارائهدهنده پروکسی
پس از تنظیم پروکسی، نیازی به نام کاربری و رمز عبور نیست:
proxies = {
'http': 'http://proxy.example.com:8080',
'https': 'http://proxy.example.com:8080'
}
توصیه: برای اکثر موارد از احراز هویت با نام کاربری/رمز عبور استفاده کنید — این سادهتر است و نیاز به هزینههای اضافی برای Cloud NAT (از 0.044 دلار در ساعت + ترافیک) ندارد.
چرخش IP و مدیریت استخر پروکسی
هنگام پارسینگ حجم بالای دادهها، استفاده از چرخش IP برای جلوگیری از مسدودیتها حیاتی است. چندین رویکرد وجود دارد:
1. چرخش در سمت ارائهدهنده (پروکسیهای چرخشی)
بسیاری از ارائهدهندگان پروکسیهای چرخشی را ارائه میدهند — یک endpoint واحد که IP را در هر درخواست یا بر اساس زمان بهطور خودکار تغییر میدهد:
# یک endpoint، 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، پردازش صحیح خطاها برای جلوگیری از از دست دادن دادهها و تجاوز از محدودیتهای زمان اجرا حیاتی است.
انواع خطاها و روشهای پردازش
| خطا | دلیل | راهحل |
|---|---|---|
| ProxyError | پروکسی در دسترس نیست یا اطلاعات کاربری نادرست است | به پروکسی دیگری از استخر سوئیچ کنید |
| Timeout | پروکسی کند یا سرور شلوغ | تایماوت 5-10 ثانیهای تنظیم کنید، با IP دیگری دوباره امتحان کنید |
| 407 Proxy Authentication Required | نام کاربری/رمز عبور نادرست است | اطلاعات کاربری را در متغیرهای محیطی بررسی کنید |
| 429 Too Many Requests | محدودیت نرخ در وبسایت هدف | تأخیر بین درخواستها اضافه کنید، از IP بیشتری استفاده کنید |
| 403 Forbidden | IP پروکسی توسط وبسایت مسدود شده است | IP را تغییر دهید، از پروکسی مسکونی بهجای دیتاسنتر استفاده کنید |
نمونهای از پردازش جامع خطاها
import requests
import time
from requests.exceptions import ProxyError, Timeout, RequestException
def fetch_with_retry(url, proxy_pool, max_retries=3):
"""
درخواست با retry خودکار و تغییر پروکسی در صورت خطا
"""
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 دارای تأخیر در شروع سرد (cold start) هستند. وابستگیها را به حداقل برسانید و از نسخههای حداقلی کتابخانهها استفاده کنید:
# 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"
# یا از endpointهای مختلف برای کشورهای مختلف استفاده کنید
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 فرصتهای گستردهای برای پارسینگ، اتوماسیون و کار با API بدون محدودیتهای IP فراهم میکند. نکات اصلی که باید در نظر داشته باشید: پردازش صحیح خطاها با منطق retry، استفاده از تایماوتها برای جلوگیری از قفل شدن، چرخش IP برای جلوگیری از مسدودیتها و ذخیره امن اطلاعات کاربری در Secret Manager.
برای اکثر وظایف پارسینگ و اتوماسیون، انتخاب بهینه پروکسیهای مسکونی خواهد بود — آنها ناشناس بودن بالایی را فراهم میکنند و درصد کمی از مسدودیتها را به دلیل استفاده از IPهای کاربران واقعی دارند. برای کار با شبکههای اجتماعی و برنامههای موبایل، پروکسیهای موبایل را توصیه میکنیم که IPهای اپراتورهای موبایل دارند و تقریباً توسط پلتفرمهایی مانند Instagram و TikTok مسدود نمیشوند.
با تنظیم صحیح Cloud Functions با پروکسی، شما یک راهحل مقیاسپذیر و اقتصادی برای پردازش حجم بالای دادهها بدون نیاز به مدیریت زیرساخت دارید.