In this Article
WebスクレイピングやWebサイトのテストでは、プロキシが欠かせません。ただし、安定してアクセスするには、複数の異なるアドレスが必要です。だからといって、プロキシを手動で切り替える必要はありません。この課題はプロキシローテーターで解決できます。この記事では、Pythonでカスタムのプロキシローテーターを一から構築します。
優れたプロキシローテーターを構築する重要性
ローテーターはプロキシプールを自動で切り替えます。リクエストごと、または指定した時間間隔ごとに、別のIPアドレスを割り当てます。これにより、対象サーバーにはトラフィックが異なるアドレスや場所から送られているように見え、同一の送信元からのリクエストだと判別されにくくなります。プロキシローテーターは、スクレイピングやマルチアカウント運用の重要な基盤です。特に接続数が多い大規模プロジェクトでは、安定した認証フローの維持、Webサイトのトラフィックガイドラインへの準拠、地域限定コンテンツへのアクセスに役立ちます。
プロキシローテーターには、見落とされがちな利点もあります。たとえば、IPが継続的に変わるため、Webサイトやトラッカーが行動をプロファイリングしにくくなり、プライバシーと匿名性の向上につながります。実際のアドレスを隠せるため、特定のデバイスを狙う攻撃者や詐欺師から特定されにくくなります。
さらに、多くのWebサイトは、ユーザーの所在地や好みに応じて異なる結果を表示します。複数のIPを使用すれば、より多様で正確なデータを取得できます。ローテーションにより、不要なパーソナライズコンテンツの影響も抑えられます。
信頼性の高いプロキシローテーターに必要な要素
リクエストごとに新しいIPを使うだけでは十分ではありません。プロキシローテーターを有用で安定したものにするには、次の機能も必要です。
- 再試行ロジック: あるプロキシが機能しない場合、ローテーターは別のプロキシに切り替えます。
- User-Agentヘッダーをローテーションし、リクエストが同じ送信元からのものと見なされにくくします。
- 403 (アクセス禁止)、429 (リクエストが多すぎる)、500 (内部サーバーエラー) などの異常なHTTPステータスコードを処理します。
- 繰り返し失敗するプロキシを無効にします。
- ボット対策システムを作動させないよう、リクエスト間に短い待機時間を入れます。
- ログ: 再試行の理由と、使用したプロキシおよびUser-Agentを確認できるようにします。
- スクレイピング前のヘルスチェック: 開始前にすべてのプロキシをテストし、利用可能であることを確認します。
- 停止したプロキシを一定時間無効にするサーキットブレーカー。
- 必要に応じて、ホストごとにセッションを固定します。これにより、マルチアカウント運用などで同じセッションを維持する必要がある場合、ドメインには同じIPからアクセスしているように見えます。
プロキシローテーターを作成する際は、これらの点を考慮してください。
Pythonでプロキシローテーターを構築する
プロキシローテーターを1つのファイルに実装します。この例では、ファイル名を proxy_rotator.pyとします。以下にコード全体と動作の説明を示します。コピーしてそのまま使用できます。認証情報などをハードコードしていないため、後から置き換える必要はありません。
始める前に
- Pythonを 公式ウェブサイトからインストールするか、最新版にアップグレードしてください。バージョン3.8以降が必要です。
- 次のコマンドで、requestsライブラリをインストールします。HTTPリクエストの送信に必要です。
pip install requests
- ご利用のプランのページからプロキシ一覧を取得します。手動でコピーして.txtファイルに貼り付けることもできます。 DataImpulse ダッシュボードからダウンロードできます。この例ではファイル名をproxies.txt.としています。内容は次のとおりです。
http://login:password.host:port
https://login:password.host:port
必要に応じて、プロキシの形式を変更できます。
インポートと初期設定
まず、使用するモジュールをインポートします。
import argparse
import logging
import os
import random
import threading
import time
from dataclasses import dataclass
from typing import Dict, Iterable, List, Optional, Tuple
import requests
from requests import Response
requestsはWebリクエストを送信し、randomはプロキシとヘッダーを選択し、threadingは複数の接続スレッドを処理します。
既定の設定
次の設定では、User-Agentのローテーション、エラー時の別プロキシへの切り替え、プロキシの動作確認を行います。
DEFAULT_USER_AGENTS: List[str] = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:125.0) Gecko/20100101 Firefox/125.0",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.0.0",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 13_4) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.5 Safari/605.1.15",
"Mozilla/5.0 (iPhone; CPU iPhone OS 16_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.5 Mobile/15E148 Safari/604.1",
"Mozilla/5.0 (Linux; Android 13; Pixel 7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Mobile Safari/537.36",
]
DEFAULT_ROTATE_ON_STATUS = {403, 407, 409, 418, 421, 425, 429, 500, 502, 503, 504}
DEFAULT_HEALTHCHECK_URL = "http://api.ipify.org"
必要に応じてUser-Agentを置き換えたり追加したり、ヘルスチェックに別のWebサイトを使用したりできます。ここでは httpbin.org/ipを使用します。ほかには、 https://httpbin.org/get (リクエストヘッダーとIPを表示)、 https://api.ipify.org、 https://api.ipify.org?format=json (JSON形式でIPを返す)、 https://ifconfig.me または https://ifconfig.me/all.json (ヘッダーとJSON形式のIPを返す)、または http://ip-api.com/json/ (国、都市、ISP、IPを表示) が利用できます。
プロキシオブジェクト
この部分では、プロキシの保存方法を定義します。
@dataclass(frozen=True)
class Proxy:
url: str
label: str = ""
def to_requests_proxies(self) -> Dict[str, str]:
return {"http": self.url, "https": self.url}
ここでも、置き換える必要はありません。
プロキシファイルの読み込み
この部分では、プロキシを.txtファイルから読み込みます。
def parse_proxy_file(path: str) -> List[Proxy]:
proxies: List[Proxy] = []
with open(path, "r", encoding="utf-8") as f:
for line in f:
raw = line.strip()
if not raw or raw.startswith("#"):
continue
raw = raw.split(" #", 1)[0].split("\t#", 1)[0].strip()
proxies.append(Proxy(url=raw))
if not proxies:
raise ValueError(f"No proxies found in file: {path}")
return proxies
再試行時の待機時間
この部分では再試行ロジックを実装します。スロットリングやサービスからのブロックを避けるため、リクエストが失敗するたびにローテーターの待機時間を長くします。
def _exp_backoff_sleep(base: float, attempt: int, max_sleep: float) -> None:
sleep_s = min(base * (2 ** attempt) + random.random() * base, max_sleep)
time.sleep(sleep_s)
プロキシローテータークラス
ローテーターの中核となる部分です。プロキシを選択し、リクエストを再試行して失敗を追跡します。また、プロキシをランダムに使うか順番に使うかを決め、タイムアウトとヘッダーを設定します。同じWebサイトへのリクエストで同じプロキシを使うようにし、プロキシを許可またはブロック済みとして記録し、ヘルスチェックも実行します。
class ProxyRotator:
def __init__(
self,
proxies: Iterable[Proxy],
strategy: str = "round_robin",
max_retries: int = 3,
timeout: Tuple[float, float] = (10.0, 30.0),
rotate_on_status: Optional[Iterable[int]] = None,
user_agents: Optional[List[str]] = None,
sticky_per_host: bool = True,
circuit_threshold: int = 2,
circuit_cooldown: float = 60.0,
backoff_base: float = 0.5,
backoff_max: float = 5.0,
healthcheck_url: Optional[str] = None,
logger: Optional[logging.Logger] = None,
) -> None:
self.proxies: List[Proxy] = list(proxies)
if not self.proxies:
raise ValueError("ProxyRotator requires at least one proxy")
if strategy not in {"round_robin", "random"}:
raise ValueError("strategy must be 'round_robin' or 'random'")
self.strategy = strategy
self.max_retries = max_retries
self.timeout = timeout
self.rotate_on_status = set(rotate_on_status or DEFAULT_ROTATE_ON_STATUS)
self.user_agents = user_agents or DEFAULT_USER_AGENTS
self.sticky_per_host = sticky_per_host
self.circuit_threshold = circuit_threshold
self.circuit_cooldown = circuit_cooldown
self.failure_counts: Dict[Proxy, int] = {p: 0 for p in self.proxies}
self.disabled_until: Dict[Proxy, float] = {p: 0.0 for p in self.proxies}
self.backoff_base = backoff_base
self.backoff_max = backoff_max
self.healthcheck_url = healthcheck_url
self._local = threading.local()
self._sticky_map: Dict[str, Proxy] = {}
self._idx = 0
self._lock = threading.Lock()
self.log = logger or logging.getLogger("proxy_rotator")
def get(self, url: str, **kwargs) -> Response:
return self.request("GET", url, **kwargs)
def post(self, url: str, **kwargs) -> Response:
return self.request("POST", url, **kwargs)
def request(self, method: str, url: str, **kwargs) -> Response:
session = getattr(self._local, "session", None)
if session is None:
session = requests.Session()
self._local.session = session
headers = kwargs.pop("headers", {}) or {}
headers.setdefault("User-Agent", random.choice(self.user_agents))
timeout = kwargs.pop("timeout", self.timeout)
last_exc: Optional[Exception] = None
last_response: Optional[Response] = None
for attempt in range(self.max_retries + 1):
proxy = self._choose_proxy(url)
if proxy is None:
raise RuntimeError("No healthy proxies currently available")
try:
self.log.debug("Using proxy %s for %s %s", proxy.url, method, url)
response = session.request(
method,
url,
headers=headers,
proxies=proxy.to_requests_proxies(),
timeout=timeout,
**kwargs,
)
last_response = response
if response.status_code in self.rotate_on_status:
self._mark_failure(proxy, f"HTTP {response.status_code}")
self._maybe_unstick_host(url, proxy)
if attempt < self.max_retries:
_exp_backoff_sleep(self.backoff_base, attempt, self.backoff_max)
continue
else:
self._mark_success(proxy)
return response
except requests.RequestException as exc:
last_exc = exc
self._mark_failure(proxy, repr(exc))
self._maybe_unstick_host(url, proxy)
if attempt < self.max_retries:
_exp_backoff_sleep(self.backoff_base, attempt, self.backoff_max)
continue
break
if last_exc:
raise last_exc
assert last_response is not None
return last_response
def health_check(self, url: Optional[str] = None, sample: Optional[int] = None) -> Dict[str, bool]:
check_url = url or self.healthcheck_url or DEFAULT_HEALTHCHECK_URL
to_test = list(self.proxies)
if sample is not None:
to_test = random.sample(to_test, k=min(sample, len(to_test)))
results: Dict[str, bool] = {}
for p in to_test:
try:
r = requests.get(check_url, proxies=p.to_requests_proxies(), timeout=(5, 10))
healthy = r.ok
except Exception:
healthy = False
results[p.url] = healthy
if healthy:
self._mark_success(p)
else:
self._mark_failure(p, "healthcheck")
return results
def _choose_proxy(self, url: str) -> Optional[Proxy]:
host = requests.utils.urlparse(url).hostname or ""
now = time.time()
with self._lock:
if self.sticky_per_host:
sticky = self._sticky_map.get(host)
if sticky and self.disabled_until.get(sticky, 0.0) <= now:
return sticky
enabled = [p for p in self.proxies if self.disabled_until.get(p, 0.0) <= now]
if not enabled:
return None
if self.strategy == "round_robin":
proxy = enabled[self._idx % len(enabled)]
self._idx = (self._idx + 1) % len(enabled)
else:
proxy = random.choice(enabled)
if self.sticky_per_host:
self._sticky_map[host] = proxy
return proxy
def _mark_failure(self, proxy: Proxy, reason: str) -> None:
with self._lock:
cnt = self.failure_counts.get(proxy, 0) + 1
self.failure_counts[proxy] = cnt
if cnt >= self.circuit_threshold:
self.disabled_until[proxy] = time.time() + self.circuit_cooldown
self.log.warning("Disabling proxy %s for %.1fs (reason: %s)", proxy.url, self.circuit_cooldown, reason)
def _mark_success(self, proxy: Proxy) -> None:
with self._lock:
self.failure_counts[proxy] = 0
self.disabled_until[proxy] = 0.0
def _maybe_unstick_host(self, url: str, proxy: Proxy) -> None:
if not self.sticky_per_host:
return
host = requests.utils.urlparse(url).hostname or ""
with self._lock:
if self._sticky_map.get(host) == proxy:
self._sticky_map.pop(host, None)
コマンドラインオプション
ターミナルから1行のコマンドでローテーターを実行するための部分です。
def build_arg_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(description="HTTP proxy rotator")
p.add_argument("--url", required=False, help="Target URL to request (for quick testing)")
p.add_argument("--method", default="GET", help="HTTP method to use (default: GET)")
p.add_argument(
"--proxies",
help="Path to proxies.txt file (one proxy URL per line). If omitted, reads PROXY_LIST env (comma-separated)",
)
p.add_argument("--strategy", choices=["round_robin", "random"], default="round_robin")
p.add_argument("--retries", type=int, default=3, help="Max retry attempts across proxies")
p.add_argument(
"--timeout",
type=float,
default=15.0,
help="Read timeout in seconds (connect timeout is fixed at 10s in this simple CLI)",
)
p.add_argument(
"--rotate-on",
default=",".join(str(s) for s in sorted(DEFAULT_ROTATE_ON_STATUS)),
help="Comma-separated status codes that trigger rotation",
)
p.add_argument(
"--no-sticky",
action="store_true",
help="Disable sticky-per-host behavior",
)
p.add_argument(
"--healthcheck",
action="store_true",
help="Run a healthcheck across proxies and print a summary before the request",
)
p.add_argument(
"--debug",
action="store_true",
help="Enable verbose logging",
)
return p
def _load_proxies_from_args(args: argparse.Namespace) -> List[Proxy]:
if args.proxies:
return parse_proxy_file(args.proxies)
env = os.getenv("PROXY_LIST", "").strip()
if env:
return [Proxy(url=p.strip()) for p in env.split(",") if p.strip()]
raise SystemExit("No proxies provided. Use --proxies or set PROXY_LIST env.")
def main_cli() -> None:
parser = build_arg_parser()
args = parser.parse_args()
logging.basicConfig(
level=logging.DEBUG if args.debug else logging.INFO,
format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
log = logging.getLogger("proxy_rotator")
proxies = _load_proxies_from_args(args)
rotate_on_set = {int(x.strip()) for x in args.rotate_on.split(",") if x.strip()}
rotator = ProxyRotator(
proxies=proxies,
strategy=args.strategy,
max_retries=args.retries,
timeout=(10.0, args.timeout),
rotate_on_status=rotate_on_set,
user_agents=DEFAULT_USER_AGENTS,
sticky_per_host=not args.no_sticky,
circuit_threshold=2,
circuit_cooldown=60.0,
backoff_base=0.5,
backoff_max=5.0,
healthcheck_url=DEFAULT_HEALTHCHECK_URL,
logger=log,
)
if args.healthcheck:
results = rotator.health_check()
total = len(results)
healthy = sum(1 for ok in results.values() if ok)
log.info("Healthcheck: %s/%s proxies OK", healthy, total)
for url, ok in results.items():
log.info(" %-40s %s", url, "OK" if ok else "BAD")
if args.url:
try:
resp = rotator.request(args.method.upper(), args.url)
print(f"Status: {resp.status_code}")
text = resp.text
preview = text[:500].replace("\n", " ")
print(f"Body preview (first 500 chars):\n{preview}")
except Exception as e:
log.error("Request failed: %s", e)
raise
else:
log.info("No --url provided; nothing else to do. Add --url to test a request.")
エントリーポイント
ここでローテーターを起動します。プロキシを読み込み、必要に応じてチェックしたうえで、リクエストを実行します。
if __name__ == "__main__":
main_cli()
プロキシローテーターを実行する
次のコマンドで、コマンドラインからローテーターを実行できます。
python proxy_rotator.py --proxies ./proxies.txt --url http://api.ipify.org
./proxies.txtを実際のプロキシファイル名に置き換え、– – URLをスクレイピングまたはテストするWebサイトのURLに置き換えます。
さらに、次のフラグも追加できます。
– – 戦略ランダム – ローテーターがプールからプロキシをランダムに選択します
– – 再試行 5 – 再試行回数を指定します
– – 健康診断 – 事前にプロキシをテストします
– – べたつきません – スティッキープロキシを無効にします。同じセッションを維持する必要がない場合にのみ使用してください
– – デバッグ – トラブルシューティングに役立つ詳細ログを表示します。
注記: 場合によっては、Python 3をインストールしている環境では、「python3 proxy_rotator.py」を使用する必要があります。 –-プロキシ ./proxies.txt –-url http://api.ipify.orgをローテーターの実行コマンドとして使用してください。
まとめ
プロキシローテーターの構築は、長くも難しくもありません。プロキシプールを最大限に活用し、再試行ロジック、ヘルスチェック、待機時間の制御を備えたローテーターにすることで、Webスクレイピングやマルチアカウント運用を効率化できます。信頼でき、倫理的に調達されたIPをお探しなら、DataImpulseをご利用ください。195の拠点から9,000万を超えるIPを提供し、途切れにくい地域横断の接続を支援します。お問い合わせは[email protected]、または「今すぐ試してください」ボタンから始めてください。