パーサープールにおける悪質なプロキシの除外：エラーメトリクスとPythonによるローテーション

WildberriesやOzonのパーサーが突然空のページ、タイムアウト、403エラーを返し始めた場合、問題は必ずしもプラットフォーム自体にあるわけではありません。ほとんどの場合、プロキシプール内の一部のIPアドレスはすでに「腐って」います: それらは禁止され、遅延されるか、CAPTCHAの下に置かれています。機能するプロキシと死んでいるプロキシを見分けるためのメトリクスを分析し、手動管理なしで除外を自動化する方法を見ていきます。

なぜプロキシプールは時間とともに劣化するのか

どんなプロキシプールでも — 信頼できるプロバイダーから購入したものであっても — 時間が経つにつれて機能するアドレスの一部を失います。理由はさまざまです: Wildberries、Ozon、AvitoのアンチフロードシステムはIPに対して評判のマークを蓄積し、一部のアドレスはプロバイダーの他のクライアントによる大量リクエストの後にブラックリストに載ります（プロキシが共有の場合）、データセンターの範囲はサブネットごとに一斉に禁止され、居住者のIPは時には単にプールから外れることがあります。なぜなら、実際のユーザーがデバイスをオフにしたからです。

プールをクリーンアップしないと、「死んでいる」プロキシの割合は非線形に増加します: 最初は5%、アクティブなパーシングを1週間続けると20-30%になります。パーサーは機能しないIPを通じて再試行にリソースを費やし始め、1ページの収集にかかる平均時間が増加し、ログには実際の問題を隠すエラーが蓄積されます（例えば、サイトのレイアウトの変更）。除外システムがなければ、プールを「予備」として膨らませるか、データ収集の失敗を定期的に受けることになります。

目標は、エラーの事実を記録するだけでなく、それを分類することです: サーバーの一時的な過負荷、特定のIPの禁止、地域的なブロック、またはパーサー側の問題（不正なヘッダー、古いクッキー）。エラーの種類によって、プロキシをプールから永久に除外する必要があるのか、一時的に「隔離」するのか、単にリクエストを再試行するのかが決まります。

除外のための主要なエラーメトリクス

一時的な障害と本当に死んでいるプロキシを見分けるためには、各IPについていくつかのメトリクスを収集する必要があります。単一のHTTPステータスに依存するのではなく、以下の基本セットを各リクエストでログに記録する価値があります:

メトリクス	何を示すか	アクション
HTTP 403 / 429	IPの禁止またはレート制限	30-60分の隔離、再試行時は削除
接続タイムアウト	プロキシが過負荷または死んでいる	3回のタイムアウトでプールから削除
レスポンスにCAPTCHA	IPがアンチボットの疑いをかけられた	リクエストの頻度を下げるかIPを交換
応答時間 > 5秒	プロキシ側の速度劣化	プール内での優先度を下げる
平均レスポンスサイズが平均以下	ダミー/ブロックページが返された	受け入れる前に内容を追加確認
成功したリクエストの割合 (success rate)	IPの総合的な品質評価	50リクエストで70%未満 — 削除

サイトのエラーコードとネットワークエラーを混同しないことが重要です。例えば、404はプロキシの問題ではなく、Wildberriesで商品が存在しないことを示しています。しかし、同じIPが他のIPが200を受け取る場所で定期的に403を受け取る場合、それはこのアドレスが特定の禁止を受けている明確な兆候です。

プロキシのスコアリングアルゴリズム

バイナリの「動作中/動作していない」ではなく、各IPのスコアを0から100まで計算する方が便利で、エラーが発生するたびに徐々に減少し、成功したリクエストで回復します。これにより、偶発的な障害の後にプロキシをすぐに除外するのではなく、システマティックに無駄なレスポンスを返すIPを迅速に除外できます。

ロジックはシンプルです: コード200で有効な内容の成功したリクエストはスコアに+2を加えます（ただし100を超えない）。タイムアウトは-15。コード403/429は-25。レスポンスにCAPTCHAがある場合は-20。スコアが閾値（例えば30）を下回ると、プロキシは固定時間隔離され、隔離から戻った後にスコアが再び10未満になると、IPはプールから完全に削除され、分析のためにログに記録されます。

このアプローチは、Dolphin AntyやAdsPowerのようなアンチデテクトブラウザの評判システムに似ており、アカウントは活動に応じて「健康」を持っています。同じメカニズムがIPアドレスのパーシングにも適しています。

コード: Pythonによる自動除外付きProxyPool

以下は、エラーメトリクスを考慮したプロキシプールを管理するためのクラスの作業フレームワークです。これは、requests、httpx、またはScrapyの任意のパーサーに接続でき、プロキシ取得のロジックをpool.get_proxy()の呼び出しに置き換えることができます。

import time
import random
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class ProxyStats:
    address: str
    score: int = 100
    quarantine_until: float = 0.0
    total_requests: int = 0
    total_success: int = 0
    consecutive_timeouts: int = 0
    is_dead: bool = False

    @property
    def success_rate(self) -> float:
        if self.total_requests == 0:
            return 1.0
        return self.total_success / self.total_requests


class ProxyPool:
    def __init__(self, proxies: list[str],
                 min_score: int = 30,
                 dead_score: int = 10,
                 quarantine_seconds: int = 1800):
        self.stats = {p: ProxyStats(address=p) for p in proxies}
        self.min_score = min_score
        self.dead_score = dead_score
        self.quarantine_seconds = quarantine_seconds

    def _is_available(self, stats: ProxyStats) -> bool:
        if stats.is_dead:
            return False
        if stats.quarantine_until > time.time():
            return False
        return True

    def get_proxy(self) -> Optional[str]:
        available = [s for s in self.stats.values() if self._is_available(s)]
        if not available:
            return None
        # スコアでソートし、トップ30%からランダムに選択して1つのIPを叩かないようにする
        available.sort(key=lambda s: s.score, reverse=True)
        top_pool = available[:max(1, len(available) // 3)]
        return random.choice(top_pool).address

    def report_success(self, proxy: str):
        s = self.stats[proxy]
        s.total_requests += 1
        s.total_success += 1
        s.consecutive_timeouts = 0
        s.score = min(100, s.score + 2)

    def report_timeout(self, proxy: str):
        s = self.stats[proxy]
        s.total_requests += 1
        s.consecutive_timeouts += 1
        s.score -= 15
        if s.consecutive_timeouts >= 3:
            self._quarantine(s)
        self._check_dead(s)

    def report_blocked(self, proxy: str, captcha: bool = False):
        s = self.stats[proxy]
        s.total_requests += 1
        s.score -= 20 if captcha else 25
        self._quarantine(s)
        self._check_dead(s)

    def _quarantine(self, s: ProxyStats):
        if s.score < self.min_score:
            s.quarantine_until = time.time() + self.quarantine_seconds

    def _check_dead(self, s: ProxyStats):
        if s.score <= self.dead_score:
            s.is_dead = True

    def stats_report(self) -> list[dict]:
        return [
            {
                "proxy": s.address,
                "score": s.score,
                "success_rate": round(s.success_rate, 2),
                "dead": s.is_dead,
            }
            for s in self.stats.values()
        ]

実際のパーサーでは、report_success、report_timeout、およびreport_blockedは、各HTTPリクエストの応答処理の中に直接挿入されます。requestsとの統合の例は以下の通りです:

import requests

def fetch(url: str, pool: ProxyPool):
    proxy = pool.get_proxy()
    if not proxy:
        raise RuntimeError("プールに利用可能なプロキシがありません")

    proxies = {"http": proxy, "https": proxy}
    try:
        resp = requests.get(url, proxies=proxies, timeout=8)
    except requests.exceptions.Timeout:
        pool.report_timeout(proxy)
        return None

    if resp.status_code in (403, 429):
        pool.report_blocked(proxy)
        return None

    if "captcha" in resp.text.lower():
        pool.report_blocked(proxy, captcha=True)
        return None

    pool.report_success(proxy)
    return resp

このアプローチは手動監視の負担を軽減します: 一度閾値を設定すれば、プールは自動的に「回復」し、死んだアドレスを排除し、一時的に停止したアドレスを隔離から戻します。

IP禁止のための閾値

普遍的な数字は存在しません — 閾値はプラットフォームとそのアンチボットの攻撃性によって異なります。しかし、Wildberries、Ozon、Avitoのパーシングにおいて実際にうまく機能する指標があります:

• 成功率が70%未満で30-50リクエストのサンプル — 現在のスコアがまだ危機的でなくても、IPを削除する理由になります。
• 3回のタイムアウトが連続 — ほぼ常にプロキシが物理的に利用できないか過負荷であることを意味し、アクティブプールに保持する意味はありません。
• 20リクエストごとに1回以上のCAPTCHA — IPはすでにアンチボットの疑いをかけられており、リクエストの頻度を下げるかアドレスを交換する方が良いです。
• 平均応答時間がプールの中央値の2倍以上 — 劣化の信号であり、危機的ではありませんが、選択時に優先度を下げる理由になります。

30-60分の隔離は通常、データセンターおよび居住者プロキシにとって十分であり、アンチボットが一時的な疑わしい活動を「忘れる」ために必要です。モバイルプロキシの場合は状況が異なります: そこではIPがキャリアによってセルを変更する際に変わるため、隔離は短くすることができます — 10-15分で、ローテーション後は物理的に別のアドレスです。

ヘルスチェックとバックグラウンドでのローテーション

エラーの実際の除外（パーシングの主流でのエラーに基づく）に加えて、バックグラウンドでのヘルスチェックを実行することも有益です — 数分ごとにプロキシを中立的なURL（例えば、httpbin.org/ip）に軽いリクエストで叩き、プロキシがWildberriesやOzonへの本番リクエストに入る前にプールの統計を更新する別のプロセスです。

import threading

def healthcheck_loop(pool: ProxyPool, interval: int = 300):
    while True:
        for proxy in list(pool.stats.keys()):
            try:
                r = requests.get(
                    "https://httpbin.org/ip",
                    proxies={"http": proxy, "https": proxy},
                    timeout=5
                )
                if r.status_code == 200:
                    pool.report_success(proxy)
                else:
                    pool.report_blocked(proxy)
            except requests.exceptions.RequestException:
                pool.report_timeout(proxy)
        time.sleep(interval)

threading.Thread(target=healthcheck_loop, args=(pool,), daemon=True).start()

このバックグラウンドプロセスは、プールが大きい（数百のIP）場合や、パーシングが連続的でなくバッチで行われる場合に特に重要です — ヘルスチェックなしでは、一部のプロキシが実行の間に「腐って」しまい、パーシングの最初の数分が生の除外に費やされ、確認済みのアドレスで即座に作業する代わりになります。

マーケットプレイスの解析に適したプロキシの種類

プロキシの種類の選択は、除外を実行する頻度や閾値の攻撃性に直接影響します。WildberriesやOzonの安定したパーシングのためには、アンチボットが行動パターンやサブネットの評判を分析するため、居住者プロキシが最も効果的です — これらのIPは実際の家庭用インターネットプロバイダーに属し、アンチボットによってサブネットごとにブロックされることは非常に少ないです。

より単純で保護されていないソース（例えば、Avitoからの公開データの収集や簡単なショーケースの価格監視）には、データセンターのプロキシがしばしば十分です — これらはより速く、安価であり、大量のリクエストにおいて重要ですが、禁止される頻度も少し高いため、除外の閾値は厳しく設定する必要があります。

プロキシの種類	除外の速度	いつ使用するか
データセンター	高い、サブネットが一斉に禁止される	簡単なショーケース、大量、低い保護
居住者	中程度、IPが点で禁止される	Wildberries、Ozon、アンチボットサイト
モバイル	低い、IPが自動的にキャリアによって変更される	厳重に保護されたソース、高頻度のパーシング

実際には、多くのチームがハイブリッドプールを使用しています: リクエストの大部分は居住者IPを通じて行われ、大量禁止が発生した場合には、モバイルプロキシが「緊急」バックアップとして接続されます。なぜなら、モバイルプロキシのIP変更はキャリアのセルローテーションによって自然に行われ、手動の介入を必要としないからです。

結論

除外システムのないプロキシプールは必然的に劣化し、クリーンアップなしで長く運用すればするほど、パーサーは死んだIPに無駄にリソースを費やします。HTTPコード、タイムアウト、CAPTCHA、成功率を考慮したスコアリングモデルは、自動的に問題のあるアドレスを特定し、手動管理なしでプールを稼働状態に保つことを可能にします。この記事のコードは、requests、httpx、Scrapy、または独自の非同期パーサーに適応できます。

もしあなたがマーケットプレイスのパーシングのためにプールを構築していて、開始時に禁止の頻度を下げたいのであれば、居住者プロキシを検討することをお勧めします — これらはWildberriesやOzonのような保護されたプラットフォームで、通常のデータセンターアドレスに比べて明らかに高い成功率を提供します。