Решение CAPTCHA — это по сути ожидание: задача отправлена, токен приходит через 10–20 секунд. Если парсер ждёт каждую CAPTCHA по очереди, время умножается на число страниц — на тысяче URL счёт уже на часы. Ускоряет не более быстрый решатель, а модель конкурентности: отправляйте несколько задач сразу, опрашивайте параллельно, забирайте токены по готовности.
Зачем нужна пакетная обработка CAPTCHA
Каждый вызов in.php и последующий опрос res.php — это сетевой round-trip плюс 10–20 секунд решения. Если в пайплайне 500 страниц и на каждой стоит reCAPTCHA или Turnstile, последовательная обработка легко растягивается на 2–3 часа. Смена модели с «одна задача за раз» на «пакет задач сразу» не ускоряет само решение CAPTCHA, а убирает простой, в котором воркер просто ждёт ответа:
- задачи отправляются пачкой, а не по одной;
- опрос идёт по всем ID параллельно, а не последовательно;
- результат каждой задачи обрабатывается сразу по готовности, не дожидаясь остальных.
Наглядное сравнение: последовательно и параллельно
Три последовательные задачи дают около 45 секунд суммарно — каждая ждёт своей очереди. Три параллельные укладываются примерно в 15 секунд: во столько же, во сколько решается одна.
Sequential (slow):
Submit #1 → Poll → Result (15s)
Submit #2 → Poll → Result (15s)
Submit #3 → Poll → Result (15s)
Total: ~45s for 3 solves
Parallel (fast):
Submit #1 ─┐
Submit #2 ─┤→ Poll all → Results arrive
Submit #3 ─┘
Total: ~15s for 3 solves
Три способа реализовать пакетную обработку CAPTCHA
В Python для этого есть три рабочих подхода — от самого простого до самого производительного. Какой выбрать, зависит от объёма и от того, насколько вам важна простота отладки:
- ThreadPoolExecutor — минимум кода, подходит для 5–50 одновременных задач;
- asyncio + aiohttp — держит сотни задач в одном потоке, оправдан от 100+ одновременных задач;
- раздельная отправка и опрос — максимальная пропускная способность на действительно больших пакетах.
ThreadPoolExecutor: конкурентность без async/await
Самый простой способ получить конкурентность в Python — пул потоков. Задача отправляется и опрашивается в отдельном потоке, а as_completed возвращает результаты по готовности, а не по порядку отправки. Для 5–50 задач этого достаточно, и код проще, чем у asyncio.
import requests
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"
def submit_task(method, **params):
"""Submit a single CAPTCHA task."""
data = {"key": API_KEY, "method": method, "json": 1}
data.update(params)
resp = requests.post(f"{BASE_URL}/in.php", data=data, timeout=30)
result = resp.json()
if result.get("status") != 1:
raise RuntimeError(f"Submit error: {result.get('request')}")
return result["request"]
def poll_result(task_id, timeout=120):
"""Poll until result is ready."""
start = time.time()
while time.time() - start < timeout:
time.sleep(5)
resp = requests.get(f"{BASE_URL}/res.php", params={
"key": API_KEY, "action": "get",
"id": task_id, "json": 1,
}, timeout=15)
data = resp.json()
if data["request"] != "CAPCHA_NOT_READY":
return data["request"]
raise TimeoutError(f"Task {task_id} timeout")
def solve_one(sitekey, pageurl):
"""Submit and poll a single task."""
task_id = submit_task("userrecaptcha", googlekey=sitekey, pageurl=pageurl)
token = poll_result(task_id)
return {"url": pageurl, "token": token}
def batch_solve(tasks, max_workers=10):
"""Solve multiple CAPTCHAs in parallel."""
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(solve_one, t["sitekey"], t["url"]): t
for t in tasks
}
for future in as_completed(futures):
task = futures[future]
try:
result = future.result()
results.append(result)
print(f"Solved: {result['url']}")
except Exception as e:
print(f"Failed: {task['url']} - {e}")
results.append({"url": task["url"], "token": None, "error": str(e)})
return results
# Usage
tasks = [
{"sitekey": "SITE_KEY_1", "url": "https://example.com/page1"},
{"sitekey": "SITE_KEY_2", "url": "https://example.com/page2"},
{"sitekey": "SITE_KEY_3", "url": "https://example.com/page3"},
]
results = batch_solve(tasks, max_workers=5)
print(f"Solved {sum(1 for r in results if r.get('token'))}/{len(tasks)}")
asyncio и aiohttp: сотни задач в одном потоке
Когда задач больше полусотни, потоки упираются в накладные расходы ОС на переключение контекста. asyncio с aiohttp держит сотни задач в одном потоке, а конкурентность ограничивает Semaphore, а не число воркеров.
import asyncio
import aiohttp
import time
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"
async def submit_task_async(session, method, **params):
data = {"key": API_KEY, "method": method, "json": 1}
data.update(params)
async with session.post(f"{BASE_URL}/in.php", data=data) as resp:
result = await resp.json()
if result.get("status") != 1:
raise RuntimeError(f"Submit error: {result.get('request')}")
return result["request"]
async def poll_result_async(session, task_id, timeout=120):
start = time.time()
while time.time() - start < timeout:
await asyncio.sleep(5)
params = {
"key": API_KEY, "action": "get",
"id": task_id, "json": 1,
}
async with session.get(f"{BASE_URL}/res.php", params=params) as resp:
data = await resp.json()
if data["request"] != "CAPCHA_NOT_READY":
return data["request"]
raise TimeoutError(f"Task {task_id} timeout")
async def solve_one_async(session, sitekey, pageurl):
task_id = await submit_task_async(
session, "userrecaptcha",
googlekey=sitekey, pageurl=pageurl,
)
token = await poll_result_async(session, task_id)
return {"url": pageurl, "token": token}
async def batch_solve_async(tasks, max_concurrent=20):
"""Solve many CAPTCHAs concurrently with asyncio."""
semaphore = asyncio.Semaphore(max_concurrent)
results = []
async def solve_with_limit(task):
async with semaphore:
try:
result = await solve_one_async(
session, task["sitekey"], task["url"],
)
return result
except Exception as e:
return {"url": task["url"], "token": None, "error": str(e)}
async with aiohttp.ClientSession() as session:
coros = [solve_with_limit(t) for t in tasks]
results = await asyncio.gather(*coros)
return results
# Usage
tasks = [
{"sitekey": "KEY", "url": f"https://example.com/page{i}"}
for i in range(50)
]
results = asyncio.run(batch_solve_async(tasks, max_concurrent=20))
solved = sum(1 for r in results if r.get("token"))
print(f"Solved: {solved}/{len(tasks)}")
Раздельная отправка и опрос: максимальная пропускная способность
Для максимальной пропускной способности отправку и опрос стоит развести: сначала отправить весь пакет и собрать ID, затем опрашивать их одним циклом. Воркер не простаивает в ожидании конкретной задачи — опрос идёт сразу по всему пулу.
import requests
import time
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"
def batch_submit(tasks):
"""Submit all tasks first, return task IDs."""
submitted = []
for task in tasks:
try:
data = {
"key": API_KEY,
"method": "userrecaptcha",
"googlekey": task["sitekey"],
"pageurl": task["url"],
"json": 1,
}
resp = requests.post(f"{BASE_URL}/in.php", data=data, timeout=30)
result = resp.json()
if result.get("status") == 1:
submitted.append({
"task_id": result["request"],
"url": task["url"],
})
time.sleep(0.1) # Brief delay between submits
except Exception as e:
print(f"Submit failed for {task['url']}: {e}")
return submitted
def batch_poll(submitted, timeout=120):
"""Poll all submitted tasks until complete."""
pending = {s["task_id"]: s for s in submitted}
results = []
start = time.time()
while pending and time.time() - start < timeout:
time.sleep(5)
for task_id in list(pending.keys()):
try:
resp = requests.get(f"{BASE_URL}/res.php", params={
"key": API_KEY, "action": "get",
"id": task_id, "json": 1,
}, timeout=15)
data = resp.json()
if data["request"] != "CAPCHA_NOT_READY":
info = pending.pop(task_id)
results.append({
"url": info["url"],
"token": data["request"],
})
except Exception:
pass
# Mark remaining as failed
for task_id, info in pending.items():
results.append({"url": info["url"], "token": None, "error": "timeout"})
return results
# Usage
tasks = [
{"sitekey": "KEY", "url": f"https://example.com/page{i}"}
for i in range(20)
]
submitted = batch_submit(tasks)
print(f"Submitted {len(submitted)} tasks")
results = batch_poll(submitted)
solved = sum(1 for r in results if r.get("token"))
print(f"Solved: {solved}/{len(tasks)}")
Ориентиры по пропускной способности
Ориентировочная скорость по числу одновременных задач. Два момента, которые стоит учитывать при выборе числа потоков:
- фактическая цифра зависит от типа CAPTCHA — Turnstile решается быстрее, чем reCAPTCHA v2;
- рост числа задач не бесконечен — выше ~50-100 одновременных задач упираетесь в сеть, а не в CaptchaAI.
| Параллельные задачи | Приблизительная скорость | Подходит для |
|---|---|---|
| 1-5 | 3-5 решений/мин | Тестирование, лёгкий парсинг |
| 5-20 | 15-60 решений/мин | Продуктивный парсинг |
| 20-50 | 60-150 решений/мин | Пайплайны большого объёма |
| 50-100 | 150-300 решений/мин | Промышленный масштаб |
Локальный контекст: сеть, регионы и данные
Для распределённых команд из России, Беларуси, Казахстана и Украины, которые часто разворачивают парсер в европейском дата-центре или работают через мобильное соединение с нестабильным аплинком, задержка сети — не редкость, а норма. Если сервер физически далёк от ocr.captchaai.com, а канал нестабилен, увеличивайте тайм-аут опроса заранее, а не после первой волны ложных таймаутов.
Второй момент — что именно вы собираете вместе с решённой CAPTCHA. При парсинге сайтов, ориентированных на пользователей из ЕС или РФ, собирайте только те данные, которые вы вправе обрабатывать по 152-ФЗ «О персональных данных» или GDPR — это вопрос диджитал-гигиены самой команды, а не функция CaptchaAI.
Типичные проблемы при пакетном решении CAPTCHA
| Проблема | Причина | Исправление |
|---|---|---|
| Ограничение скорости (429) | Слишком много отправок в секунду | Добавить задержку 100 мс между отправками |
| Много таймаутов | Тайм-аут опроса слишком мал | Увеличить до 120–180 с |
| Падение отдачи выше 50 одновременных задач | Узкое место в сети | Использовать asyncio (aiohttp) вместо потоков |
| Результаты перепутаны | Проблема с отслеживанием ID задачи | Использовать словарь с ключом task_id |
Часто задаваемые вопросы
Сколько потоков нужно для ~1000 задач в час?
Смотрите таблицу выше: 10–20 одновременных задач дают 1800–3600 решений в час с запасом. Потоки ограничивают, сколько задач решается одновременно, а не общее число решений за месяц.
Какой тайм-аут опроса ставить при нестабильном соединении?
120 секунд обычно достаточно, но при мобильной сети или удалённом сервере увеличьте до 150–180 секунд.
Потоки или asyncio — что выбрать для батча?
Для 5–50 задач ThreadPoolExecutor проще в отладке. Со 100+ задач разница в накладных расходах заметна, и aiohttp с семафором держит нагрузку эффективнее.
Пакетная обработка увеличивает стоимость решения?
Нет. Задача тарифицируется одинаково — поодиночке или в пакете, оплата идёт за потоки. Батчинг экономит время, а не деньги.
Связанные материалы
Масштабируйте решение CAPTCHA — попробуйте CaptchaAI для высокопроизводительной пакетной обработки.
