Bir Prefect Flow’unda Yeniden Denemelerle CAPTCHA Nasıl Çözülür

Prefect’te bir captcha adımı, üzerinde yeniden denemeler olan tek bir task'tır ve flow'un tamamı yaklaşık yirmi satırdır. Diğer barındırılan platformlardan gelenleri şaşırtan ayrıntı hoş bir ayrıntı: Prefect Cloud kodunuzu asla çalıştırmaz. İşi zamanlar ve kendi makinenizdeki bir worker onu giden bir bağlantı üzerinden alır. Yani çözücü 127.0.0.1 üzerinde durabilir ve flow ona ulaşır; bu, Zapier veya Make.com’da geçerli değildir. Gerçekten ters giden iki şey yeniden deneme davranışı ile önbelleklemedir ve ikisi de birer argümandan ibarettir.

Neye ihtiyacınız var

• Prefect 3 ve Python 3.10 veya üstü, ayrıca CapSkip Python SDK'sı.
• Bir Prefect Cloud çalışma alanı ya da kendi barındırdığınız bir Prefect sunucusu. Bunun için ikisi de aynı şekilde çalışır.
• Korumalı formun sayfa URL’si ve sitekey'i.
• Worker ile çözücü aynı makineyi paylaştığında Local modda, paylaşmadığında Server modda çalışan CapSkip. İkisi de şurada anlatılıyor: bağlantı ayarları.

# pip install prefect
pip install -U prefect capskip

# Point the CLI at your workspace, then start a worker
# on the machine that should run the flows.
prefect cloud login

Flow'unuz gerçekte nerede çalışır

Bütün ağ kurulumunuzu belirleyen soru budur, o yüzden önce onu yanıtlayın. Prefect’in varsayılanı hibrit bir modeldir: orkestrasyon katmanı barındırılır, yürütme katmanı sizindir. Prefect Cloud meta verileri saklar ve çalıştırmaları koordine eder; kodunuzu yürütmez ve ağınıza gelen erişime ihtiyaç duymaz. Kendi altyapınızın içinde başlattığınız bir worker, iş için dışarıya yoklama yapar ve çalıştırmaları yerel olarak gönderir.

Pratik sonucu açıkça söylemekte fayda var. CapSkip ile aynı Windows makinesinde bir Process worker çalıştırın; flow, tıpkı masanızdaki bir betiğin yapacağı gibi 127.0.0.1:8080 adresini çağırır. Tünel yok, genel adres yok, sertifika yok. Bu, yalnızca buluta dayalı bir otomasyon platformundaki durumun tam tersidir ve bir orkestratörün çözüm işini koymak için rahat bir yer olmasının başlıca nedenidir.

Bir istisna var ve bir work pool seçmeden önce bunu bilmelisiniz. Prefect Managed work pool'ları flow'unuzu sizin altyapınızda değil, Prefect’in kendi altyapısında çalıştırır; bu kullanışlıdır ama loopback seçeneğini tamamen ortadan kaldırır. Bu yapılandırmada çözücünün Server moda ve erişilebilir bir adrese ihtiyacı vardır. Prefect, Managed çalıştırmaların kullandığı altı statik giden adresi yayımlar; böylece güvenlik duvarında çözücünün portuna tam olarak bunlara izin verip diğer her şeyi düşürebilirsiniz. Managed çalıştırmaların ayrıca resmi bir Prefect imajı kullanması gerekir ve 24 saatle sınırlıdır; bu da burada genellikle bir Process veya Docker work pool'unun daha iyi uyduğunun bir başka nedenidir.

Adım 1: anahtarı bir Secret block'a koyun

API anahtarını flow dosyasına sabit olarak yazmayın. Prefect bir Secret block ile gelir, değerler arka uçta beklerken şifrelenir ve bir tanesini yüklemek iki satırdır. Bir kez bir Python kabuğundan kaydedin.

# pip install prefect
from prefect.blocks.system import Secret

secret = Secret(value="YOUR_API_KEY")
secret.save("capskip-api-key")

# Rotating it later needs overwrite, or the save is refused.
# secret.save("capskip-api-key", overwrite=True)

Adım 2: çözüm task'ını yazın

Tek task, tek çözüm. Ona yeniden denemeler verin; çünkü bir çözücü çağrısı bir ağ çağrısıdır ve ağ çağrıları başarısız olur. Prefect sabit bir gecikme, bir gecikme listesi ya da üstel geri çekilme yardımcısı kabul eder; bir jitter faktörü de yeniden denemeleri yayar, böylece bir grup başarısızlık aynı anda geri dönmez.

Asıl önemli argüman diğeri. Bir CAPTCHA token'ı tek kullanımlıktır ve birkaç dakika içinde geçerliliğini yitirir; dolayısıyla asla bir önbellekten gelmemelidir. Prefect 3, sonuç kalıcılığı açık olmadıkça önbelleklemeyi kapalı tutar; bu da çoğu insanın tesadüfen güvende olduğu anlamına gelir. Ekibiniz kalıcılığı genel olarak açtıysa, ki pek çok ekip açtı, yeniden denenen bir task ilk seferde ürettiği token'ı geri verebilir ve form onu reddeder. Politikayı açıkça ayarlayın ve bunu düşünmeyi bırakın.

# pip install capskip
from prefect import task
from prefect.tasks import exponential_backoff
from prefect.cache_policies import NO_CACHE
from prefect.blocks.system import Secret
from capskip import CapSkip

# NO_CACHE matters: a token is valid once and expires fast.
@task(
    retries=3,
    retry_delay_seconds=exponential_backoff(backoff_factor=5),
    retry_jitter_factor=0.5,
    cache_policy=NO_CACHE,
)
def solve_recaptcha(sitekey: str, page_url: str) -> str:
    key = Secret.load("capskip-api-key").get()
    solver = CapSkip(host="127.0.0.1", port=8080, apiKey=key)
    return solver.recaptcha(sitekey=sitekey, url=page_url)["code"]

Tek bir metot reCAPTCHA v2, Invisible, Enterprise ve v3'ü kapsar. Varyantlar ayrı çağrılar değil, seçeneklerdir: invisible=1, enterprise=1 ya da bir action ile birlikte version="v3". Turnstile ve GeeTest'in kendi metotları vardır ve aynı biçimi alır. Tam parametre listeleri şurada: CapSkip API dokümantasyonu.

Adım 3: asla geçmeyecek hataları yeniden denemeyin

Körlemesine yeniden denemeler, deterministik başarısızlıklarda zaman harcar. Hatalı biçimlendirilmiş bir argüman her denemede aynı şekilde başarısız olur; sayfaya ait olmayan bir sitekey de öyle. Bir yeniden deneme koşulu fonksiyonu durumu alır ve karar verir; False döndürmek task'ı orijinal istisnayla birlikte hemen sonlandırır.

# Retry the transient ones. Fail fast on the rest.
from capskip import ValidationException, ApiException

def worth_retrying(task, task_run, state) -> bool:
    try:
        state.result()
    except (ValidationException, ApiException):
        return False   # bad arguments or a bad sitekey
    except Exception:
        return True    # solver down, or a timeout
    return True

Bunu task üzerinde retry_condition_fn olarak geçirin. NetworkException, CapSkip'in çalışmadığı ya da host'un yanlış olduğu anlamına gelir; TimeoutException ise çözümün, varsayılanı 300 saniye olan recaptchaTimeout'u aştığı anlamına gelir. İkisi de gerçekten bir deneme daha hak eder. Bu ikisi, ValidationException ve ApiException ile birlikte, ortak bir temelden türer; dolayısıyla başarısızlıkları tek bir yerde ele almayı tercih ederseniz CapSkipError'ı yakalamak işe yarar.

Tam çalışan örnek

Flow'un tamamı. Sayfayı alın, içinden sitekey'i çıkarın, çözün, sonra token'ı formla birlikte geri gönderin. Her adım bir task'tır; dolayısıyla her biri kendi yeniden denemelerini, kendi log'larını ve çalıştırma grafiğinde kendi girdisini alır.

# pip install prefect capskip httpx
import re
import httpx
from prefect import flow, task
from prefect.tasks import exponential_backoff
from prefect.cache_policies import NO_CACHE
from prefect.blocks.system import Secret
from capskip import CapSkip

PAGE_URL = "https://example.com/page-with-recaptcha"

@task(retries=2, retry_delay_seconds=5)
def read_sitekey(page_url: str) -> str:
    html = httpx.get(page_url, timeout=30).text
    match = re.search(r'data-sitekey=["\']([^"\']+)', html)
    if not match:
        raise RuntimeError("No data-sitekey on the page.")
    return match.group(1)

@task(
    retries=3,
    retry_delay_seconds=exponential_backoff(backoff_factor=5),
    cache_policy=NO_CACHE,
)
def solve_recaptcha(sitekey: str, page_url: str) -> str:
    key = Secret.load("capskip-api-key").get()
    solver = CapSkip(host="127.0.0.1", port=8080, apiKey=key)
    return solver.recaptcha(sitekey=sitekey, url=page_url)["code"]

@task(retries=2, cache_policy=NO_CACHE)
def submit_form(page_url: str, token: str) -> int:
    reply = httpx.post(
        page_url,
        data={"g-recaptcha-response": token},
        timeout=30,
    )
    return reply.status_code

@flow(name="captcha-protected-submit")
def run():
    sitekey = read_sitekey(PAGE_URL)
    token = solve_recaptcha(sitekey, PAGE_URL)
    return submit_form(PAGE_URL, token)

if __name__ == "__main__":
    print(run())

Çözümü gönderimden hemen önce yapın; daha erken zamanlanmış bir adımda asla. Yukarı akıştaki bir task bitene kadar on dakika bir sonuç deposunda bekleyen bir token, form onu gördüğünde çoktan ölmüştür.

Çözücüyü boğmadan toplu çözüm yapmak

Prefect, task'ları varsayılan olarak bir thread havuzu üzerinden eşzamanlı çalıştırır; dolayısıyla yüz çözüm, submit üzerinde bir liste üretecinden ibarettir ve hiçbir task runner yapılandırması gerekmez. Bu, tek bir makineye yöneltmek isteyeceğinizden muhtemelen daha fazla paralelliktir.

Kontrol aracı, global bir eşzamanlılık sınırıdır. Sınırı CLI ile bir kez oluşturun, sonra task'ın içinde bir slot işgal edin; üst sınırı aşan her çalıştırma üst üste yığılmak yerine bekler.

# Create the limit once. Six solves in flight at a time.
prefect gcl create capskip --limit 6

# The limit is enforced across every flow run, not per flow.
from prefect import flow, task
from prefect.cache_policies import NO_CACHE
from prefect.concurrency.sync import concurrency
from prefect.futures import wait
from capskip import CapSkip

@task(retries=3, cache_policy=NO_CACHE)
def solve_one(sitekey: str, page_url: str) -> str:
    with concurrency("capskip", occupy=1):
        solver = CapSkip(host="127.0.0.1", port=8080)
        return solver.recaptcha(sitekey=sitekey, url=page_url)["code"]

@flow
def solve_many(sitekey: str, urls):
    # submit, not map: map would iterate the sitekey string.
    futures = [solve_one.submit(sitekey, u) for u in urls]
    wait(futures)

Sınır, çalışma alanındaki her flow çalıştırması için geçerlidir; üç zamanlama aynı çözücüye yöneldiğinde tam olarak istediğiniz şey budur. Dağıtımı tek bir süreç içinde yapmayı tercih ederseniz, Python SDK'sının AsyncCapSkip'i gerçek bir asyncio istemcisidir ve bu yaklaşım şurada ele alınıyor: CAPTCHA'ları paralel çözme kılavuzu.

Çözücüyü başka bir makinede çalıştırmak

Worker'lar yer değiştirir. Masanızdaki bir Process worker, bir sunucuda Docker worker'a, sonra bir Kubernetes work pool'una dönüşür ve bir noktada flow artık çözücünün çalıştığı makinede olmaz. Kodda host dışında hiçbir şey değişmez.

CapSkip'in iki bağlantı modu var. Local, 127.0.0.1 adresine bağlanır ve yalnızca o cihaza yanıt verir. Server ise ağınıza veya genel IP'nize bağlanır; böylece bir VM'deki, bir konteyner sunucusundaki ya da bir Managed work pool'undaki worker, API üzerinden aynı Windows makinesini çağırır. Statik bir genel IP o adresi sabit tutar. Her iki durumda da bu hâlâ sizin donanımınızdır ve kullanım hâlâ ölçülmez; dolayısıyla yoğun bir günün maliyeti moda göre değişmez.

# Same SDK, same call. Only the host moves.
solver = CapSkip(host="10.0.0.12", port=8080, apiKey=key)

Çözücü bir ağ adresini dinlemeye başladığında anahtar doğrulamasını açın ve her worker'a kendi anahtarını verin; böylece biri, diğerlerine dokunmadan iptal edilebilir. Her iki mod da şurada adım adım anlatılıyor: CapSkip kurulum kılavuzu.

Sık görülen hatalar ve anlamları

FAQ

Kısa özet

Anahtarı bir Secret block'ta saklayın, çözümü yeniden denemeler ve NO_CACHE içeren bir task'a sarın ve onu gönderimden hemen önceki adımda çağırın. CapSkip'in bulunduğu makinede bir worker çalıştırın; host 127.0.0.1 olarak kalır. Bir grubu dağıtmadan önce global bir eşzamanlılık sınırı ekleyin. Bütün bunların Python tarafı şurada ele alınıyor: Python CAPTCHA çözücü sayfası. Aynı üç çağrı Node.js, PHP ve C#'ta da mevcuttur ve şurada listelenmiştir: CAPTCHA çözme SDK sayfası.

Bunu saatlik olarak zamanlamadan önce bilmeye değer bir şey var. CapSkip şunu sunar: captcha atlatma ve bunu zaten sahip olduğunuz donanımda yapar; dolayısıyla günde on bin çözüm yapan bir flow, on çözüm yapan bir flow ile aynı maliyettedir.