تغییر دستی پروکسی هنگام کار با صدها درخواست، اتلاف وقت و پول است. روتیشن API امکان تغییر خودکار آدرسهای IP هنگام مسدودسازی، توزیع بار و مقیاسپذیری پارسینگ یا مالتی اکانتینگ را فراهم میکند. در این راهنما، نحوه تنظیم تغییر خودکار پروکسی برای وظایف مختلف را بررسی خواهیم کرد: از پارسینگ مارکتپلیسها تا فارم اکانتها در Facebook Ads.
این مطلب هم برای توسعهدهندگانی که پارسرهایی با Python یا Node.js مینویسند و هم برای آربیتراژورهایی که از ابزارهای آماده با یکپارچهسازی API استفاده میکنند، مناسب است.
چرا به اتوماسیون تغییر پروکسی از طریق API نیاز داریم
روتیشن خودکار آدرسهای IP از طریق API چندین وظیفه حیاتی را حل میکند که متخصصان در حوزههای مختلف با آنها مواجه هستند:
برای پارسینگ مارکتپلیسها و وبسایتها: هنگام جمعآوری داده از Wildberries، Ozon یا Avito، هر IP میتواند تعداد محدودی درخواست انجام دهد (معمولاً 50-200 در ساعت). روتیشن API امکان تغییر خودکار به IP جدید هنگام رسیدن به محدودیت یا دریافت کپچا را فراهم میکند و جمعآوری مداوم داده را تضمین میکند.
برای آربیتراژ و مالتی اکانتینگ: هنگام کار با 20-50 کابینت تبلیغاتی Facebook Ads یا اکانتهای Instagram، نیاز به جداسازی هر پروفایل است. API امکان اختصاص برنامهریزیشده پروکسی منحصربهفرد به هر اکانت در Dolphin Anty یا AdsPower و بازسازی خودکار نشستها هنگام مسدودسازی را میدهد.
برای اتوماسیون SMM: سرویسهای پست انبوه در Instagram، TikTok یا VK باید اقدامات را بین آدرسهای IP توزیع کنند تا از rate limits اجتناب کنند. API امکان دریافت پویای پروکسیهای جدید برای هر نشست یا گروه اکانت را فراهم میکند.
مزایای اصلی اتوماسیون API در مقایسه با تغییر دستی:
- سرعت: تغییر IP به صورت برنامهریزیشده در میلیثانیه و بدون دخالت انسان انجام میشود
- مقیاسپذیری: امکان مدیریت همزمان هزاران پروکسی از طریق یک رابط واحد
- تحمل خطا: جایگزینی خودکار پروکسیهای غیرفعال بدون توقف فرآیند
- انعطافپذیری: تنظیم قوانین روتیشن برای وظیفه خاص: بر اساس زمان، تعداد درخواستها، جغرافیا
- صرفهجویی: استفاده بهینه از ترافیک از طریق توازن بار
سناریوی معمول استفاده: شما قیمتهای رقبا را در Wildberries پارس میکنید. بدون API، باید به صورت دستی مسدودسازیها را ردیابی کنید، وارد پنل ارائهدهنده پروکسی شوید، دادههای جدید را کپی کنید و آنها را در اسکریپت قرار دهید. با API همه اینها به صورت خودکار انجام میشود: اسکریپت خطای 429 (Too Many Requests) را دریافت میکند، درخواست به API سرویس پروکسی ارسال میکند، IP جدید دریافت میکند و کار را ادامه میدهد.
انواع روتیشن پروکسی: sticky sessions در مقابل تغییر خودکار
قبل از تنظیم اتوماسیون، درک تفاوت بین انواع روتیشن آدرسهای IP مهم است. انتخاب استراتژی به وظیفه شما بستگی دارد.
Sticky Sessions (پروکسیهای نشستی)
هنگام استفاده از sticky sessions، یک آدرس IP برای مدت زمان مشخصی (معمولاً از 5 تا 30 دقیقه) به نشست شما اختصاص داده میشود. تغییر فقط پس از پایان زمان نشست یا با درخواست API شما انجام میشود.
چه زمانی استفاده کنیم:
- کار با اکانتها در شبکههای اجتماعی (Instagram، Facebook) - تغییر مکرر IP باعث سوءظن میشود
- پر کردن فرمهای چند صفحهای که نیاز به حفظ نشست دارند
- تست تبلیغات از یک منطقه خاص در طول نشست
- پارسینگ سایتها با احراز هویت که تغییر IP منجر به خروج از سیستم میشود
نمونه درخواست API برای ایجاد sticky session (معمولاً از فرمت ویژه لاگین استفاده میشود):
// فرمت: username-session-SESSIONID:password
// SESSIONID — هر رشتهای، یکسان = یک IP
proxy = "username-session-abc123:password@gate.proxycove.com:8000"
// همه درخواستها با session-abc123 یک IP را برای مدت نشست دریافت میکنند
// برای IP جدید از SESSIONID دیگری استفاده کنید: session-xyz789
روتیشن خودکار در هر درخواست
آدرس IP در هر اتصال جدید به سرور پروکسی تغییر میکند. این رفتار استاندارد پروکسیهای مسکونی بدون مشخص کردن پارامترهای نشست است.
چه زمانی استفاده کنیم:
- پارسینگ انبوه بدون احراز هویت (قیمتها، مخاطبین، آگهیها)
- دور زدن rate limits تهاجمی در 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 سرویس پروکسی پیادهسازی میشود.
چه زمانی استفاده کنیم:
- بهینهسازی مصرف ترافیک - تغییر فقط در صورت نیاز
- کار با سایتهایی که الگوها را ردیابی میکنند (تغییر بیش از حد مکرر = مسدودسازی)
- توازن بین ناشناس بودن و پایداری نشست
| نوع روتیشن | فرکانس تغییر IP | وظایف | مصرف ترافیک |
|---|---|---|---|
| Sticky Session | 5-30 دقیقه | مالتی اکانتینگ، احراز هویت | کم |
| روتیشن خودکار | هر درخواست | پارسینگ، دور زدن rate limits | متوسط |
| بر اساس تایمر | قابل تنظیم | همهمنظوره | بهینهشده |
مبانی کار با 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 — ایجاد sticky session جدید با پارامترها (کشور، شهر، مدت زمان)
احراز هویت معمولاً از طریق کلید 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 استفاده میشود. برای روتیشن خودکار، یک کلاس wrapper ایجاد میکنیم که پروکسی را هنگام خطاها تغییر میدهد:
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') # بدون GUI
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);
یکپارچهسازی API با مرورگرهای ضد شناسایی: 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", # ID گروه پروفایلها
"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)
چنین اتوماسیونی هنگام کار با دهها اکانت بسیار مهم است. به جای کپی دستی دادههای پروکسی در هر پروفایل، اسکریپت را اجرا میکنید و در عرض چند دقیقه زیرساخت آماده دریافت میکنید.
مدیریت خطاها و fallback خودکار
هنگام کار با پروکسی، موقعیتهای اجتنابناپذیری وجود دارد: IP توسط سایت هدف مسدود شده، سرور پروکسی پاسخ نمیدهد، ترافیک تمام شده است. مدیریت صحیح خطاها کلید اتوماسیون پایدار است.
انواع خطاها و استراتژیهای مدیریت
| خطا | علت | راهحل |
|---|---|---|
| HTTP 403 Forbidden | IP در لیست سیاه سایت | تغییر پروکسی، افزودن تاخیر |
| HTTP 429 Too Many Requests | تجاوز از rate limit | تغییر IP، افزایش فاصله |
| ProxyError / Timeout | سرور پروکسی پاسخ نمیدهد | حذف از استخر، انتخاب بعدی |
| 407 Proxy Authentication Required | نام کاربری/رمز عبور نادرست | بررسی credentials، بهروزرسانی |
| کپچا در صفحه | سایت ربات را شناسایی کرده | تغییر IP، استفاده از پروکسیهای موبایل |
پیادهسازی سیستم هوشمند retry
به جای تکرار ساده درخواست، سیستمی با تاخیر نمایی و لیست سیاه پروکسیهای غیرفعال ایجاد میکنیم:
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}")
مانیتورینگ و هشدارها
برای سیستمهای production، ردیابی سلامت استخر پروکسی در زمان واقعی مهم است. لاگگیری معیارها را اضافه کنید:
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))
بهترین شیوهها و بهینهسازی مصرف ترافیک
برای به حداکثر رساندن کارایی و کاهش هزینهها، این توصیهها را دنبال کنید:
- استفاده از sticky sessions برای وظایف با حالت: اگر نیاز به حفظ نشست دارید (ورود به سیستم، سبد خرید)، از پروکسیهای نشستی استفاده کنید تا از تغییرات غیرضروری IP جلوگیری کنید
- کش کردن پاسخها: دادههایی که به ندرت تغییر میکنند (لیست دستهبندیها، اطلاعات استاتیک) را کش کنید تا از درخواستهای تکراری جلوگیری کنید
- تنظیم تاخیرهای بهینه: تاخیر بیش از حد کوتاه = مسدودسازی، بیش از حد طولانی = کاهش سرعت. برای هر سایت آزمایش کنید (معمولاً 2-5 ثانیه)
- استفاده از User-Agent واقعی: User-Agent مرورگرهای محبوب را چرخش دهید تا طبیعیتر به نظر برسید
- توزیع بار بین پروکسیها: به جای استفاده تصادفی، درخواستها را به صورت یکنواخت توزیع کنید
- مانیتورینگ کیفیت پروکسی: پروکسیهایی با نرخ خطای بالا را به صورت خودکار حذف کنید
- استفاده از جغرافیای مناسب: برای سایتهای روسی از پروکسیهای روسیه، برای آمریکایی از آمریکا استفاده کنید
نمونه پیادهسازی کش ساده:
import hashlib
import json
from datetime import datetime, timedelta
class CachedProxyClient:
def __init__(self, rotator, cache_ttl=3600):
self.rotator = rotator
self.cache = {}
self.cache_ttl = cache_ttl # ثانیه
def _get_cache_key(self, url, params=None):
"""ایجاد کلید کش از URL و پارامترها"""
key_string = url + json.dumps(params or {}, sort_keys=True)
return hashlib.md5(key_string.encode()).hexdigest()
def get(self, url, params=None, use_cache=True):
"""درخواست GET با کش"""
cache_key = self._get_cache_key(url, params)
# بررسی کش
if use_cache and cache_key in self.cache:
cached_data, cached_time = self.cache[cache_key]
if datetime.now() - cached_time < timedelta(seconds=self.cache_ttl):
print(f"✓ بازیابی از کش: {url}")
return cached_data
# درخواست جدید
response = self.rotator.request_with_retry(url)
# ذخیره در کش
self.cache[cache_key] = (response, datetime.now())
return response
# استفاده:
cached_client = CachedProxyClient(rotator, cache_ttl=1800) # 30 دقیقه
# اولین درخواست - از شبکه
data1 = cached_client.get("https://api.example.com/categories")
# درخواست دوم - از کش
data2 = cached_client.get("https://api.example.com/categories")
نتیجهگیری
اتوماسیون تغییر پروکسی از طریق API ابزار قدرتمندی است که کارایی پارسینگ، آربیتراژ و مالتی اکانتینگ را چندین برابر افزایش میدهد. با پیادهسازی صحیح میتوانید:
- جمعآوری داده را بدون مسدودسازی مقیاسپذیر کنید
- دهها و صدها اکانت را به صورت همزمان مدیریت کنید
- مصرف ترافیک را بهینه کنید و هزینهها را کاهش دهید
- فرآیندها را کاملاً خودکار کنید و زمان را آزاد کنید
نکات کلیدی برای شروع:
- نوع روتیشن مناسب برای وظیفه خود را انتخاب کنید (sticky/auto/timer)
- مدیریت خطاها و سیستم retry را پیادهسازی کنید
- معیارها را مانیتور کنید و پروکسیهای مشکلدار را حذف کنید
- از کش و تاخیرهای بهینه برای صرفهجویی در ترافیک استفاده کنید
- با حجم کوچک شروع کنید و تدریجاً مقیاس دهید
برای کسانی که به دنبال پروکسی با API قدرتمند برای اتوماسیون هستند، ProxyCove ارائه میدهد:
- API RESTful با مستندات کامل
- پشتیبانی از sticky sessions با مدت زمان قابل تنظیم
- روتیشن خودکار و دستی
- استخر 10M+ IP مسکونی از 195 کشور
- آمار و مانیتورینگ در زمان واقعی
با اتوماسیون صحیح، پروکسیها از مشکل به مزیت رقابتی تبدیل میشوند که به شما امکان میدهد بر روی رشد کسبوکار تمرکز کنید، نه بر مسائل فنی.