Prefect Flow में Retries के साथ कैप्चा कैसे हल करें

Prefect में कैप्चा वाला step retries लगा हुआ एक task है, और पूरा flow क़रीब बीस लाइनों का है। दूसरे hosted platforms से आने वालों को जो बात चौंकाती है वह सुखद है: Prefect Cloud आपका code कभी चलाता ही नहीं। वह काम schedule करता है, और आपकी अपनी मशीन पर चलता एक worker उसे एक outbound connection के ज़रिए उठा लेता है। इसलिए सॉल्वर 127.0.0.1 पर बैठ सकता है और flow उस तक पहुँच जाता है, जो Zapier या Make.com पर सच नहीं है। असल में जो दो चीज़ें बिगड़ती हैं वे हैं retry का बर्ताव और caching, और दोनों के लिए एक-एक argument ही काफ़ी है।

आपको क्या चाहिए

• Prefect 3 और Python 3.10 या उससे नया, साथ में CapSkip Python SDK।
• एक Prefect Cloud workspace या एक self-hosted Prefect server। इस काम के लिए दोनों एक जैसे ही चलते हैं।
• सुरक्षित form का पेज URL, और उसकी sitekey।
• जब worker और सॉल्वर एक ही मशीन साझा करते हों तो CapSkip Local mode में चल रहा हो, और जब न करते हों तो Server mode में। दोनों का वर्णन यहाँ है: कनेक्शन सेटिंग्स.

# 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 असल में कहाँ चलता है

यही वह सवाल है जो आपका पूरा network setup तय करता है, इसलिए पहले इसी का जवाब दें। Prefect का डिफ़ॉल्ट एक hybrid मॉडल है: orchestration की परत hosted है, और execution की परत आपकी। Prefect Cloud metadata सहेजता है और runs में तालमेल बिठाता है, वह आपका code नहीं चलाता, और उसे आपके network तक कोई inbound पहुँच नहीं चाहिए। आपके अपने infrastructure के भीतर शुरू किया गया worker काम के लिए बाहर की ओर poll करता है और runs को लोकल रूप से चलाता है।

इसका व्यावहारिक नतीजा साफ़-साफ़ कह देना चाहिए। CapSkip वाली उसी Windows मशीन पर एक Process worker चलाइए और flow ठीक वैसे ही 127.0.0.1:8080 को कॉल करता है जैसे आपकी मेज़ पर रखी कोई script करती। न कोई tunnel, न कोई public address, न कोई certificate। यह सिर्फ़ cloud पर चलने वाले किसी ऑटोमेशन platform की हालत से ठीक उलटा है, और यही मुख्य वजह है कि हल करने का काम रखने के लिए कोई orchestrator आरामदेह जगह है।

एक अपवाद है और work pool चुनने से पहले आपको उसे जान लेना चाहिए। Prefect Managed work pools आपका flow आपके नहीं बल्कि Prefect के अपने infrastructure पर चलाते हैं, जो सुविधाजनक तो है पर loopback वाला विकल्प पूरी तरह ख़त्म कर देता है। उस configuration में सॉल्वर को Server mode और एक पहुँच योग्य address चाहिए। Prefect छह static outbound addresses प्रकाशित करता है जिन्हें Managed runs इस्तेमाल करते हैं, इसलिए आप firewall से होकर सॉल्वर के port तक ठीक उन्हीं को अनुमति दे सकते हैं और बाक़ी सब गिरा सकते हैं। Managed runs को एक आधिकारिक Prefect image भी इस्तेमाल करना पड़ता है और उनकी सीमा 24 घंटे है, और यह एक और वजह है कि यहाँ आम तौर पर कोई Process या Docker work pool ज़्यादा फ़िट बैठता है।

चरण 1: key को एक Secret block में रखें

API key को flow फ़ाइल में hardcode न करें। Prefect एक Secret block देता है, मान backend में at rest encrypted रहते हैं, और उसे लोड करना दो लाइनों का काम है। इसे एक बार किसी Python shell से सेव कर लें।

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

चरण 2: solve वाला task लिखें

एक task, एक solve। उसे retries दें, क्योंकि सॉल्वर की कॉल एक network call है और network calls नाकाम होती हैं। Prefect एक तय देरी, देरियों की एक सूची, या एक exponential backoff helper लेता है, और एक jitter factor retries को फैला देता है ताकि नाकामियों का एक जत्था एक साथ क़दम मिलाकर वापस न आए।

अहम argument दूसरा वाला है। कैप्चा token एक ही बार इस्तेमाल होता है और कुछ ही मिनटों में ख़त्म हो जाता है, इसलिए वह कभी किसी cache से नहीं आना चाहिए। Prefect 3 तब तक caching बंद रखता है जब तक result persistence चालू न हो, यानी ज़्यादातर लोग संयोग से ही सुरक्षित रहते हैं। अगर आपकी टीम ने persistence को globally चालू कर रखा है, और बहुतों ने किया है, तो दोबारा चलाया गया task वही token लौटा सकता है जो उसने पहली बार बनाया था और form उसे ठुकरा देगा। policy को साफ़ तौर पर सेट कर दीजिए और इसके बारे में सोचना बंद कर दीजिए।

# 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"]

एक ही method reCAPTCHA v2, Invisible, Enterprise और v3 को कवर करता है। variants अलग calls नहीं बल्कि options हैं: invisible=1, enterprise=1, या किसी action के साथ version="v3"। Turnstile और GeeTest के अपने methods हैं और ढाँचा भी वही है। पूरी parameter सूचियाँ यहाँ हैं: CapSkip API डॉक्यूमेंटेशन.

चरण 3: उन errors को दोबारा न आज़माएँ जो कभी पास नहीं होंगी

आँख मूँदकर की गई retries उन नाकामियों पर वक़्त बर्बाद करती हैं जो हर बार एक जैसी रहती हैं। कोई ग़लत बना argument हर प्रयास पर एक ही तरह fail होता है, और वही हाल उस sitekey का है जो उस पेज की है ही नहीं। कोई retry condition function state पाता है और तय करता है, और False लौटाने पर task तुरंत मूल exception के साथ ख़त्म हो जाता है।

# 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

उसे task पर retry_condition_fn के रूप में पास करें। NetworkException का मतलब है CapSkip चल नहीं रहा या host ग़लत है, और TimeoutException का मतलब है कि solve recaptchaTimeout से आगे खिंच गया, जो डिफ़ॉल्ट रूप से 300 seconds है। दोनों वाक़ई एक और प्रयास के लायक़ हैं। ये दोनों, और साथ में ValidationException तथा ApiException, सभी एक ही साझा base से निकलते हैं, इसलिए अगर आप नाकामियों को एक ही जगह संभालना चाहें तो CapSkipError पकड़ना काम कर जाता है।

पूरा चलने वाला उदाहरण

पूरा flow। पेज fetch करें, उसमें से sitekey निकालें, हल करें, फिर token को form के साथ वापस post कर दें। हर step एक task है, इसलिए हर एक को अपनी retries, अपने logs और run graph में अपनी जगह मिलती है।

# 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())

submit करने से ठीक पहले हल करें, कभी किसी पहले वाले schedule किए गए step में नहीं। जो token किसी upstream task के ख़त्म होने का इंतज़ार करते हुए दस मिनट तक result store में पड़ा रहता है, वह form तक पहुँचने तक मरा हुआ token होता है।

सॉल्वर पर बोझ लादे बिना पूरा batch हल करना

Prefect डिफ़ॉल्ट रूप से एक thread pool के ज़रिए tasks को concurrently चलाता है, इसलिए सौ solves का मतलब है submit पर एक list comprehension और task runner की कोई configuration बिल्कुल नहीं। यह उससे ज़्यादा parallelism है जितनी आप शायद एक ही मशीन की तरफ़ तानना चाहेंगे।

इसका नियंत्रण एक global concurrency limit है। limit को CLI से एक बार बना लें, फिर task के अंदर एक slot घेर लें, और सीमा से ऊपर का कोई भी run ढेर लगाने के बजाय इंतज़ार करता है।

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

यह limit workspace के हर flow run पर लागू होती है, और जब तीन schedules एक ही सॉल्वर की तरफ़ इशारा कर रहे हों तो आपको यही चाहिए। अगर आप fan out को एक ही process के अंदर करना चाहें, तो Python SDK का AsyncCapSkip एक असली asyncio client है और वह तरीक़ा यहाँ कवर किया गया है: समानांतर में कैप्चा हल करने की गाइड.

सॉल्वर को किसी दूसरी मशीन पर चलाना

Workers इधर-उधर जाते रहते हैं। आपकी मेज़ का Process worker किसी server पर Docker worker बन जाता है, फिर एक Kubernetes work pool, और किसी मोड़ पर flow उस मशीन पर रहता ही नहीं जिस पर सॉल्वर चलता है। code में host के अलावा कुछ नहीं बदलता।

CapSkip में दो connection modes हैं। Local, 127.0.0.1 से bind होता है और सिर्फ़ उसी डिवाइस को जवाब देता है। Server, आपके network या public IP से bind होता है, जिससे किसी VM पर चलता कोई worker, कोई container host या कोई Managed work pool उसी Windows मशीन को API के ज़रिए कॉल करता है। एक static public IP उस address को स्थिर रखता है। दोनों ही सूरतों में यह अब भी आपका अपना hardware है और अब भी unmetered है, इसलिए किसी व्यस्त दिन की लागत mode के साथ नहीं बदलती।

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

जैसे ही सॉल्वर किसी network address पर सुनने लगे, key validation चालू कर दें, और हर worker को उसकी अपनी key दें ताकि किसी एक को बाक़ियों को छुए बिना रद्द किया जा सके। दोनों modes को यहाँ विस्तार से समझाया गया है: CapSkip सेटअप गाइड.

आम errors और उनका मतलब

FAQ

संक्षेप में

key को एक Secret block में रखें, solve को retries और NO_CACHE वाले एक task में लपेटें, और उसे submit से ठीक पहले वाले step में कॉल करें। worker को उसी मशीन पर चलाएँ जिस पर CapSkip है और host 127.0.0.1 ही रहता है। किसी batch को फैलाने से पहले एक global concurrency limit जोड़ें। इस सबका Python वाला पहलू यहाँ कवर किया गया है: Python कैप्चा सॉल्वर पेज। Node.js, PHP और C# में भी वही तीन calls मौजूद हैं, और वे यहाँ सूचीबद्ध हैं: CAPTCHA solving SDK पेज.

इसे हर घंटे schedule करने से पहले एक बात जान लेना ज़रूरी है। CapSkip कैप्चा बायपास उसी hardware पर करता है जो पहले से आपका अपना है, इसलिए दिन में दस हज़ार हल करने वाले flow की लागत उतनी ही है जितनी दस हल करने वाले की।