APIを通じたプロキシ切替の自動化：スクレイピングとアービトラージのためのローテーション設定方法

数百のリクエストを処理する際の手動プロキシ切り替えは、時間とコストの無駄です。APIローテーションにより、ブロック時のIPアドレス自動切り替え、負荷分散、スクレイピングやマルチアカウント運用のスケーリングが可能になります。本ガイドでは、マーケットプレイスのスクレイピングからFacebook Adsアカウントのファーミングまで、様々なタスクに対応したプロキシ自動切り替えの設定方法を解説します。

本資料は、PythonやNode.jsでスクレイパーを開発する開発者、API統合機能を持つ既製ツールを使用するアービトラージャーの両方に適しています。

APIによるプロキシ自動切り替えが必要な理由

API経由のIPアドレス自動ローテーションは、様々な分野の専門家が直面するいくつかの重要な課題を解決します:

手動切り替えと比較したAPI自動化の主な利点:

• 速度: IP切り替えがミリ秒単位でプログラム的に実行され、人間の介入不要
• スケーラビリティ: 単一インターフェースで数千のプロキシを同時管理可能
• 耐障害性: プロセスを停止せずに動作しないプロキシを自動交換
• 柔軟性: 特定のタスクに応じたローテーションルールの設定:時間、リクエスト数、地域別
• コスト削減: 負荷分散によるトラフィックの最適利用

典型的な使用シナリオ:Wildberriesで競合他社の価格をスクレイピングする場合。APIなしでは、ブロックを手動で監視し、プロキシプロバイダーのパネルにアクセスし、新しいデータをコピーしてスクリプトに貼り付ける必要があります。APIを使用すれば、これらすべてが自動化されます:スクリプトが429エラー(Too Many Requests)を受信すると、プロキシサービスのAPIにリクエストを送信し、新しいIPを取得して作業を継続します。

プロキシローテーションのタイプ:スティッキーセッション vs 自動切り替え

自動化を設定する前に、IPアドレスローテーションのタイプの違いを理解することが重要です。戦略の選択はタスクによって異なります。

スティッキーセッション(セッション型プロキシ)

スティッキーセッションを使用すると、1つのIPアドレスが一定期間(通常5〜30分)セッションに固定されます。切り替えはセッション時間の終了時またはAPIリクエストによってのみ発生します。

使用すべき場合:

• SNSアカウント(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の厳格なレート制限の回避
• 10〜20リクエスト後にIPをBANするサイトからのデータ収集
• 異なる地域からのコンテンツアクセス可能性の確認

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を通じて実装されるハイブリッドアプローチです。

使用すべき場合:

• トラフィック消費の最適化 — 必要時のみ切り替え
• パターンを追跡するサイトでの作業(頻繁すぎる切り替え = BAN)
• 匿名性とセッション安定性のバランス

プロキシサービスAPIの基本

最新のプロキシプロバイダーの多くは、WebパネルとAPIの2つの管理方法を提供しています。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')  # 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"ページ {i}のHTML取得、長さ: {len(html)}")

bot.driver.quit()

プロキシローテーション用middlewareを持つScrapy

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により、プロファイル作成とプロキシ紐付けを自動化できます。

API経由のDolphin Anty自動化

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)

これで、Dolphin Antyに50のプロファイルができ、それぞれが一意のブラウザフィンガープリントとプロキシを持ちます。プロファイルをプログラム的に起動するには:

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)

このような自動化は、数十のアカウントを扱う際に非常に重要です。各プロファイルにプロキシデータを手動でコピーする代わりに、スクリプトを実行して数分で準備完了のインフラを取得できます。

エラー処理と自動フォールバック

プロキシを使用する際、避けられない状況があります:ターゲットサイトによる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:
            # すべてのプロキシがBAN — ブラックリストクリア
            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"試行 {attempt+1} 前に{delay}秒待機")
                    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"{proxy} でProxyError")
                self.mark_error(proxy)
                
            except requests.exceptions.Timeout:
                print(f"{proxy} でTimeout")
                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))

ベストプラクティスとトラフィック消費の最適化

プロキシローテーションを効果的に使用し、コストを最小限に抑えるための推奨事項:

1. タスクに適したローテーションタイプを選択

• スティッキーセッション: アカウント作業、認証が必要なサイト、長いセッション
• リクエストごとのローテーション: 大量の匿名スクレイピング、厳格なレート制限
• タイマー式: 柔軟性が必要な場合、トラフィック最適化

2. リクエスト間に遅延を追加

プロキシを変更しても、同じサイトへの毎秒リクエストは疑念を招きます。人間の行動を模倣します:

import random
import time

def human_delay():
    """人間らしいランダム遅延"""
    delay = random.uniform(2, 5)  # 2〜5秒
    time.sleep(delay)

for page in range(100):
    response = session.get(url)
    process_data(response)
    human_delay()  # 各リクエスト後

3. プロキシプールをセグメント化

異なるタスクには異なるプロキシセットを使用:

proxy_pools = {
    "wildberries": ["proxy1", "proxy2", "proxy3"],  # WB専用
    "ozon": ["proxy4", "proxy5", "proxy6"],         # Ozon専用
    "avito": ["proxy7", "proxy8", "proxy9"]         # Avito専用
}

# サイトごとに独自のプール
wb_session = RotatingProxySession(proxy_pools["wildberries"])
ozon_session = RotatingProxySession(proxy_pools["ozon"])

4. User-Agentとヘッダーをローテーション

IP変更だけでは不十分 — ブラウザフィンガープリントも変更します:

import random

user_agents = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36...",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36...",
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36..."
]

def get_random_headers():
    return {
        "User-Agent": random.choice(user_agents),
        "Accept-Language": random.choice(["en-US,en;q=0.9", "ru-RU,ru;q=0.9"]),
        "Accept-Encoding": "gzip, deflate, br",
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
    }

response = session.get(url, headers=get_random_headers())

5. トラフィック消費を監視

レジデンシャルプロキシは通常GBごとに課金されます。消費を追跡します:

class TrafficMonitor:
    def __init__(self):
        self.total_bytes = 0
    
    def log_response(self, response):
        # レスポンスサイズを計算
        size = len(response.content)
        self.total_bytes += size
        
        # MBに変換
        mb = self.total_bytes / (1024 * 1024)
        print(f"消費トラフィック: {mb:.2f} MB")
        
        # 制限チェック
        if mb > 1000:  # 1GB
            print("⚠ トラフィック制限に近づいています!")

monitor = TrafficMonitor()

for url in urls:
    response = session.get(url)
    monitor.log_response(response)

6. 地域ターゲティングを使用

特定の国や都市からのプロキシが必要な場合、APIパラメータで指定します:

# 特定の地域のプロキシリクエスト
response = requests.post(
    "https://api.provider.com/v1/session/create",
    json={
        "country": "US",
        "state": "California",
        "city": "Los Angeles"
    },
    headers={"Authorization": f"Bearer {API_KEY}"}
)

proxy_data = response.json()["session"]

まとめ

API経由のプロキシ自動切り替えは、スクレイピング、アービトラージ、マルチアカウント運用を行う専門家にとって必須のツールです。適切な実装により:

• 手動IP管理に費やす時間を数十時間削減
• ブロックやキャプチャの数を最小限に抑制
• 数百、数千のアカウントやリクエストにスケール
• トラフィックとプロキシコストを最適化
• 安定性と耐障害性を向上

重要なのは、タスクに適したローテーションタイプを選択し、エラー処理を適切に実装し、ベストプラクティスに従うことです。本ガイドの例は、Python、Node.js、または他の言語でのプロキシ自動化の開始点として使用できます。