Apache Airflow DAG'lerinde CAPTCHA Nasıl Çözülür (Python)

Bir Airflow captcha adımı sıradan bir Python görevidir. Çözücüyü çağırırsınız, bir token geri alırsınız, onu aynı görevde kullanırsınız. Kurulacak bir operator ya da yazılacak bir plugin yoktur. İnsanları asıl tökezleten şey konumdur: worker'larınız zamanlayıcının onları koyduğu yerde çalışır ve dizüstünüzün loopback adresine bağlı bir çözücüye başka bir host'taki container'dan erişilemez. Önce bu kısmı doğru yapın; sonrasında DAG on beş satırdır.

Neye ihtiyacınız var

• CapSkip SDK'sı, worker'larınızın kullandığı imajın veya virtualenv'in içine kurulmuş olacak şekilde Airflow 2.x veya 3.x.
• Doğrulama döndüren bir hedef. Her zaman geçen bir test anahtarı değil, bir widget'ın arkasındaki bir sayfa.
• Worker'larınızın erişebileceği bir makinede Server modunda, ya da worker ile çözücü aynı makineyse Local modda çalışan bir CapSkip. İkisi de şurada açıklanmıştır: bağlantı ayarları.

# Install into the worker environment, not just the scheduler.
pip install capskip

Çözücü nerede durmalı

Bütün problem bu, o yüzden en başa koyuyoruz. Bir görev bir worker sürecinin içinde çalışır ve gerçek kurulumların çoğunda o worker, elle yönettiğiniz her şeyden farklı bir host üzerindeki bir container'dır. O container'ın içindeki loopback adresi container'ın kendisidir; dolayısıyla SDK'yı oraya yöneltmek hiçbir şey bulmaz.

Neredeyse her Airflow kurulumunun cevabı Server modudur. Uygulamada dinleme adresini değiştirir, SDK host'unu o makineye yöneltirsiniz ve filodaki her worker tek bir çözücüyü paylaşır. Çağıranlar kendi ağınızın dışındaysa statik bir genel IP önerilir. Bu, ürünün ne olduğunu değiştirmez: hâlâ sizin donanımınızdır ve kullanım hâlâ ölçülmez; yani onu loopback adresinden çıkarmak yalnızca nerede çalıştığını değiştirir, başka hiçbir şeyi değil. CapSkip bir Windows uygulamasıdır, dolayısıyla pratikte bu, filonun bağlandığı tek bir Windows makinesi demektir.

Adım 1: host'u bir kez yapılandırın

Adresi DAG dosyasına sabit kodlamayın. SDK, CAPSKIP_HOST, CAPSKIP_PORT ve CAPSKIP_API_KEY değerlerini ortamdan okur; worker ortam değişkenlerini ayarlamanın bir yolu zaten elinizde olduğu için Airflow’a en temiz oturan yöntem budur. Değeri metadata veritabanında tutmayı tercih ederseniz bir Airflow Variable de işinizi görür.

# pip install capskip
import os
from capskip import CapSkip


def get_solver():
    """One place that knows where the solver lives."""
    return CapSkip(
        host=os.environ.get("CAPSKIP_HOST", "127.0.0.1"),
        port=int(os.environ.get("CAPSKIP_PORT", "8080")),
        recaptchaTimeout=300,
    )

İstemciyi modül kapsamında değil, görevin içinde oluşturun. Airflow her DAG dosyasını bir döngü içinde ayrıştırır ve import zamanında oluşturulan her şey, worker'da olduğu gibi zamanlayıcıda da her ayrıştırmada yeniden oluşturulur.

Adım 2: çözümü tek bir görev olarak yazın

En kısa yol TaskFlow dekoratörleridir. Airflow 3 bunları SDK modülünden, Airflow 2 ise decorators modülünden içe aktarır; gövde her iki durumda da aynıdır.

# Airflow 3.x. On 2.x use: from airflow.decorators import dag, task
from airflow.sdk import dag, task

PAGE = "https://example.com/page-with-recaptcha"
SITEKEY = "YOUR_SITEKEY"


@task(retries=2)
def fetch_protected_page():
    solver = get_solver()

    # Solve and use in the same task. The token is short lived.
    token = solver.recaptcha(sitekey=SITEKEY, url=PAGE)["code"]

    return post_form(PAGE, token)

Entegrasyonun tamamı bu. Çözücü çağrısı senkrondur, çeyrek saniyeden başlayan bir geri çekilmeyle sizin yerinize yoklama yapar ve code alanında token'ı tutan bir sözlük döndürür.

Adım 3: token'ı görevler arasında geçirmeyin

Adını koymaya değer hata budur, çünkü Airflow onu doğal hissettiriyor. TaskFlow dönüş değerleri XCom'a dönüşür; dolayısıyla token döndüren bir çözüm görevi ile onu tüketen bir alt görev iyi tasarım gibi görünür. Bu bir yarıştır. Bir reCAPTCHA token'ı kabaca iki dakika geçerli kalır ve iki Airflow görevi arasındaki boşluk, sizin kontrol etmediğiniz bir zamanlayıcı kararıdır. Üstüne bir kuyruk gecikmesi, dolu bir havuz ya da bir worker yeniden başlatması ekleyin; token yolda sona erer.

Arıza aralıklıdır, yani en kötü türden: DAG testte çalışır, üretimde ise zamanın yüzde birkaçında başarısız olur; form reddedilir ve çözücüden hiçbir hata gelmez. Bir reCAPTCHA token'ının ne kadar süre geçerli kaldığına dair yazı zamanlamaları ele alıyor. Bir DAG'de kural tek satıra iner: aynı görevin içinde çözün ve gönderin, yeniden çözme işini de yeniden denemeye bırakın.

DAG'in tamamı

# pip install capskip
import os
from datetime import datetime, timedelta

import requests
from airflow.sdk import dag, task
from capskip import CapSkip

PAGE = "https://example.com/page-with-recaptcha"
SITEKEY = "YOUR_SITEKEY"


@dag(
    schedule="@hourly",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["scraping"],
)
def protected_source():

    @task(retries=2, retry_delay=timedelta(minutes=2), pool="captcha")
    def scrape():
        solver = CapSkip(
            host=os.environ.get("CAPSKIP_HOST", "127.0.0.1"),
            port=int(os.environ.get("CAPSKIP_PORT", "8080")),
        )
        token = solver.recaptcha(sitekey=SITEKEY, url=PAGE)["code"]

        # Same task, so the token is seconds old when it is used.
        r = requests.post(
            PAGE,
            data={"g-recaptcha-response": token},
            timeout=60,
        )
        r.raise_for_status()
        return len(r.text)

    scrape()


protected_source()

Oradaki pool argümanı gerçek bir iş yapıyor. Bir havuz, kurulumun tamamında aynı anda kaç görev örneğinin çalışacağını sınırlar; böylece iki yüz çalıştırmalık bir backfill, tek bir makineye karşı iki yüz eşzamanlı çözüm açmaz. Arayüzde captcha adında bir havuz oluşturun, ona istediğiniz paralel çözüm sayısını verin; o havuzu adıyla belirten her görev bu sınırın arkasında kuyruğa girer.

Zarar değil fayda getiren yeniden denemeler

Görevde retries değerini ayarlayın ve çözüm ile gönderimin tamamının tekrarlanmasına izin verin. Token görev gövdesinin içinde alındığı için, bir yeniden deneme otomatik olarak taze bir tane alır; istediğiniz davranış budur ve iki adımın birlikte kalmasının nedeni de budur.

Yeniden denemeye bir gecikme verin. Sizi az önce doğrulamaya sokmuş bir siteye karşı anında yapılan bir deneme genelde yine doğrulamaya takılır ve zamanlanmış bir pipeline'da birkaç dakikanın hiçbir maliyeti yoktur. İki yeniden deneme genellikle yeter: üçüncü bir başarısızlık normalde yanlış bir sitekey ya da erişilemeyen bir çözücüdür ve bunların hiçbiri beklemekle düzelmez.

Yaygın hatalar

Kodların tam listesi ve her birini neyin tetiklediği şurada: CapSkip API dokümantasyonu.

FAQ

Kısa özet

Çözücüyü worker'ların erişebileceği bir makineye koyun, host'u ortamda ayarlayın ve çözüm ile gönderimi bir havuzun arkasında birkaç yeniden denemeyle tek görevde tutun. Geri kalan her şey normal bir DAG'dir. Bunun tarama tarafı için şuraya bakın: web scraping için CAPTCHA çözücü sayfası, istemci yüzeyi için ise şuraya bakın: Python CAPTCHA çözücü sayfası. Fiyatlandırma modelinin kendini gösterdiği yer zamanlanmış işlerdir. Her çalıştırmada çözüm yapan saatlik bir DAG, sayaçla ücretlendiren bir sağlayıcıda gerçek bir faturadır; aradaki fark da burada: captcha çözücü zaten sahip olduğunuz donanımda çalıştığı için, DAG günde bir kez de dakikada bir kez de tetiklense maliyet aynıdır.