Khi bạn gửi 50 request reCAPTCHA v2 tới CaptchaAI bằng vòng lặp requests đồng bộ, chương trình phải đợi lần lượt từng response — trong khi event loop của Python thừa sức xử lý cả 50 request đó song song. Đó là lý do aiohttp là lựa chọn hợp lý khi bạn cần giải nhiều CAPTCHA cùng lúc mà không muốn tự dựng một pool thread. Bài này hướng dẫn viết một client aiohttp không đồng bộ gọi API CaptchaAI: gửi task tới in.php, polling res.php, và ghép asyncio.gather để giải hàng loạt trong một session HTTP duy nhất.
aiohttp hay requests đồng bộ?
- Với một CAPTCHA đơn lẻ: khác biệt gần như không đáng kể, dùng cách nào cũng được.
- Với 20, 50 hay 200 CAPTCHA cùng lúc — ví dụ quét giá trên hàng loạt trang danh mục Shopee hoặc Lazada —
requestsđồng bộ buộc bạn tự quản lýThreadPoolExecutor. - aiohttp xử lý toàn bộ trong một event loop, ít overhead hơn và dễ khống chế qua
asyncio.Semaphore(xem phần bên dưới). - httpx cũng hỗ trợ async theo cú pháp gần giống
requestsnếu bạn đã quen; xem hướng dẫn tích hợp httpx nếu muốn so sánh.
Yêu cầu trước khi bắt đầu
- Python 3.8+
- aiohttp 3.8+
- API key CaptchaAI — lấy tại đây
pip install aiohttp
Client CaptchaAI không đồng bộ
Class dưới đây bọc bốn lệnh gọi cốt lõi:
submitgửi task tớiin.phpvà trả về task ID.pollgọires.phpmỗi 5 giây cho tới khi có kết quả hoặc hếttimeout.solvegộpsubmitvàpollthành một lệnh gọi duy nhất, dùng cho hầu hết các ví dụ trong bài.get_balancekiểm tra số dư tài khoản mà không cần rời khỏi vòng lặp async.
import aiohttp
import asyncio
class AsyncCaptchaAI:
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://ocr.captchaai.com"
async def submit(self, session, params):
"""Submit a CAPTCHA task and return the task ID."""
params["key"] = self.api_key
async with session.get(
f"{self.base_url}/in.php", params=params
) as resp:
text = await resp.text()
if not text.startswith("OK|"):
raise Exception(f"Submit failed: {text}")
return text.split("|")[1]
async def poll(self, session, task_id, timeout=300):
"""Poll for the result with a timeout."""
params = {
"key": self.api_key,
"action": "get",
"id": task_id,
}
deadline = asyncio.get_event_loop().time() + timeout
while asyncio.get_event_loop().time() < deadline:
await asyncio.sleep(5)
async with session.get(
f"{self.base_url}/res.php", params=params
) as resp:
text = await resp.text()
if text == "CAPCHA_NOT_READY":
continue
if text.startswith("OK|"):
return text.split("|", 1)[1]
raise Exception(f"Solve failed: {text}")
raise TimeoutError(f"Task {task_id} timed out after {timeout}s")
async def solve(self, session, params, timeout=300):
"""Submit and poll in one call."""
task_id = await self.submit(session, params)
return await self.poll(session, task_id, timeout)
async def get_balance(self, session):
"""Check account balance."""
params = {"key": self.api_key, "action": "getbalance"}
async with session.get(
f"{self.base_url}/res.php", params=params
) as resp:
return float(await resp.text())
Mọi phương thức đều nhận session làm tham số thay vì tự tạo — nhờ vậy bạn dùng chung một aiohttp.ClientSession cho mọi lần giải, tận dụng connection pool có sẵn thay vì mở kết nối mới mỗi lần gọi.
Giải một CAPTCHA duy nhất
Ví dụ tối giản dưới đây kiểm tra số dư trước, sau đó giải một reCAPTCHA v2 — loại có SLA dưới 60 giây trên CaptchaAI với tỷ lệ giải thành công cao. token trả về là chuỗi bạn gắn vào field g-recaptcha-response khi submit lại form gốc.
import asyncio
import os
async def main():
solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])
async with aiohttp.ClientSession() as session:
# Check balance
balance = await solver.get_balance(session)
print(f"Balance: ${balance:.2f}")
# Solve reCAPTCHA v2
token = await solver.solve(session, {
"method": "userrecaptcha",
"googlekey": "6Le-wvkS...",
"pageurl": "https://example.com",
})
print(f"Token: {token[:50]}...")
asyncio.run(main())
Giải nhiều CAPTCHA cùng lúc bằng asyncio.gather
Khi cần giải hàng loạt — ví dụ quét CAPTCHA trên nhiều trang sản phẩm để đối chiếu giá trên Tiki hay Sendo — gom các coroutine solve() vào một list rồi chạy asyncio.gather với return_exceptions=True, để một task lỗi không làm sập toàn bộ batch.
async def solve_batch(urls, site_key):
solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])
async with aiohttp.ClientSession() as session:
tasks = [
solver.solve(session, {
"method": "userrecaptcha",
"googlekey": site_key,
"pageurl": url,
})
for url in urls
]
results = await asyncio.gather(*tasks, return_exceptions=True)
for url, result in zip(urls, results):
if isinstance(result, Exception):
print(f"FAILED {url}: {result}")
else:
print(f"SOLVED {url}: {len(result)} chars")
return results
urls = [
"https://example.com/page1",
"https://example.com/page2",
"https://example.com/page3",
"https://example.com/page4",
"https://example.com/page5",
]
asyncio.run(solve_batch(urls, "6Le-wvkS..."))
Log FAILED/SOLVED theo từng URL giúp bạn biết ngay task nào cần retry mà không phải dò lại toàn bộ danh sách.
Thu thập dữ liệu kèm xử lý CAPTCHA tự động
Hàm dưới đây tải trang, kiểm tra xem HTML có chứa g-recaptcha không, và chỉ gọi CaptchaAI khi thật sự cần — tránh tốn thread một cách vô ích trên các trang không có CAPTCHA.
async def scrape_with_captcha(url, site_key):
solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])
async with aiohttp.ClientSession() as session:
# Fetch the page
async with session.get(url) as resp:
html = await resp.text()
# Check if page has a CAPTCHA
if "g-recaptcha" not in html:
return html # No CAPTCHA, return content
# Solve the CAPTCHA
token = await solver.solve(session, {
"method": "userrecaptcha",
"googlekey": site_key,
"pageurl": url,
})
# Submit with solved token
async with session.post(url, data={
"g-recaptcha-response": token,
}) as resp:
return await resp.text()
Giới hạn số lượng giải đồng thời bằng Semaphore
Gửi quá nhiều task cùng lúc không giúp bạn giải nhanh hơn nếu vượt quá số thread trong gói CaptchaAI đang dùng — request thừa chỉ xếp hàng chờ. asyncio.Semaphore giới hạn số coroutine chạy song song tại một thời điểm; đặt max_concurrent khớp với số thread bạn có. Ví dụ, gói STANDARD ($30/tháng, 15 thread) hợp với max_concurrent=15, còn ADVANCE ($90/tháng, 50 thread) cho phép đẩy lên 50.
async def solve_with_limit(urls, site_key, max_concurrent=10):
solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])
semaphore = asyncio.Semaphore(max_concurrent)
async def solve_one(session, url):
async with semaphore:
return await solver.solve(session, {
"method": "userrecaptcha",
"googlekey": site_key,
"pageurl": url,
})
async with aiohttp.ClientSession() as session:
tasks = [solve_one(session, url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
solved = sum(1 for r in results if not isinstance(r, Exception))
print(f"Solved {solved}/{len(urls)} CAPTCHAs")
return results
In tỷ lệ solved/total ngay sau khi gather hoàn tất giúp bạn theo dõi độ ổn định của batch qua từng lần chạy, thay vì chỉ biết kết quả cuối cùng.
Giải Cloudflare Turnstile không đồng bộ
Cùng một client, chỉ cần đổi method sang turnstile và truyền sitekey thay vì googlekey là giải được Cloudflare Turnstile — loại CAPTCHA có SLA dưới 10 giây, nhanh hơn đáng kể so với reCAPTCHA v2.
async def solve_turnstile(url, sitekey):
solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])
async with aiohttp.ClientSession() as session:
token = await solver.solve(session, {
"method": "turnstile",
"sitekey": sitekey,
"pageurl": url,
})
return token
Xử lý lỗi thường gặp
Bốn lỗi dưới đây chiếm phần lớn các case gặp phải khi chạy client này ở môi trường production:
ClientConnectorError— nguyên nhân: sự cố mạng; cách khắc phục: kiểm tra kết nối.Submit failed: ERROR_ZERO_BALANCE— nguyên nhân: hết số dư; cách khắc phục: nạp thêm vào tài khoản.TimeoutError— nguyên nhân: giải chậm hơn dự kiến; cách khắc phục: tăng tham sốtimeout.RuntimeError: Event loop is closed— nguyên nhân: dùngasyncio.runtrong Jupyter; cách khắc phục: dùngnest_asyncio.
Nếu bạn chạy ví dụ trong Jupyter/Colab thay vì script độc lập, lỗi RuntimeError: Event loop is closed gần như chắc chắn sẽ xuất hiện — cài nest_asyncio trước khi gọi asyncio.run, hoặc chuyển sang chạy bằng script .py thông thường.
Câu hỏi thường gặp
- Cần bao nhiêu thread để chạy hàng nghìn CAPTCHA mỗi ngày? Phụ thuộc vào loại CAPTCHA và gói bạn dùng. CaptchaAI tính phí theo thread chứ không theo mỗi lần giải, nên số thread quyết định mức độ song song, không phải giá mỗi CAPTCHA. Với reCAPTCHA v2 (SLA dưới 60 giây/lần), một gói STANDARD 15 thread chạy hết công suất đã xử lý được hàng nghìn lượt/ngày; cần nhiều hơn thì tăng lên ADVANCE (50 thread) hoặc PREMIUM (100 thread) thay vì mở nhiều API key song song.
- Có thể giải Cloudflare Turnstile bằng client này không? Có — chỉ cần đổi
methodthànhturnstilevà truyềnsitekeythay vìgooglekey, phần còn lại của client (submit/poll/solve) dùng chung logic, không cần viết class riêng. - Chạy được bao nhiêu tác vụ giải đồng thời cùng lúc? CaptchaAI xử lý được hơn 100 request đồng thời ở phía API; giới hạn thực tế nằm ở số thread trong gói bạn đang dùng. Dùng
asyncio.Semaphoređể khớpmax_concurrentvới số thread đó, tránh gửi thừa task rồi phải xếp hàng chờ. - Một session aiohttp có tái sử dụng được cho nhiều lần giải không? Nên tái sử dụng. Một
aiohttp.ClientSessionduy trì connection pool, nên các request tiếp theo tớiin.php/res.phpnhanh hơn so với việc mở session mới cho mỗi lần giải.
Hướng dẫn liên quan
- Tích hợp httpx với CaptchaAI cho async I/O
- Giải CAPTCHA song song ở quy mô lớn
- Tích hợp Scrapy để tự động giải CAPTCHA khi crawl
