captcha解決を高速化：パイプラインの遅延を減らす方法
速いcaptchaソルバーとは、適切なタイミングで、並列に、無駄なリクエストをせずに呼び出せるソルバーのことです。Scarnoxの解決呼び出しは、トークンの準備ができると戻るので、その所要時間はcaptchaの種類、対象サイト、そのときの負荷によって変わります。このガイドは、固定の時間を約束しません。自分のコードで制御できる部分、それを計測する方法、そして多くの解決を同時に行う方法を示します。そうすれば、実際に使う時間を有効に使えます。
解決時間を決めるもの
重要なのは3つで、そのうち自分の手で変えられるのは1つだけです。
- captchaそのもの。 チャレンジの種類によって、かかる時間は異なります。これはサイトのcaptchaの性質であり、あなたのコードの問題ではありません。
- ソルバー側の負荷。 混雑した時間帯は、1回の呼び出しが遅くなることがあります。リトライのロジックは、呼び出しを早々に諦めるのではなく、その状況に対処するべきです。
- 自分のパイプライン。 必要以上に早く解決したり、すべてのリクエストを直列に処理したり、無関係な作業を待ったりすると、時間が無駄になります。
手をかけるべきは3つ目です。それは計測して変えられる部分だからです。
最適化の前に計測する
呼び出しにタイマーを付けて、結果をログに残してください。知りたいのは、解決時間と、その前後にコードが費やす時間です。
import time
import requests
API_KEY = "scarnox_YOURAPIKEY"
URL = "https://api.scarnox.com/api/tasks/create"
def timed_solve(sitekey: str, page_url: str) -> dict:
started = time.perf_counter()
r = requests.post(
URL,
headers={"Authorization": f"Bearer {API_KEY}"},
json={
"captcha_type": "PopularCaptchaTaskProxyless",
"site_key": sitekey,
"site_url": page_url,
},
timeout=185,
)
elapsed = time.perf_counter() - started
data = r.json()
if data.get("errorId") != 0:
return {"ok": False, "error": data.get("errorCode"), "seconds": elapsed}
return {"ok": True, "token": data["solution"]["token"], "seconds": elapsed}
result = timed_solve("10000000-ffff-ffff-ffff-000000000001", "https://example.com/signup")
print(result["ok"], round(result["seconds"], 2), "s")計測したいページに対して、ジョブを実行する時間帯に実行してください。数値を残しておきましょう。変更が本当に役立ったかどうかが分かります。
並列に解決する
解決はそれぞれ独立しています。10個のサイトキーを扱うなら、1つずつ待つと、他の処理に使えたはずの時間が無駄になります。各解決はほとんどの時間を待ち時間に費やすHTTPリクエストが1回なので、ほとんどのスクリプトにはスレッドプールで十分です。
from concurrent.futures import ThreadPoolExecutor
jobs = [
("10000000-ffff-ffff-ffff-000000000001", "https://example.com/signup"),
("10000000-ffff-ffff-ffff-000000000001", "https://example.com/register"),
("10000000-ffff-ffff-ffff-000000000001", "https://example.com/contact"),
]
def run(job):
sitekey, page_url = job
return timed_solve(sitekey, page_url)
# 少なめに始める。エラーが少ない間だけワーカー数を増やす。
with ThreadPoolExecutor(max_workers=5) as pool:
results = list(pool.map(run, jobs))
for r in results:
status = "ok" if r["ok"] else r["error"]
print(f"{r['seconds']:.2f} s {status}")ワーカー数を増やしたときに、レート制限や処理能力の問題に見えるエラーが出たら、減らしてください。並列化が役立つのは、サーバーがそれに追いつけるときだけです。
早すぎる解決をしない
最速のコードは、まだ使わないトークンを待たないコードです。トークンを事前にまとめて取得して保存しておくのは、よくある間違いです。トークンは発行から約2分で期限切れになるため、保存したトークンは必要になったときには古くなっていることが多いです。フォームを投稿する直前に解決してください。詳しい理由はcaptchaトークンの期限切れ：原因と対処方法で説明しています。
残りの経路を短く保つ
- サイトごとにHTTPセッションを1つ保ち、リクエストごとに再接続しないようにする。
- 解決と投稿の間に、遅いパース処理やデータベースの処理を入れない。
- 自分のリクエストには適切なタイムアウトを設定する。解決の呼び出しはトークンの準備ができるまでブロックするので約185秒を使いますが、他の呼び出しはずっと短くするべきです。
- プロキシはタスクが必要とするときだけ送る。プロキシのタスクの種類は
proxyフィールドをhttp://user:pass@host:portの形式で受け取りますが、プロキシを追加するだけで速くなるわけではありません。
期限のないものは再利用する
一部の作業は、複数の解決の間で安全に共有できます。次のものはホットパスから外してください。
- サイトキー。 ページのテンプレートごとに一度読み取って保存します。トークンのように期限切れにはなりません。フォームが失敗し始めたときだけ再度読み取ってください。
- HTTP接続。 キープアライブのセッションを使えば、呼び出しごとのTLSハンドシェイクを避けられます。リクエストごとではなく、スレッドやワーカーごとに1つのセッションを作ってください。
- 解析済みの設定。 対象リスト、プロキシリスト、設定は起動時に一度だけ読み込み、解決の関数の中でファイルを読まないようにしてください。
これらの変更はコストがほとんどかからず、すべての呼び出しに現れるオーバーヘッドを取り除きます。
平均ではなく末尾を計測する
解決時間の平均は、バッチを遅らせる遅い呼び出しを隠してしまいます。解決ごとの時間を記録して、中央値と95パーセンタイルを見てください。最も遅いジョブの時間で終わるパイプラインは、末尾に制限されています。そのため、リトライの遅い呼び出しの扱いを少し変えるだけで、中央値をいじるよりも効果があることがあります。
import statistics
def summarise(seconds: list[float]) -> str:
ordered = sorted(seconds)
p95 = ordered[int(0.95 * (len(ordered) - 1))]
return f"median {statistics.median(ordered):.2f} s, p95 {p95:.2f} s, n={len(ordered)}"
print(summarise([r["seconds"] for r in results]))変更の前後で集計を比べてください。p95がほとんど動かないなら、時間はコードの別の場所で使われています。
速さだけでなく結果を確認する
速く失敗しても、それは失敗です。エラーを返した解決の数を数え、エラー率の上昇は速度を落とすべきサインとして扱ってください。失敗した解決は課金されませんが、それにかけた時間は、パイプラインが失った時間です。完全なリトライのパターンはcaptcha解決のエラー処理にあります。