Scarnox · Blog

Captcha snel oplossen: hoe je de latentie in je pipeline verlaagt

Een snelle captcha-solver is er een die je op het juiste moment aanroept, parallel, zonder verzoeken te verspillen. De Scarnox-oplossingsaanroep keert terug wanneer het token klaar is, dus de duur hangt af van het captchatype, de doelsite en de actuele belasting. Deze gids belooft geen vaste tijd. Hij laat zien wat je zelf in je code beheert, hoe je het meet en hoe je veel oplossingen tegelijk draait, zodat de tijd die je wel besteedt goed wordt besteed.

Wat de oplostijd bepaalt

Drie dingen zijn belangrijk, en er is er maar één die je in handen hebt:

Richt je inspanning op het derde punt. Dat is het deel dat je kunt meten en veranderen.

Meet voordat je optimaliseert

Omhul de aanroep met een timer en log het resultaat. Je wilt de oplostijd weten en de tijd die je code eromheen besteedt:

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")

Draai dit tegen de pagina's die je belangrijk vindt, op de momenten waarop je je taken draait. Bewaar de getallen. Ze vertellen je of een wijziging echt heeft geholpen.

Draai oplossingen parallel

Oplossingen zijn onafhankelijk van elkaar. Als je tien sitekeys hebt, verspil je tijd door er één voor één op te wachten, terwijl je die tijd aan de andere kunt besteden. Een threadpool is voor de meeste scripts genoeg, omdat elke oplossing één HTTP-verzoek is dat het grootste deel van zijn tijd wacht:

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)

# Begin klein. Verhoog het aantal workers alleen zolang de fouten laag blijven.
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}")

Als je bij het verhogen van het aantal workers fouten ziet die op snelheids- of capaciteitsproblemen lijken, verlaag het dan weer. Parallelliteit helpt alleen wanneer de server het kan bijhouden.

Los niet vroegtijdig op

De snelste code is de code die niet wacht op een token dat hij nog niet gebruikt. Tokens vooraf aanvragen en opslaan is de gebruikelijke fout. Tokens verlopen ongeveer twee minuten na uitgifte, dus een opgeslagen token is vaak verouderd wanneer je het nodig hebt. Los op op het moment dat het formulier op het punt staat verzonden te worden. De uitgebreide redenering staat in captchatoken verlopen: oorzaken en oplossingen.

Houd de rest van het pad kort

Hergebruik wat niet verloopt

Sommige werkzaamheden kunnen veilig over oplossingen heen worden gedeeld. Houd deze buiten het hete pad:

Deze wijzigingen kosten weinig en halen overhead weg die bij elke aanroep terugkomt.

Meet de staart, niet het gemiddelde

Een gemiddelde oplostijd verbergt de trage aanroepen die een batch ophouden. Registreer de oplostijd van elke oplossing, en kijk dan naar de mediaan en het 95e percentiel. Een pipeline die klaar is op het moment van de traagste job, wordt bepaald door de staart. Een kleine wijziging in hoe je trage aanroepen opnieuw probeert, kan dus belangrijker zijn dan elke wijziging aan de mediaan:

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]))

Vergelijk de samenvatting voor en na elke wijziging. Als het p95 nauwelijks beweegt, wordt de tijd ergens anders in je code besteed.

Controleer het resultaat, niet alleen de snelheid

Een snelle fout is nog steeds een fout. Tel de oplossingen die een fout teruggeven, en behandel een stijgend foutenpercentage als signaal om te vertragen. Mislukte oplossingen worden niet in rekening gebracht, maar de tijd die eraan wordt besteed is nog steeds tijd die je pipeline verliest. Het volledige patroon voor nieuwe pogingen staat in foutafhandeling van de captcha-solver.

Volgende stappen