IPブロックとは——スクレイピングの最大の壁
大規模なスクレイピングを行うと、対象サイトのサーバーから同一IPアドレスへのリクエストを制限（IPブロック）されることがあります。これはサーバー保護のための正当な措置であり、スクレイパー側は以下の対策が必要です。
前提として：スクレイピングはサイトの利用規約を確認し、許可された範囲内で、サーバーに過大な負荷をかけない形で実施することが大原則です。
プロキシとは
プロキシサーバーは、クライアントとWebサーバーの間に立つ中継サーバーです。プロキシを経由することで、WebサーバーからはプロキシサーバーのIPアドレスが送信元として見えます。
クライアント → プロキシA → Webサーバー（プロキシAのIPを検知）
クライアント → プロキシB → Webサーバー（プロキシBのIPを検知）
プロキシの種類と選び方
| 種類 | 説明 | コスト | 信頼性 |
|---|---|---|---|
| 無料プロキシ | 誰でも使える公開プロキシ | 無料 | 低（すぐブロックされる） |
| 共有プロキシ | 複数人で共有 | 安価 | 中 |
| 専用プロキシ | 自分専用 | 中程度 | 高 |
| 住宅用プロキシ（Residential） | 実際のISP IPアドレス | 高価 | 最高 |
| データセンタープロキシ | クラウドIPアドレス | 中程度 | 中〜高 |
Pythonでのプロキシローテーション実装
import requests
import random
import time
from itertools import cycle
class ProxyRotator:
"""プロキシをラウンドロビンでローテーションする"""
def __init__(self, proxies: list[str]):
"""
proxies: ["http://user:pass@host:port", ...]
"""
self._proxies = proxies
self._cycle = cycle(proxies)
self._failed = set()
def get_proxy(self) -> dict:
"""使用可能なプロキシを1つ返す"""
for _ in range(len(self._proxies)):
proxy = next(self._cycle)
if proxy not in self._failed:
return {'http': proxy, 'https': proxy}
# 全プロキシが失敗したらリセット
self._failed.clear()
return {'http': self._proxies[0], 'https': self._proxies[0]}
def mark_failed(self, proxy_url: str) -> None:
"""失敗したプロキシを記録"""
# dictからURLを取り出す場合
self._failed.add(proxy_url)
@property
def available_count(self) -> int:
return len(self._proxies) - len(self._failed)
スクレイパークラスへの組み込み
import requests
from requests.exceptions import ProxyError, ConnectionError, Timeout
class ResilientScraper:
def __init__(self, proxies: list[str], delay: tuple = (1.0, 3.0)):
self.rotator = ProxyRotator(proxies)
self.delay = delay
self.session = requests.Session()
self.session.headers.update({
'User-Agent': (
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) '
'AppleWebKit/537.36 (KHTML, like Gecko) '
'Chrome/124.0.0.0 Safari/537.36'
),
'Accept-Language': 'ja-JP,ja;q=0.9',
})
def fetch(self, url: str, max_retries: int = 3) -> str | None:
"""プロキシローテーション付きでページを取得"""
for attempt in range(max_retries):
proxy = self.rotator.get_proxy()
try:
response = self.session.get(
url,
proxies=proxy,
timeout=15,
)
response.raise_for_status()
# 成功時のウェイト
time.sleep(random.uniform(*self.delay))
return response.text
except (ProxyError, ConnectionError) as e:
# プロキシエラー：そのプロキシを無効化
proxy_url = list(proxy.values())[0]
self.rotator.mark_failed(proxy_url)
print(f"プロキシ失敗 [{attempt+1}/{max_retries}]: {proxy_url}")
except Timeout:
print(f"タイムアウト [{attempt+1}/{max_retries}]: {url}")
time.sleep(5)
except requests.HTTPError as e:
if e.response.status_code == 403:
print(f"403 Forbidden: IPブロックの可能性 {url}")
time.sleep(10)
elif e.response.status_code == 429:
print(f"429 Too Many Requests: {e.response.headers.get('Retry-After', 60)}秒待機")
time.sleep(int(e.response.headers.get('Retry-After', 60)))
else:
raise
return None
def fetch_many(self, urls: list[str]) -> dict[str, str | None]:
"""複数URLを一括取得"""
results = {}
for i, url in enumerate(urls):
print(f"[{i+1}/{len(urls)}] 取得中: {url}")
results[url] = self.fetch(url)
return results
実際の使い方
# プロキシリスト（環境変数または設定ファイルから読む）
import os
proxies = os.environ.get('PROXY_LIST', '').split(',')
# 例: "http://user:pass@proxy1.example.com:8080,http://user:pass@proxy2.example.com:8080"
scraper = ResilientScraper(proxies=proxies, delay=(1.5, 3.5))
urls = [
'https://example.com/page/1',
'https://example.com/page/2',
'https://example.com/page/3',
]
results = scraper.fetch_many(urls)
for url, html in results.items():
if html:
print(f"取得成功: {url} ({len(html)}文字)")
else:
print(f"取得失敗: {url}")
セッション管理でクッキーを維持する
# ログインが必要なサイトではセッションを維持
login_session = requests.Session()
login_session.post('https://example.com/login', data={
'email': os.environ['EMAIL'],
'password': os.environ['PASSWORD'],
})
# ログイン後のプロキシ付きリクエスト
proxy = rotator.get_proxy()
protected_page = login_session.get(
'https://example.com/protected',
proxies=proxy,
)
まとめ
プロキシローテーションはIPブロックを回避するための技術ですが、あくまでも倫理的・合法的なスクレイピングの範囲内で使用することが重要です。弊社では適切なウェイト・robots.txt遵守・利用規約確認を前提に、プロキシを組み合わせた安定したデータ収集システムを構築しています。
スクレイピング・データ収集システムの開発についてはお気軽にご相談ください。
