Scrapyフレームワークでのプロキシ設定：コード例付きの完全ガイド

Scrapyは、ウェブスクレイピングのための最も強力なPythonフレームワークの1つですが、プロキシの適切な設定がなければ、数分でブロックされてしまいます。このガイドでは、Scrapyにプロキシを統合するすべての方法を示します：最も簡単な設定から、自動エラーハンドリングを伴う高度なIPアドレスローテーション手法まで。

この資料は、大規模なeコマースサイトや保護されたサイトのパースに関する実際の経験に基づいています。すぐにプロジェクトで使用できるコードの例を提供します。

なぜScrapyはプロキシなしでブロックされるのか

現代のウェブサイトは、パースからの多層防御を使用しています。User-Agentやリクエスト間の遅延を設定しても、IPアドレスは以下のいくつかの兆候で自動化を示します：

• リクエストの頻度：1つのIPが1分間に100回以上のリクエストを行うのは、明らかなボットの兆候です。
• 行動パターン：ランダムな遷移なしにページを順次巡回する。
• JavaScriptの欠如：ScrapyはJSを実行しないため、簡単に検出されます。
• ジオロケーション：自宅のネットワークではなく、データセンターからのアクセス。

結果として、数時間または数日のIPバンが発生します。特に、マーケットプレイス（Amazon、Wildberries、Ozon）、ソーシャルメディア、Cloudflareを使用しているサイトは、非常に攻撃的な保護を行っています。プロキシは、この問題を解決し、リクエストを多数のIPアドレスに分散させます。

Scrapyでのプロキシの基本設定

最も簡単な方法は、スパイダーの設定に直接プロキシを指定することです。この方法は、テストや単一のプロキシサーバーでの小規模データのパースに適しています。

方法1：リクエストのmetaを介して

import scrapy

class MySpider(scrapy.Spider):
    name = 'example'
    start_urls = ['https://example.com']
    
    def start_requests(self):
        proxy = 'http://username:password@proxy.example.com:8080'
        
        for url in self.start_urls:
            yield scrapy.Request(
                url=url,
                callback=self.parse,
                meta={'proxy': proxy}
            )
    
    def parse(self, response):
        # パースロジック
        self.log(f'Scraped {response.url} via {response.meta["proxy"]}')

プロキシの形式は、プロトコルと認証方法によって異なります：

• http://proxy.example.com:8080 — 認証なし
• http://user:pass@proxy.example.com:8080 — ユーザー名/パスワードあり
• socks5://user:pass@proxy.example.com:1080 — SOCKS5プロキシ

方法2：settings.pyでのグローバル設定

# settings.py

# すべてのリクエスト用のHTTPプロキシ
HTTPPROXY_ENABLED = True
HTTPPROXY_AUTH_ENCODING = 'utf-8'

# 環境変数を介した設定
HTTP_PROXY = 'http://username:password@proxy.example.com:8080'
HTTPS_PROXY = 'http://username:password@proxy.example.com:8080'

この方法は迅速なテストには便利ですが、プロダクションには適していません：IPのローテーションがなく、プロキシがダウンすると全てのパーサーが停止し、異なるサイトに異なるプロキシを使用することができません。

カスタムプロキシミドルウェアの作成

プロダクションパースには、プロキシプールを管理し、エラーを処理し、自動的にIPをローテーションする独自のミドルウェアが必要です。以下は基本的な実装です：

# middlewares.py

import random
from scrapy import signals
from scrapy.exceptions import NotConfigured

class RandomProxyMiddleware:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
    
    @classmethod
    def from_crawler(cls, crawler):
        # 設定からプロキシリストをロード
        proxy_list = crawler.settings.getlist('PROXY_LIST')
        
        if not proxy_list:
            raise NotConfigured('PROXY_LIST not configured')
        
        return cls(proxy_list)
    
    def process_request(self, request, spider):
        # プールからランダムなプロキシを選択
        proxy = random.choice(self.proxy_list)
        request.meta['proxy'] = proxy
        
        spider.logger.info(f'Using proxy: {proxy}')
    
    def process_exception(self, request, exception, spider):
        # エラーが発生した場合、別のプロキシを試みる
        proxy = random.choice(self.proxy_list)
        request.meta['proxy'] = proxy
        
        spider.logger.warning(
            f'Proxy error, switching to: {proxy}'
        )
        
        return request

次に、settings.pyでミドルウェアの使用を設定します：

# settings.py

# プロキシのリスト（ファイルまたはAPIからロード可能）
PROXY_LIST = [
    'http://user1:pass1@proxy1.example.com:8080',
    'http://user2:pass2@proxy2.example.com:8080',
    'http://user3:pass3@proxy3.example.com:8080',
    # ... 効率的なローテーションのために50以上のプロキシを追加
]

# ミドルウェアを接続
DOWNLOADER_MIDDLEWARES = {
    'myproject.middlewares.RandomProxyMiddleware': 350,
    'scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware': 400,
}

# エラー時の再試行
RETRY_TIMES = 3
RETRY_HTTP_CODES = [500, 502, 503, 504, 408, 429]

プロキシのローテーション：3つの実用的な方法

ランダムなプロキシの選択（上記の例のように）は最も簡単ですが、最も効果的な方法ではありません。さまざまなシナリオに対する3つのローテーション戦略を考えてみましょう。

方法1：ラウンドロビン（順次ローテーション）

プロキシは順番に選ばれます。負荷を均等に分散させるのに適しています：

class RoundRobinProxyMiddleware:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
        self.current_index = 0
    
    @classmethod
    def from_crawler(cls, crawler):
        proxy_list = crawler.settings.getlist('PROXY_LIST')
        return cls(proxy_list)
    
    def process_request(self, request, spider):
        # 次のプロキシを順番に取得
        proxy = self.proxy_list[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.proxy_list)
        
        request.meta['proxy'] = proxy

方法2：ブラックリストを使用したスマートローテーション

問題のあるプロキシを追跡し、一時的にローテーションから除外します：

import time
from collections import defaultdict

class SmartProxyMiddleware:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
        self.proxy_errors = defaultdict(int)
        self.blacklist = set()
        self.blacklist_timeout = 300  # 5分
        self.blacklist_time = {}
    
    @classmethod
    def from_crawler(cls, crawler):
        proxy_list = crawler.settings.getlist('PROXY_LIST')
        return cls(proxy_list)
    
    def get_working_proxies(self):
        # ブラックリストからタイムアウトが切れたプロキシを除外
        current_time = time.time()
        expired = [
            proxy for proxy, ban_time in self.blacklist_time.items()
            if current_time - ban_time > self.blacklist_timeout
        ]
        
        for proxy in expired:
            self.blacklist.discard(proxy)
            self.proxy_errors[proxy] = 0
        
        # 動作中のプロキシを返す
        return [p for p in self.proxy_list if p not in self.blacklist]
    
    def process_request(self, request, spider):
        working_proxies = self.get_working_proxies()
        
        if not working_proxies:
            spider.logger.error('すべてのプロキシがブラックリストに登録されています！')
            return
        
        proxy = random.choice(working_proxies)
        request.meta['proxy'] = proxy
    
    def process_response(self, request, response, spider):
        # ブロックされた場合はブラックリストに追加
        if response.status in [403, 429, 503]:
            proxy = request.meta.get('proxy')
            self.proxy_errors[proxy] += 1
            
            if self.proxy_errors[proxy] >= 3:
                self.blacklist.add(proxy)
                self.blacklist_time[proxy] = time.time()
                spider.logger.warning(
                    f'プロキシ {proxy} は {self.blacklist_timeout}s ブラックリストに登録されました'
                )
        
        return response

方法3：プロバイダーのAPIを介したローテーション

多くのプロキシプロバイダー（レジデンシャルプロキシを含む）は、リクエストごとに自動的にIPを変更するローティングエンドポイントを提供しています：

# settings.py

# 自動ローテーションのための単一のエンドポイント
ROTATING_PROXY = 'http://username:password@rotating.proxy.com:8080'

# シンプルなミドルウェア
class RotatingProxyMiddleware:
    def __init__(self, proxy):
        self.proxy = proxy
    
    @classmethod
    def from_crawler(cls, crawler):
        proxy = crawler.settings.get('ROTATING_PROXY')
        return cls(proxy)
    
    def process_request(self, request, spider):
        # 単一のURLだが、各リクエストは新しいIPで行われる
        request.meta['proxy'] = self.proxy

これはプロダクションに最も便利な方法です：プロキシプールを管理する必要がなく、プロバイダーが自動的にIPの質を監視し、問題のあるものを交換します。特に、レジデンシャルプロキシでは、IPプールが数百万のアドレスに達することがあります。

認証：ユーザー名/パスワード vs IPホワイトリスト

プロキシプロバイダーは2つの認証方法を提供しています。選択は接続速度と設定の便利さに影響します。

ユーザー名：パスワード認証

ユーザー名とパスワードはプロキシのURLに渡されます。Scrapyは自動的にこれをHTTPヘッダーProxy-Authorizationに変換します：

proxy = 'http://username:password@proxy.example.com:8080'
request.meta['proxy'] = proxy

# Scrapyは自動的にヘッダーを追加します：
# Proxy-Authorization: Basic dXNlcm5hbWU6cGFzc3dvcmQ=

利点：任意のIPから動作し、コード内でプロキシを簡単に変更できます。
欠点：各リクエストにわずかなオーバーヘッド（約50-100ms）があり、コード内に資格情報が平文で表示されます。

IPホワイトリスト認証

サーバーのIPをプロバイダーのホワイトリストに追加し、認証は不要です：

proxy = 'http://proxy.example.com:8080'  # ユーザー名/パスワードなし
request.meta['proxy'] = proxy

利点：50-100ms速く、安全性が高い（コード内に資格情報がない）。
欠点：特定のIPからのみ動作し、サーバーを変更する際にホワイトリストを更新する必要があります。

エラーハンドリングと自動IP変更

質の高いプロキシを使用しても、タイムアウト、接続拒否、ブロックなどのエラーが発生します。エラーの適切な処理は、パーサーの安定した動作にとって重要です。

HTTPステータスの処理

class ProxyMiddleware:
    def process_response(self, request, response, spider):
        # プロキシを変更して再試行する必要があるコード
        ban_codes = [403, 407, 429, 503]
        
        if response.status in ban_codes:
            proxy = request.meta.get('proxy')
            spider.logger.warning(
                f'{proxy}から{response.status}を受け取りました。再試行します...'
            )
            
            # 新しいプロキシで再試行のためにマーク
            request.meta['dont_retry'] = False
            request.meta['proxy'] = self.get_new_proxy()
            
            return request
        
        return response

ネットワーク例外の処理

from twisted.internet.error import TimeoutError, ConnectionRefusedError
from scrapy.exceptions import IgnoreRequest

class ProxyMiddleware:
    def process_exception(self, request, exception, spider):
        # プロキシへの接続エラー
        proxy_errors = (
            TimeoutError,
            ConnectionRefusedError,
            ConnectionLost,
        )
        
        if isinstance(exception, proxy_errors):
            proxy = request.meta.get('proxy')
            spider.logger.error(
                f'プロキシ {proxy} の接続に失敗しました：{exception}'
            )
            
            # プロキシを変更して再試行
            request.meta['proxy'] = self.get_new_proxy()
            return request
        
        # その他のエラーには標準の処理を使用
        return None

コンテンツによるブロックの検出

一部のサイトはHTTP 200を返しますが、キャプチャやブロックページを表示します：

class ProxyMiddleware:
    def process_response(self, request, response, spider):
        # コンテンツ内のブロックの兆候
        ban_indicators = [
            'captcha',
            'access denied',
            'blocked',
            'unusual traffic',
            'robot check',
        ]
        
        body_text = response.text.lower()
        
        if any(indicator in body_text for indicator in ban_indicators):
            spider.logger.warning(
                f'{request.meta.get("proxy")}からブロックページが検出されました'
            )
            
            # プロキシを変更して再試行
            request.meta['proxy'] = self.get_new_proxy()
            return request
        
        return response

Scrapyに適したプロキシの種類

プロキシの種類の選択は、ターゲットサイト、予算、必要なパース速度によって異なります。以下は主要なオプションの比較です：

選択に関する推奨

マーケットプレイスのパースの場合（Amazon、Wildberries、Ozon、AliExpress） — レジデンシャルプロキシのみを使用してください。これらのサイトはデータセンターを攻撃的にブロックします。ローテーションとジオターゲティングが必要です（例えば、Wildberries用のロシアのIP）。

ニュースサイト、ブログ、フォーラムのパースの場合 — データセンタープロキシが適しています。保護は最小限で、速度と低コストのトラフィックが重要です。

Cloudflareを使用しているサイトのパースの場合 — レジデンシャルプロキシが必須です。Cloudflareのデータセンターはほぼ瞬時に検出します。Scrapyにcloudscraperライブラリを追加してJSチャレンジを回避してください。

Google検索、SEOツールのパースの場合 — ジオターゲティングを伴うレジデンシャルプロキシが必要です。Googleは国や都市によって異なる結果を表示します。

高度な技術：セッションとスティッキーIP

一部のサイトをパースする際には、セッション全体で同じIPを保持する必要があります（認証、ショッピングカート、多段階フォーム）。以下は、Scrapyでスティッキーセッションを実現する方法です。

1つのドメイン用のスティッキーIP

from urllib.parse import urlparse

class StickyProxyMiddleware:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
        # 辞書：ドメイン -> プロキシ
        self.domain_proxy_map = {}
    
    @classmethod
    def from_crawler(cls, crawler):
        proxy_list = crawler.settings.getlist('PROXY_LIST')
        return cls(proxy_list)
    
    def process_request(self, request, spider):
        # URLからドメインを抽出
        domain = urlparse(request.url).netloc
        
        # このドメインに既にプロキシがある場合はそれを使用
        if domain in self.domain_proxy_map:
            proxy = self.domain_proxy_map[domain]
        else:
            # そうでない場合は新しいものを選択して記憶
            proxy = random.choice(self.proxy_list)
            self.domain_proxy_map[domain] = proxy
            spider.logger.info(f'{domain}に{proxy}を割り当てました')
        
        request.meta['proxy'] = proxy

セッションタイムアウト付きのスティッキーIP

より高度なオプション：プロキシはドメインに一定の時間（例えば10分）バインドされ、その後変更されます：

import time
from urllib.parse import urlparse

class SessionProxyMiddleware:
    def __init__(self, proxy_list, session_timeout=600):
        self.proxy_list = proxy_list
        self.session_timeout = session_timeout  # 10分
        # 辞書：ドメイン -> (プロキシ, 作成時間)
        self.sessions = {}
    
    @classmethod
    def from_crawler(cls, crawler):
        proxy_list = crawler.settings.getlist('PROXY_LIST')
        timeout = crawler.settings.getint('PROXY_SESSION_TIMEOUT', 600)
        return cls(proxy_list, timeout)
    
    def get_proxy_for_domain(self, domain):
        current_time = time.time()
        
        # アクティブなセッションがあるか確認
        if domain in self.sessions:
            proxy, created_at = self.sessions[domain]
            
            # セッションが切れていなければ同じプロキシを使用
            if current_time - created_at < self.session_timeout:
                return proxy
        
        # 新しいプロキシで新しいセッションを作成
        new_proxy = random.choice(self.proxy_list)
        self.sessions[domain] = (new_proxy, current_time)
        
        return new_proxy
    
    def process_request(self, request, spider):
        domain = urlparse(request.url).netloc
        proxy = self.get_proxy_for_domain(domain)
        request.meta['proxy'] = proxy

Cookieミドルウェアとの統合

完全なセッションには、プロキシとクッキーを同期させる必要があります。Scrapyはドメインごとにクッキーを別々に保存しますが、プロキシを変更する際にはクッキーをクリアする必要があります：

# settings.py

# クッキーのミドルウェアを有効にする
COOKIES_ENABLED = True
COOKIES_DEBUG = False

# プロキシとクッキーを同期するためのミドルウェア
class ProxyCookieMiddleware:
    def process_request(self, request, spider):
        # 現在のプロキシを取得
        current_proxy = request.meta.get('proxy')
        
        # プロキシが変更された場合はクッキーをクリア
        previous_proxy = request.meta.get('previous_proxy')
        
        if previous_proxy and previous_proxy != current_proxy:
            # このドメインのクッキーをクリア
            jar = spider.crawler.engine.downloader.middleware.middlewares[0].jars
            domain = urlparse(request.url).netloc
            
            if domain in jar:
                jar[domain].clear()
                spider.logger.info(f'{domain}のクッキーをクリアしました')
        
        request.meta['previous_proxy'] = current_proxy

結論

Scrapyでのプロキシの適切な設定は、ブロックなしで安定したパースの基盤です。基本的な統合から、高度なローテーション技術やセッション管理まで、すべての重要な側面を考察しました。

主な結論：

• プロダクションでは、スマートローテーションと問題のあるIPのブラックリストを持つカスタムミドルウェアを使用してください。
• すべてのタイプのエラーを処理してください：HTTPステータス、ネットワーク例外、コンテンツによるブロック。
• タスクに応じてプロキシの種類を選択してください：シンプルなサイトにはデータセンター、保護されたサイトにはレジデンシャル。
• 認証が必要なサイトには、ドメインにプロキシをバインドしたスティッキーセッションを使用してください。
• 10-50のプロキシのプールから始め、負荷が増加したらスケールアップしてください。

保護されたサイト（マーケットプレイス、ソーシャルメディア、Cloudflareを使用しているサイト）をパースする予定がある場合は、レジデンシャルプロキシを使用することをお勧めします。これにより、最大限の匿名性と最小限のブロックリスクが確保されます。高速パースには、ローティングエンドポイントと10,000以上のIPのプールを持つプロバイダーを選択してください。

この記事のすべてのコード例は、Scrapy 2.xでテストされており、プロダクションでの使用に適しています。これらを自分のタスクに合わせて調整し、プロジェクトの成長に応じてスケールアップしてください。