Scarnox · Blog

captcha解決のエラー処理：リトライ、バックオフ、再試行してはいけない失敗

captcha解決は、管理できない外部システムへのネットワーク呼び出しであり、しかもそのチャレンジは予告なく変わることがあります。失敗をうまく扱えるかどうかで、パイプラインが穏やかに劣化するか、リトライの嵐で時間を浪費するかが決まります。

Scarnoxでは、経済面で助けになります。失敗した解決は無料なので、再試行にかかるのは時間だけです。ただし、再試行すべきでない失敗を再試行すると、やはり時間を無駄にします。さらに、激しく再試行しすぎると、別の問題を引き起こすこともあります。このガイドでは、Pythonジョブにそのまま貼り付けられるリトライ方針を示します。

Scarnoxは現在、PopularCaptchaを解決します。同じエラー処理はどのcaptchaソルバーにも当てはまるので、以下の方針はcaptchaの種類に依存しません。

必ず最初にerrorIdを確認する

解決のレスポンスにはerrorIdフィールドがあります。0は成功を意味し、トークンはsolution.tokenにあります。それ以外の値は解決の失敗を意味します。失敗したレスポンスにはerrorCodeとerrorDescriptionも含まれます。両方をログに残してください。後で検索するのはコードで、実行がおかしくなったときに読むのは説明です。

{"errorId": 1, "errorCode": "...", "errorDescription": "..."}

HTTPステータスが200だからといって、成功と決めつけないでください。毎回errorIdを確認してください。

再試行すべき失敗

一時的な失敗もあります。これらは指数バックオフと小さなランダム遅延を付けて再試行してください。

これらの失敗は、待つだけで次の試行が成功することがあるため、再試行する価値があります。

再試行の前に修正すべき失敗

他の失敗は、待っても解消されません。同じ入力で再試行すると、同じように失敗します。先に修正してください。

どちらのリストにも入らない失敗は、修正が必要な失敗として扱ってください。エラーの文言を見て判断し、そのコードを適切なリストに追加します。リストは短く、定期的にレビューしてください。

そのまま貼り付けられる方針

import os
import random
import time

import requests

API_URL = "https://api.scarnox.com/api/tasks/create"
HEADERS = {"Authorization": f"Bearer {os.environ['SCARNOX_KEY']}"}
MAX_ATTEMPTS = 4

class SolveError(Exception):
    def __init__(self, code: str, description: str, retryable: bool):
        super().__init__(f"{code}: {description}")
        self.code = code
        self.retryable = retryable

def solve_once(payload: dict) -> str:
    try:
        r = requests.post(API_URL, headers=HEADERS, json=payload, timeout=185)
    except requests.RequestException as exc:
        # こちら側のネットワークエラーまたはタイムアウト：再試行して安全
        raise SolveError("NETWORK", str(exc), retryable=True)

    if r.status_code == 429 or r.status_code >= 500:
        raise SolveError(f"HTTP_{r.status_code}", r.text[:200], retryable=True)

    try:
        data = r.json()
    except ValueError:
        raise SolveError(f"HTTP_{r.status_code}", r.text[:200], retryable=False)

    if data.get("errorId") == 0:
        return data["solution"]["token"]

    # APIが失敗を報告した。ログで一時的と確認できたコードだけ再試行する。
    code = data.get("errorCode", "UNKNOWN")
    retryable = code in TRANSIENT_CODES
    raise SolveError(code, data.get("errorDescription", ""), retryable=retryable)

# ログで一時的なものと確認できたAPIエラーコードだけをここに追加する。
TRANSIENT_CODES: set[str] = set()

def solve(payload: dict) -> str:
    for attempt in range(MAX_ATTEMPTS):
        try:
            return solve_once(payload)
        except SolveError as err:
            last_attempt = attempt == MAX_ATTEMPTS - 1
            if not err.retryable or last_attempt:
                raise
            delay = min(30, 2 ** attempt) + random.random()  # バックオフ + ジッター
            print(f"attempt {attempt + 1} failed ({err.code}), retrying in {delay:.1f}s")
            time.sleep(delay)
    raise RuntimeError("unreachable")

この方針は、修正が必要な失敗を再試行せず、試行回数を固定して止まります。TRANSIENT_CODESの集合は意図的に空で始まっています。ログで一時的であると確認できたコードだけを追加してください。

ジッターが重要な理由

50個のワーカーが同時に失敗し、全員が正確に2秒ずつ待つと、全員が同時に再試行して、また同時に失敗します。わずか数分の1秒のランダムな遅延を入れると、再試行が散らばります。これは、大手のAPIプロバイダーが自社のクライアントに勧めている方法と同じです。

タイムアウト

HTTPのタイムアウトは、通常の解決時間よりずっと長く、約185秒に設定してください。リクエストはトークンの準備ができるまでブロックされるので、クライアント側のタイムアウトが短いと、遅れて成功したはずのものを失敗として扱い、その後の再試行が無駄になります。解決時間より長いクライアントのタイムアウトが安全な設定です。

トークンはキャッシュできない

トークンは約2分で期限切れになります。実行の間でトークンをキャッシュしたり、2回目の送信に再利用したりしないでください。送信がトークンを使った後に失敗したら、古いトークンを再送するのではなく、もう一度解決してください。詳しくはトークンの期限切れのガイドを参照してください。

ワーカー間の同時実行とバックオフ

多くのワーカーが1つのキーを共有する場合、1つのワーカーのバックオフは他のワーカーを遅くしません。セマフォやキューなどで同時解決数に共有の上限を設けて、リトライの集中が大量のリクエストに変わらないようにしてください。単純な同時実行数の上限で、レート制限の失敗の多くは発生前に防げます。

役立つログ

失敗した試行ごとに、errorCode、試行番号、サイトキー、ページのホスト、そしてリトライしたかどうかを記録してください。APIキーやプロキシのパスワードはログに出さないでください。時間ごとにコードを集計するダッシュボードがあれば、問題がコード、プロキシ、サービスのどこにあるかを素早く判断できます。

次のステップ

方針を無料トライアルでテストしてください。アカウントを作成すると、無料のトライアルクレジットで始められます。カード登録は不要です。