Python asyncio से कैप्चा समानांतर में कैसे हल करें

अगस्त 11, 2026 टिप्पणियाँ बंद

solve captchas in parallel - How to Solve CAPTCHAs in Parallel with Python asyncio

अगर आपको Python से कैप्चा समानांतर में हल करने हैं, तो उपयोग करें AsyncCapSkip के साथ asyncio.gather। Python अकेला ऐसा CapSkip SDK है जिसमें async क्लाइंट एक उपनाम नहीं बल्कि असली एसिंक्रोनस इम्प्लीमेंटेशन है, इसलिए दस reCAPTCHA का एक बैच लगभग उतने ही समय में पूरा होता है जितना उनमें से सबसे धीमा लेता है, न कि दसों का जोड़। यह गाइड चलने वाला कोड, सॉल्वर पर बोझ न पड़े इसके लिए समानांतरता को सीमित करने का तरीका, और एक विफलता से पूरा बैच न बिगड़े यह सब बताती है।

Python क्लाइंट ही दिलचस्प क्यों है

चारों SDK में एक चीज़ इस नाम से निर्यात होती है AsyncCapSkip। लेकिन इनमें से सिर्फ़ एक ही अलग इम्प्लीमेंटेशन है।

SDK	क्या है `AsyncCapSkip` है	आप काम समानांतर में कैसे चलाते हैं
Python	एक असली एसिंक्रोनस क्लाइंट	`await asyncio.gather(...)`
Node.js	इसका उपनाम `CapSkip`	`await Promise.all([...])`
.NET	इसका उपनाम `CapSkipClient`	`await Task.WhenAll(...)`
PHP	केवल उपनाम, सोर्स कोड की एकरूपता के लिए	सिंक्रोनस, कोई समानांतरता नहीं

Node और .NET पहले से ही नॉन-ब्लॉकिंग हैं, इसलिए वहाँ उपनाम से कोई फ़र्क नहीं पड़ता। PHP सिंक्रोनस है और वहाँ उपनाम से कुछ भी हासिल नहीं होता। Python में यह अंतर मायने रखता है: सामान्य CapSkip क्लाइंट पोलिंग के दौरान इवेंट लूप को ब्लॉक कर देता है, इसलिए उसे coroutine में डालने से कागज़ पर समानांतरता मिलती है और व्यवहार में क्रमिक समय।

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

Python 3.10 या उससे नया
pip install capskip
CapSkip अपने API सर्वर के साथ चल रहा हो और यहाँ सुन रहा हो 127.0.0.1:8080
काम करने के लिए sitekey और पेज URL की एक सूची

कुछ भी आपकी मशीन से बाहर नहीं जाता, इसलिए न कोई रेट लिमिट तय करनी है और न ही प्रयोग करते समय प्रति-हल कोई मीटर चलता है। हर प्रकार के लिए पूरे मेथड सिग्नेचर इस पेज पर हैं कैप्चा हल करने वाला SDK पेज पर हैं।

क्रमिक संस्करण, और उसकी कीमत

ज़्यादातर लोग इसी रूप से शुरुआत करते हैं। यह सही है, और यह धीमा है।

# pip install capskip
from capskip import CapSkip

solver = CapSkip(host="127.0.0.1", port=8080)

targets = [
    ("SITEKEY_A", "https://example.com/page-a"),
    ("SITEKEY_B", "https://example.com/page-b"),
    ("SITEKEY_C", "https://example.com/page-c"),
]

# Each call blocks until that one CAPTCHA comes back.
for sitekey, url in targets:
    result = solver.recaptcha(sitekey=sitekey, url=url)
    print(result["code"][:40])   # the token, truncated for the log

reCAPTCHA हल करने में ज़्यादातर समय इंतज़ार ही होता है। जब सॉल्वर काम करता है तब आपकी प्रक्रिया खाली बैठी रहती है, फिर अगले पर जाती है और फिर खाली बैठती है। तीन हल में तीन हल जितना ही वास्तविक समय लगता है। तीस में तीस जितना।

asyncio.gather से एक साथ कई कैप्चा हल करें

क्लाइंट बदलिए, कॉल को await कीजिए, और उन सबको सौंप दीजिए gather। प्रकार मिलाए जा सकते हैं: एक ही बैच में reCAPTCHA, Turnstile और GeeTest रखना बिल्कुल ठीक है।

# pip install capskip
import asyncio
from capskip import AsyncCapSkip

async def main():
    solver = AsyncCapSkip(host="127.0.0.1", port=8080)

    # gather starts all three now and waits for the slowest.
    results = await asyncio.gather(
        solver.recaptcha(sitekey="SITEKEY_A", url="https://example.com/page-a"),
        solver.turnstile(sitekey="SITEKEY_B", url="https://example.com/page-b"),
        solver.normal("captcha.png"),
    )

    for r in results:
        print(r["code"][:40])   # token for widgets, text for images

asyncio.run(main())

हर मेथड उन्हीं मुख्य फ़ील्ड वाला एक dict लौटाता है: captchaId और code. Turnstile अतिरिक्त रूप से लौटाता है userAgent, जिसे चैलेंज-पेज का हल भेजते समय token के साथ वापस भेजना ज़रूरी है। GeeTest इसमें जोड़ता है challenge, validate और seccode, और कच्चा JSON इसमें रखता है code.

सेमाफोर से समानांतरता सीमित करें

किसी लोकल डेमन पर दो सौ हल एक साथ फेंककर उम्मीद मत लगाइए। हल करना आपकी अपनी मशीन पर होने वाला CPU का काम है, इसलिए एक सीमा के बाद आप बस खुद से ही कतार लगा रहे होते हैं और हर अलग हल धीमा हो जाता है। सेमाफोर एक तय संख्या को ही चालू रखता है।

import asyncio
from capskip import AsyncCapSkip

# Start at 4 or 5, then measure. More is not automatically faster.
sem = asyncio.Semaphore(5)

async def solve_one(solver, sitekey, url):
    async with sem:
        return await solver.recaptcha(sitekey=sitekey, url=url)

async def run(targets):
    solver = AsyncCapSkip()
    tasks = [solve_one(solver, k, u) for k, u in targets]
    return await asyncio.gather(*tasks)

यह संख्या अंदाज़े से नहीं, समय मापकर चुनिए। वही 20 लक्ष्य 2, 5 और 10 पर चलाइए और जो आपके हार्डवेयर पर सबसे पहले पूरा हो उसे रखिए। सही उत्तर आपके CPU पर निर्भर करता है, SDK पर नहीं।

हर टास्क के लिए नहीं, एक ही क्लाइंट

एक ही AsyncCapSkip बनाइए और उसे coroutine के बीच साझा कीजिए, जैसा ऊपर दिखाया गया है। हर टास्क के लिए अलग बनाना फ़िज़ूलखर्ची है और इससे कुछ नहीं मिलता: क्लाइंट कॉन्फ़िगरेशन रखता है, प्रति-हल कोई स्थिति नहीं।

एक विफलता से पूरा बैच बिगड़ने से रोकें

डिफ़ॉल्ट रूप से gather पहला ही अपवाद ऊपर भेज देता है और उस समय चल रही बाकी सब चीज़ों के परिणाम आप खो देते हैं। यह पास कीजिए return_exceptions=True , तब अपवाद परिणाम सूची में सामान्य आइटम की तरह आते हैं और आप सफल तथा असफल को अलग कर सकते हैं।

from capskip import (
    AsyncCapSkip, ApiException, NetworkException,
    TimeoutException, ValidationException,
)

results = await asyncio.gather(*tasks, return_exceptions=True)

for target, r in zip(targets, results):
    if isinstance(r, TimeoutException):
        print("timed out, worth retrying:", target)
    elif isinstance(r, ApiException):
        print("api rejected this one:", target, r)
    elif isinstance(r, NetworkException):
        print("solver unreachable, stop the run:", target)
    elif isinstance(r, Exception):
        raise r
    else:
        print("ok:", r["code"][:40])

चारों अपवाद प्रकार हर CapSkip SDK में एक जैसे हैं, और सभी एक आधार CapSkipError से निकलते हैं, अगर आप एक ही प्रकार पकड़ना पसंद करें। ValidationException का मतलब है कि आपके पैरामीटर ग़लत हैं और दोबारा कोशिश भी ठीक वैसे ही विफल होगी। NetworkException का मतलब आमतौर पर यह है कि ऐप चल नहीं रहा, और यह पूरे रन की समस्या है, किसी एक लक्ष्य की नहीं।

टाइमआउट और पोलिंग, जो हर प्रकार के लिए अलग व्यवहार करते हैं

दो अलग-अलग टाइमआउट लागू होते हैं, और मिले-जुले प्रकारों वाला बैच दोनों से बंधा होता है।

विकल्प	डिफ़ॉल्ट	किस पर लागू
`defaultTimeout`	120 सेकंड	इमेज कैप्चा
`recaptchaTimeout`	300 सेकंड	reCAPTCHA, Turnstile, GeeTest
`pollingInterval`	5 सेकंड	The अधिकतम दो पोल के बीच का अंतराल

pollingInterval को बदलने से पहले समझ लेना ठीक रहेगा। SDK एक समान अंतराल पर पोल नहीं करता। यह 250ms से शुरू होकर आपके तय किए मान की ओर धीरे-धीरे बढ़ता है, यही वजह है कि SDK से किया गया हल आमतौर पर कच्चे API की “रुकिए, फिर हर पाँच सेकंड में पोल कीजिए” वाली सलाह से बने हाथ के लूप से पहले लौट आता है। इसे बढ़ाने पर तेज़ हल देर से आते हैं। घटाने पर बिना किसी लाभ के रिक्वेस्ट बढ़ जाती हैं।

GeeTest चैलेंज समाप्त हो जाते हैं, इसलिए बैच पहले से न बनाएँ

समानांतर चलने पर यह बात ख़ास तौर पर चुभती है। GeeTest का gt मान हर साइट के लिए स्थिर होता है, पर challenge एक ही बार इस्तेमाल होता है और लगभग एक मिनट में समाप्त हो जाता है। अगर आप पहले पचास चैलेंज इकट्ठा करके फिर हल करना शुरू करेंगे, तो कतार के आख़िर वाले भेजे जाने से पहले ही मर चुके होंगे। हर चैलेंज उसी हल से ठीक पहले लाइए जिसमें वह इस्तेमाल होना है।

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

# pip install capskip
import asyncio
from capskip import AsyncCapSkip, ApiException, TimeoutException

TARGETS = [
    ("SITEKEY_A", "https://example.com/page-a"),
    ("SITEKEY_B", "https://example.com/page-b"),
    ("SITEKEY_C", "https://example.com/page-c"),
]

async def solve_one(solver, sem, sitekey, url):
    async with sem:
        return await solver.recaptcha(sitekey=sitekey, url=url)

async def main():
    solver = AsyncCapSkip(host="127.0.0.1", port=8080)
    sem = asyncio.Semaphore(5)

    tasks = [solve_one(solver, sem, k, u) for k, u in TARGETS]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    tokens = {}
    for (sitekey, url), r in zip(TARGETS, results):
        if isinstance(r, (ApiException, TimeoutException)):
            print("failed:", url, r)
        else:
            tokens[url] = r["code"]

    print(len(tokens), "of", len(TARGETS), "solved")
    return tokens

asyncio.run(main())

पूरा तरीका बस यही है: एक साझा क्लाइंट, एक सेमाफोर, return_exceptions=True, और अंत में token का एक dict। इसे किसी स्क्रैपर में डाल दीजिए और कैप्चा वाला चरण अड़चन नहीं रहेगा। वेब स्क्रैपिंग के लिए कैप्चा सॉल्वर पेज बताता है कि यह एक बड़ी पाइपलाइन में कहाँ फ़िट होता है।

यही विचार दूसरे SDK में

अगर आप इसे पोर्ट कर रहे हैं, तो समानांतरता का साधन बदलेगा, ढाँचा नहीं। Node पहले से नॉन-ब्लॉकिंग है, इसलिए सामान्य क्लाइंट ही काफ़ी है।

// npm install capskip
const { CapSkip } = require('capskip');

const solver = new CapSkip({ host: '127.0.0.1', port: 8080 });

// AsyncCapSkip here is just an alias. Promise.all does the work.
const results = await Promise.all([
  solver.recaptcha('SITEKEY_A', 'https://example.com/page-a'),
  solver.turnstile('SITEKEY_B', 'https://example.com/page-b'),
]);

console.log(results.map(r => r.code));

.NET में भी यही कहानी है, इसके साथ Task.WhenAll, और PHP के पास समानांतरता है ही नहीं। अगर आपको समानांतर हल चाहिए और भाषा आप चुन सकते हैं, तो Python ही वह है जिसमें इसी काम के लिए बना क्लाइंट है। Python कैप्चा सॉल्वर पेज पर बाकी इंटरफ़ेस है।

आम ग़लतियाँ

ग़लती	क्या होता है	फिक्स
इसका उपयोग `CapSkip` coroutine के भीतर	इवेंट लूप ब्लॉक होता है, इसलिए समय क्रमिक ही रहता है	उपयोग करें `AsyncCapSkip`
सेमाफोर नहीं	कतार गहरी होते ही हर हल धीमा हो जाता है	चालू काम सीमित कीजिए, लगभग 5 से शुरू कीजिए
सादा `gather`	एक विफलता बाकी सभी परिणाम मिटा देती है	`return_exceptions=True`
GeeTest चैलेंज पहले से लाना	बाद वाले भेजे जाने से पहले समाप्त हो जाते हैं	हर एक को हल करने से ठीक पहले लाइए
इसे बढ़ाना `pollingInterval`	तेज़ हल जल्दी नहीं, देर से लौटते हैं	इसे डिफ़ॉल्ट पर ही रहने दीजिए
इमेज हल पर प्रॉक्सी लगाना	इमेज कैप्चा के लिए समर्थित नहीं	प्रॉक्सी केवल reCAPTCHA, Turnstile और GeeTest पर लागू होती हैं

अगर आप देखना चाहते हैं कि SDK असल में क्या भेज रहा है, तो हर प्रकार के लिए कच्ची रिक्वेस्ट और रिस्पॉन्स यहाँ है API डॉक्युमेंटेशन. Python का अपना asyncio टास्क संदर्भ कवर करता है gather इसके अर्थ विस्तार से बताता है।

अक्सर पूछे जाने वाले सवाल

मैं एक साथ कितने कैप्चा हल कर सकता हूँ?

टकराने के लिए कोई कोटा है ही नहीं, इसलिए सीमा आपका अपना हार्डवेयर है। हल लोकल रूप से होता है, इसलिए समानांतरता खाते के स्तर से नहीं बल्कि CPU से बंधी है। एक साथ पाँच से शुरू कीजिए, एक तय बैच का समय मापिए, और वहीं से समायोजित कीजिए।

क्या मैं एक ही gather कॉल में कैप्चा के प्रकार मिला सकता हूँ?

हाँ। recaptcha, turnstile, geetest और normal सभी एक ही क्लाइंट पर coroutine हैं और इन्हें साथ में await किया जा सकता है। याद रखिए कि इमेज हल 120 सेकंड का टाइमआउट लेते हैं और बाकी 300 का।

क्या मुझे इसकी जगह थ्रेड इस्तेमाल करने चाहिए?

सिर्फ़ तब, जब आपका आसपास का कोड पहले से थ्रेडेड हो। यहाँ का काम I/O पर आधारित इंतज़ार है, और asyncio इसी के लिए बना है; साथ ही एक इवेंट लूप थ्रेड पूल से सस्ता पड़ता है। अगर आप किसी सिंक्रोनस कोडबेस में फँसे हैं, तो सामान्य CapSkip क्लाइंट भी काम करता है।

क्या AsyncCapSkip को बंद करना ज़रूरी है?

SDK में न कोई close मेथड दर्ज है और न ही async context manager, इसलिए एक क्लाइंट बनाइए, पूरे रन में उसी का उपयोग कीजिए, और प्रक्रिया ख़त्म होने पर उसे स्वाभाविक रूप से समाप्त होने दीजिए।

सारांश

उपयोग करें AsyncCapSkip, एक ही क्लाइंट साझा कीजिए, चालू हलों को सेमाफोर से सीमित कीजिए, और यह पास कीजिए return_exceptions=True , ताकि एक ख़राब लक्ष्य पूरा बैच न मिटा दे। इससे कैप्चा की एक कतार क्रमिक अड़चन से बदलकर एक ही इंतज़ार बन जाती है।

आप समानांतरता खुलकर बढ़ा सकते हैं इसकी वजह यह है कि सॉल्वर आपकी अपनी मशीन पर चलता है। न प्रति-हल कोई बिल है और न अजनबियों के साथ साझा कोई कतार, इसलिए स्केल करना आपके CPU का सवाल है, किसी और की रेट लिमिट का नहीं। जब आपके बैच छोटे नहीं रह जाते, तब एक लोकल कैप्चा बायपास टूल यही व्यावहारिक फ़र्क लाता है।

कैप्चा टाइप्स

डेवलपर्स

AI इंटीग्रेशन

विकल्प

शुरुआत करना

डेवलपर संसाधन