Швидке розв'язання капчі: як скоротити затримку в пайплайні
Швидкий солвер капчі — це той, який ви викликаєте у правильний момент, паралельно і без зайвих запитів. Виклик розв'язання Scarnox повертається, коли токен готовий, тож його тривалість залежить від типу капчі, цільового сайту і поточного навантаження. Цей посібник не обіцяє фіксованого часу. Він показує, що ви контролюєте у власному коді, як це виміряти і як виконувати багато розв'язань одночасно, щоб час, який ви витрачаєте, витрачався добре.
Що визначає час розв'язання
Три речі мають значення, і лише на одну з них ви можете вплинути:
- Сама капча. Деякі типи викликів займають більше часу, ніж інші. Це властивість капчі сайту, а не вашого коду.
- Навантаження на солвер. У період пікового навантаження один виклик може бути повільнішим. Логіка повторів має це обробляти, а не переривати виклик достроково.
- Ваш власний пайплайн. Розв'язання занадто рано, серіалізація кожного запиту і очікування непов'язаної роботи додають час, який можна прибрати.
Зосередьтеся на третьому пункті. Це та частина, яку ви можете виміряти і змінити.
Спершу вимірюйте, потім оптимізуйте
Обгорніть виклик таймером і записуйте результат. Вам потрібно знати час розв'язання і час, який ваш код витрачає навколо нього:
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")
Запускайте це на сторінках, які вас цікавлять, у ті години, коли працюють ваші завдання. Зберігайте результати. Вони покажуть, чи зміна справді допомогла.
Запускайте розв'язання паралельно
Розв'язання незалежні. Якщо у вас десять sitekey для обробки, очікування кожного по черзі марнує час, який можна витратити на інші. Пул потоків достатній для більшості скриптів, бо кожне розв'язання — це один HTTP-запит, який більшість часу чекає:
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}")
Якщо при збільшенні кількості воркерів з'являються помилки, схожі на проблеми з лімітами або потужністю, зменшіть її. Паралельність допомагає лише тоді, коли сервер за нею встигає.
Не розв'язуйте заздалегідь
Найшвидший код — той, що не чекає на токен, який ще не потрібен. Типова помилка — запитувати токени наперед і зберігати їх. Токени спливають приблизно через дві хвилини після видачі, тож збережений токен часто застарілий, коли він знадобився. Розв'язуйте, коли форма вже готова до відправки. Детальніше про причини описано в посібнику про прострочений токен капчі.
Тримайте решту шляху короткою
- Тримайте одну HTTP-сесію на сайт, щоб не перепідключатися при кожному запиті.
- Не запускайте повільний парсинг або роботу з базою даних між розв'язанням і відправкою.
- Використовуйте розумний таймаут для власних запитів. Виклик розв'язання займає близько 185 секунд, бо блокується, доки токен не буде готовий. Ваші інші виклики мають бути значно коротшими.
- Надсилайте проксі лише тоді, коли завдання його потребує. Типи завдань із проксі приймають поле
proxyу форматіhttp://user:pass@host:port, але сам по собі проксі не прискорює роботу.
Використовуйте повторно те, що не спливає
Частину роботи можна безпечно ділити між розв'язаннями. Тримайте їх поза гарячим шляхом:
- Sitekey. Прочитайте sitekey один раз для шаблону сторінки і збережіть його. Він не спливає, як токен. Перечитуйте лише тоді, коли форма починає збоїти.
- HTTP-з'єднання. Сесія з keep-alive уникає нового TLS-рукостискання при кожному виклику. Створюйте одну сесію на потік або воркер, а не на запит.
- Розібрана конфігурація. Завантажуйте список цілей, список проксі і налаштування один раз під час старту, а не читайте файли всередині функції розв'язання.
Ці зміни коштують мало і прибирають накладні витрати, які з'являються в кожному виклику.
Вимірюйте хвіст, а не середнє
Середній час розв'язання приховує повільні виклики, які затримують пакет. Записуйте час кожного розв'язання, а потім дивіться на медіану і 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 майже не рухається, час витрачається деінде у вашому коді.
Перевіряйте результат, а не лише швидкість
Швидка помилка все одно є помилкою. Рахуйте розв'язання, які повертають помилку, і вважайте зростання частки помилок сигналом сповільнитися. Невдалі розв'язання не тарифікуються, але час, витрачений на них, — це час, який втратив ваш пайплайн. Повний шаблон повторів описано в обробці помилок солвера капчі.