Temporal Workflow Activity’sinde CAPTCHA Nasıl Çözülür

Temporal’da bir captcha çözümü Activity içine girer. Workflow metoduna değil, workflow’un çağırdığı bir yardımcıya değil, bir Activity’ye. Workflow kodu, workflow her devam ettiğinde geçmişten yeniden oynatılır; bu yüzden deterministik olmak zorundadır: ağ çağrısı yok, rastgelelik yok, saat okuması yok. Bir çözüm bu üçünü aynı anda yapar. Bir kez Activity içine girdikten sonra geriye bir yeniden deneme politikası ve bir parça zamanlama disiplini kalır; hepsi topu topu kırk satır.

Neye ihtiyacınız var

• Python 3.10 veya üzeri, Temporal Python SDK ve CapSkip Python SDK.
• Bağlanılacak bir Temporal Service. Yerel bir dev server da Temporal Cloud da burada işe yarar.
• Korumalı formun sayfa URL’si ve sitekey'i.
• Worker ile çözücü aynı makineyi paylaşıyorsa Local modda, paylaşmıyorsa Server modda CapSkip. Her ikisi de şurada anlatılıyor: bağlantı ayarları.

# pip install temporalio
pip install -U temporalio capskip httpx

# A local service to develop against.
temporal server start-dev

Çözüm neden workflow kodunda duramaz

Temporal, bir worker yeniden başlatmasından, bir deploy’dan ya da bir haftalık uykudan sonra durumu yeniden kurmak için workflow’un geçmişini yeniden oynatır. Bunun iki kez aynı sonucu vermesi için workflow kodunun deterministik olması gerekir. SDK bunun neleri dışarıda bıraktığını açıkça söyler: ağ IO’su yok, threading yok, rastgelelik yok, süreçlere dış çağrı yok, global durum değişikliği yok. Hatta workflow kodunu, her çalıştırmada modülleri yeniden import eden bir sandbox içinde çalıştırır; activity import’larının bir geçiş bloğuna sarılmasının nedeni budur.

Bir CAPTCHA çözümü kuralı üç kez birden çiğner. Bir ağ çağrısıdır, döndürdüğü token her denemede farklıdır ve ne kadar sürdüğü makineye bağlıdır. Bunu workflow metoduna koyarsanız geliştirmede çalışıyor gibi görünür, sonra bir worker çalışma ortasında ilk kez yeniden başladığında bir determinizm hatası üretir.

İyi haber şu ki bu kısıt size bir şey kazandırıyor. Bir Activity, yazdığınız bir döngüyle değil, tanımladığınız bir politikayla Temporal’ın kendisi tarafından yeniden denenir ve sonucu geçmişe kaydedilir. Yani başarılı olan bir çözüm, yeniden oynatmada asla tekrarlanmaz; tek kullanımlık bir cevabı olan bir şey için tam da istediğiniz şey budur.

Adım 1: çözüm activity’si

Python’ın CapSkip SDK’sı takma ad değil, gerçek bir asyncio istemcisiyle gelir; bu yüzden asenkron bir activity doğal seçimdir ve thread havuzu gerektirmez. sitekey ile sayfa URL’sini argüman olarak alın ve token’ı döndürün.

# pip install capskip
import os
from temporalio import activity
from capskip import AsyncCapSkip

@activity.defn
async def solve_recaptcha(sitekey: str, page_url: str) -> str:
    solver = AsyncCapSkip(
        host=os.environ.get("CAPSKIP_HOST", "127.0.0.1"),
        port=8080,
        apiKey=os.environ.get("CAPSKIP_API_KEY", "capskip"),
    )
    result = await solver.recaptcha(sitekey=sitekey, url=page_url)
    return result["code"]

Tek bir metot reCAPTCHA v2, Invisible, Enterprise ve v3’ü kapsar. Varyantlar ayrı metotlar değil, aynı çağrı üzerindeki seçeneklerdir: invisible değeri 1, enterprise değeri 1 ya da version değeri v3 artı bir action. Turnstile ve GeeTest’in aynı biçimde kendi metotları vardır ve tam parametre listesi şurada: CapSkip API dokümantasyonu.

Senkron istemciyi tercih ederseniz activity’nin düz bir def olması ve worker’ın bir activity_executor alması gerekir, çünkü Temporal senkron activity’leri bir thread havuzunda çalıştırır. Yukarıdaki asenkron sürüm bundan tamamen kaçınır; bu da Python SDK’sının diğerlerinden gerçekten daha hoş olduğu birkaç yerden biri.

Adım 2: çözücünün kendi süresinden daha uzun zaman aşımları

Her activity bir start_to_close_timeout ister ve insanlar kendi çözümlerini sessizce burada bozar. CapSkip reCAPTCHA, Turnstile ve GeeTest’te 300 saniyeye kadar, görüntü CAPTCHA’larında ise 120 saniyeye kadar yoklama yapar. Activity zaman aşımını bunun altına ayarlarsanız Temporal, çözücü hâlâ çalışırken denemeyi iptal eder, sonra yeniden dener ve tek bir form için havada iki çözümünüz olur.

Activity’ye çözücünün kendi sınırının üzerinde pay bırakın. Beş dakikalık bir çözücü tavanına karşı altı dakika rahattır.

# Longer than recaptchaTimeout, which defaults to 300s.
from datetime import timedelta
from temporalio.common import RetryPolicy

SOLVE_TIMEOUT = timedelta(minutes=6)

SOLVE_RETRIES = RetryPolicy(
    initial_interval=timedelta(seconds=5),
    backoff_coefficient=2.0,
    maximum_attempts=4,
    # These fail identically every time. Do not burn attempts.
    non_retryable_error_types=["ValidationException", "ApiException"],
)

Yeniden denenmeyecekler listesi istisna sınıfı adına göre eşleşir ve doldurmaya değer. ValidationException eksik ya da bozuk bir argüman demektir; ApiException ise API’nin isteği reddettiği, genellikle sayfa URL’sine ait olmayan bir sitekey demektir. İkisi de ikinci denemede düzelmez. Gerçekten yeniden denemeyi hak eden ikisi NetworkException ve TimeoutException: ilki çözücünün çalışmadığı ya da host’un yanlış olduğu, ikincisi çözümün yoklama zaman aşımını aştığı anlamına gelir. Dördü de ortak bir tabandan türer, yani her şeyi tek bir yerde ele almayı tercih ederseniz CapSkipError’ı yakalamak işinizi görür.

Adım 3: çözümü ilk değil, son yapın

Kalıcı yürütme, token süresinin dolmasını burada başka herhangi bir orkestratördekinden daha kolay yanlış yapmanıza yol açar. Bir Temporal workflow’u bir sinyal bekleyebilir, bir gün uyuyup devam edebilir ve kaydedilmiş activity sonuçları geçmişten değişmeden geri gelir. Yani erken çözen, bir onay bekleyen, sonra gönderen bir workflow, dün üretilmiş bir token’ı yeniden oynatır.

Bir reCAPTCHA token’ı bir kez kabul edilir ve yaklaşık iki dakikada geçerliliğini yitirir. Workflow’u, çözüm gönderimden hemen önceki adım olacak ve arada bloke edebilecek hiçbir şey kalmayacak şekilde sıralayın. Bu sorunun genel biçimi şu rehberde anlatılıyor: reCAPTCHA token süresinin dolması.

Tam çalışan örnek

Üç activity ve bunları sırayla çağıran bir workflow. sitekey’i okuyun, çözün, gönderin. Her activity kendi zaman aşımını ve kendi yeniden deneme politikasını alır ve her biri Temporal arayüzünde ayrı ayrı görünür; böylece bir şey yavaşladığında hangi adım olduğunu görebilirsiniz.

# activities.py
import os, re, httpx
from temporalio import activity
from capskip import AsyncCapSkip

@activity.defn
async def read_sitekey(page_url: str) -> str:
    async with httpx.AsyncClient(timeout=30) as client:
        html = (await client.get(page_url)).text
    found = re.search(r'data-sitekey=["\']([^"\']+)', html)
    if not found:
        raise RuntimeError("No data-sitekey on the page.")
    return found.group(1)

@activity.defn
async def solve_recaptcha(sitekey: str, page_url: str) -> str:
    solver = AsyncCapSkip(host=os.environ.get("CAPSKIP_HOST", "127.0.0.1"))
    return (await solver.recaptcha(sitekey=sitekey, url=page_url))["code"]

@activity.defn
async def submit_form(page_url: str, token: str) -> int:
    async with httpx.AsyncClient(timeout=30) as client:
        reply = await client.post(
            page_url, data={"g-recaptcha-response": token}
        )
    return reply.status_code

Workflow’un kendisi sıralamanın ötesinde hiçbir mantık taşımaz. Mesele de bu: başarısız olabilecek her şey bir activity içindedir ve workflow, yeniden oynatmadan sağ çıkan deterministik kısımdır.

# workflow.py
from datetime import timedelta
from temporalio import workflow
from temporalio.common import RetryPolicy

with workflow.unsafe.imports_passed_through():
    from activities import read_sitekey, solve_recaptcha, submit_form

@workflow.defn
class SubmitProtectedForm:
    @workflow.run
    async def run(self, page_url: str) -> int:
        sitekey = await workflow.execute_activity(
            read_sitekey, page_url,
            start_to_close_timeout=timedelta(seconds=60),
        )
        # Solve directly before the submit. Tokens go stale.
        token = await workflow.execute_activity(
            solve_recaptcha, args=[sitekey, page_url],
            start_to_close_timeout=timedelta(minutes=6),
            retry_policy=RetryPolicy(
                maximum_attempts=4,
                non_retryable_error_types=["ValidationException"],
            ),
        )
        return await workflow.execute_activity(
            submit_form, args=[page_url, token],
            start_to_close_timeout=timedelta(seconds=60),
        )

İki argümanlı activity’lerdeki args listesine dikkat edin. Tek bir konumsal argüman doğrudan geçirilebilir, ama birden fazlası args üzerinden gitmek zorundadır ve bunu yanlış yapmak bu SDK’da en sık görülen ilk hatadır.

# worker.py - run this where CapSkip can be reached
import asyncio
from temporalio.client import Client
from temporalio.worker import Worker
from activities import read_sitekey, solve_recaptcha, submit_form
from workflow import SubmitProtectedForm

async def main():
    client = await Client.connect("localhost:7233")
    worker = Worker(
        client,
        task_queue="captcha-queue",
        workflows=[SubmitProtectedForm],
        activities=[read_sitekey, solve_recaptcha, submit_form],
    )
    await worker.run()

asyncio.run(main())

Worker nerede çalışır, çözücü nerede çalışır

Temporal ikisini net biçimde ayırır ve bu sizin lehinize çalışır. Temporal Service işi zamanlar ve geçmişi saklar. Kodunuzu asla çalıştırmaz. Kendi altyapınızda başlattığınız bir worker, servise uzun süreli giden bir bağlantı tutar ve kuyruktan task alır. Ağınıza asla gelen bir bağlantı açılmaz.

Yani CapSkip ile aynı Windows makinesindeki bir worker, tıpkı masanızdaki bir betiğin yapacağı gibi 127.0.0.1:8080 adresini çağırır ve bu Temporal Cloud’da da geçerli kalır. Temporal Cloud’daki bulut, hesaplama katmanı değil, orkestrasyon katmanıdır.

Worker yer değiştirdiği anda host değişir, başka hiçbir şey değişmez. Bir konteynerdeki, bir Linux VM’deki ya da Kubernetes’teki bir worker, bir Windows çözücüye loopback üzerinden ulaşamaz; bu yüzden çözücü Server moda geçer. Local mod 127.0.0.1’e bağlanır ve yalnızca o cihaza yanıt verir. Server mod ağ adresinize ya da genel IP’nize bağlanır ve statik bir genel IP adresi sabit tutar. Her iki modda da donanım hâlâ sizin ve hâlâ sayaçsız, yani günde on bin kez çalışan bir workflow, bir kez çalışanla aynı maliyete sahiptir.

# One environment variable, no code change.
# CAPSKIP_HOST=10.0.0.12 on the worker.
solver = AsyncCapSkip(host=os.environ["CAPSKIP_HOST"], port=8080)

Çözücü bir ağ adresini dinlemeye başladığında anahtar doğrulamasını açın ve her worker filosuna 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 rehberi.

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

FAQ

Kısa özet

Çözümü bir activity’ye koyun, asla workflow metoduna değil. Ona çözücünün kendi 300 saniyelik tavanından daha uzun bir start_to_close_timeout verin, ValidationException ve ApiException’ı yeniden denenmeyecek olarak işaretleyin ve workflow’u, çözüm gönderimden önceki son şey olacak şekilde sıralayın. Worker’ı CapSkip’in olduğu yerde çalıştırın, host 127.0.0.1 olarak kalsın. Python yüzeyinin geri kalanı şurada: Python CAPTCHA çözücü sayfası, aynı üç çağrı Node.js, PHP ve C# için de var, şurada listelendiği gibi: CAPTCHA çözme SDK sayfası. reCAPTCHA seçeneklerinin kendisi ise şurada: reCAPTCHA v2 çözücü sayfası.

Buna bir zamanlama yöneltmeden önce bilinmeye değer bir şey. CapSkip, zaten sahip olduğunuz donanımda çalışan bir captcha çözücü olduğundan, her dakika tetiklenen bir workflow, haftada bir kez tetiklenenle aynı maliyete sahiptir.