Tích hợp HTTPX + CaptchaAI

Cần giải CAPTCHA bằng Python mà không mở trình duyệt? HTTPX là lựa chọn đáng cân nhắc: một HTTP client hiện đại, chạy được cả chế độ đồng bộ lẫn async và hỗ trợ HTTP/2. Ghép với CaptchaAI, bạn có một tích hợp gọn nhẹ dùng lại được cho mọi loại CAPTCHA được hỗ trợ. Bài này đi thẳng vào code chạy được cho cả client đồng bộ và bất đồng bộ.

Bốn bước giải CAPTCHA với HTTPX

Dù dùng chế độ nào, luồng làm việc với CaptchaAI luôn theo cùng bốn bước:

1. Gửi CAPTCHA tới in.php kèm method, googlekey/sitekey và pageurl.
2. Nhận về ID task từ phản hồi dạng OK|<id>.
3. Polling res.php mỗi vài giây cho đến khi trạng thái khác CAPCHA_NOT_READY.
4. Lấy token trả về, đặt vào field tương ứng (ví dụ g-recaptcha-response) rồi gửi request tiếp theo.

Toàn bộ code bên dưới chỉ là các cách khác nhau để hiện thực hóa bốn bước này.

Yêu cầu môi trường

pip install httpx

Client đồng bộ với HTTPX

Với script tuần tự, một class bọc quanh httpx.Client là đủ: gửi task tới in.php, rồi polling res.php mỗi 5 giây cho đến khi có token hoặc chạm timeout.

import httpx
import time
import os

class CaptchaAISync:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"
        self.client = httpx.Client(timeout=30)

    def solve(self, params, timeout=300):
        params["key"] = self.api_key

        # Submit
        resp = self.client.get(f"{self.base_url}/in.php", params=params)
        text = resp.text

        if not text.startswith("OK|"):
            raise Exception(f"Submit failed: {text}")

        task_id = text.split("|")[1]

        # Poll
        deadline = time.time() + timeout
        poll_params = {"key": self.api_key, "action": "get", "id": task_id}

        while time.time() < deadline:
            time.sleep(5)
            result = self.client.get(
                f"{self.base_url}/res.php", params=poll_params
            )

            if result.text == "CAPCHA_NOT_READY":
                continue
            if result.text.startswith("OK|"):
                return result.text.split("|", 1)[1]
            raise Exception(f"Solve failed: {result.text}")

        raise TimeoutError(f"Task {task_id} timed out")

    def get_balance(self):
        resp = self.client.get(f"{self.base_url}/res.php", params={
            "key": self.api_key, "action": "getbalance"
        })
        return float(resp.text)

    def close(self):
        self.client.close()

# Usage
solver = CaptchaAISync(os.environ["CAPTCHAAI_API_KEY"])

token = solver.solve({
    "method": "userrecaptcha",
    "googlekey": "6Le-wvkS...",
    "pageurl": "https://example.com",
})
print(f"Token: {token[:50]}...")
solver.close()

Hàm get_balance gọi res.php với action=getbalance để kiểm tra số dư trước mỗi batch lớn — một thói quen nên có khi bạn tự động hóa số lượng nhiều.

Client bất đồng bộ: giải nhiều CAPTCHA song song

Điểm mạnh thật sự của HTTPX là async. Dùng httpx.AsyncClient, bạn gửi nhiều task cùng lúc và gom kết quả bằng asyncio.gather, thay vì chờ từng CAPTCHA giải xong mới gửi cái tiếp theo.

import httpx
import asyncio
import os

class CaptchaAIAsync:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"
        self.client = httpx.AsyncClient(timeout=30)

    async def solve(self, params, timeout=300):
        params["key"] = self.api_key

        # Submit
        resp = await self.client.get(
            f"{self.base_url}/in.php", params=params
        )
        text = resp.text

        if not text.startswith("OK|"):
            raise Exception(f"Submit failed: {text}")

        task_id = text.split("|")[1]

        # Poll
        deadline = asyncio.get_event_loop().time() + timeout
        poll_params = {"key": self.api_key, "action": "get", "id": task_id}

        while asyncio.get_event_loop().time() < deadline:
            await asyncio.sleep(5)
            result = await self.client.get(
                f"{self.base_url}/res.php", params=poll_params
            )

            if result.text == "CAPCHA_NOT_READY":
                continue
            if result.text.startswith("OK|"):
                return result.text.split("|", 1)[1]
            raise Exception(f"Solve failed: {result.text}")

        raise TimeoutError(f"Task {task_id} timed out")

    async def get_balance(self):
        resp = await self.client.get(f"{self.base_url}/res.php", params={
            "key": self.api_key, "action": "getbalance"
        })
        return float(resp.text)

    async def close(self):
        await self.client.aclose()

# Usage
async def main():
    solver = CaptchaAIAsync(os.environ["CAPTCHAAI_API_KEY"])

    # Solve multiple concurrently
    tasks = [
        solver.solve({
            "method": "userrecaptcha",
            "googlekey": "6Le-wvkS...",
            "pageurl": f"https://example.com/page{i}",
        })
        for i in range(5)
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)
    for i, r in enumerate(results):
        if isinstance(r, Exception):
            print(f"Page {i}: FAILED - {r}")
        else:
            print(f"Page {i}: solved ({len(r)} chars)")

    await solver.close()

asyncio.run(main())

Vài điểm cần nhớ khi chạy song song:

• Số CAPTCHA async giải cùng lúc bị chặn bởi số thread (luồng giải đồng thời) trong gói của bạn. range(5) gửi 5 task song song, khớp đúng gói BASIC ($15/tháng, 5 thread).
• CaptchaAI tính theo thread, số lần giải không giới hạn cho mỗi thread — muốn đồng thời cao hơn thì nâng thread, không trả thêm theo từng lần giải.
• return_exceptions=True giúp một task lỗi không kéo cả batch sập.

Bật HTTP/2 để giảm chi phí kết nối

HTTPX hỗ trợ HTTP/2 — cài thêm phần mở rộng là dùng được ngay:

pip install httpx[http2]

client = httpx.AsyncClient(http2=True, timeout=30)

HTTP/2 ghép nhiều request qua một kết nối duy nhất (multiplexing), giảm độ trễ khi bạn liên tục gửi và polling nhiều CAPTCHA.

Ví dụ thực tế: giải CAPTCHA khi scraping

Tình huống quen thuộc với các đội QA và thu thập dữ liệu tại Việt Nam — ví dụ team outsourcing ở TP.HCM kiểm thử luồng form, hay theo dõi giá công khai trên sàn TMĐT: tải trang, dò data-sitekey, nhờ CaptchaAI giải, rồi POST lại form kèm token. Chỉ dùng cho hệ thống bạn có quyền kiểm thử hoặc dữ liệu công khai.

import httpx
import re
import os

async def scrape_with_captcha(url, solver):
    async with httpx.AsyncClient() as client:
        # Fetch page
        resp = await client.get(url)
        html = resp.text

        # Check for reCAPTCHA
        match = re.search(
            r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', html
        )
        if not match:
            return html

        site_key = match.group(1)
        token = await solver.solve({
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": url,
        })

        # Submit form with token
        resp = await client.post(url, data={
            "g-recaptcha-response": token,
        })
        return resp.text

async def main():
    solver = CaptchaAIAsync(os.environ["CAPTCHAAI_API_KEY"])
    content = await scrape_with_captcha("https://example.com", solver)
    print(f"Got {len(content)} chars")
    await solver.close()

asyncio.run(main())

Token trả về đi vào trường g-recaptcha-response — đúng tên field mà backend reCAPTCHA v2 mong đợi khi bạn POST form.

httpx, requests hay aiohttp: chọn cái nào?

Cách chọn nhanh giữa ba thư viện:

• httpx đồng bộ — khi bạn muốn thay requests với ít thay đổi code nhất.
• httpx async — khi cần mức đồng thời cao và HTTP/2 cho việc giải hàng loạt.
• aiohttp — khi bạn đã quen nó cho workload async thuần và không cần HTTP/2.

Mẹo chạy ổn định trong production

Vài điểm giúp tích hợp bền hơn khi lên môi trường thật:

• Đặt timeout hợp lý cho cả request lẫn vòng polling để tránh treo tiến trình.
• Kiểm tra số dư bằng getbalance trước mỗi batch lớn để không giải nửa chừng thì hết credit.
• Luôn dùng return_exceptions=True với gather để một task lỗi không làm sập cả lô.
• Giãn nhịp gửi task thay vì dồn toàn bộ lên endpoint cùng một thời điểm.
• Khi xử lý dữ liệu thu thập, tuân thủ Nghị định 13/2023/NĐ-CP: chỉ giữ dữ liệu cần thiết và ghi log truy cập để phục vụ kiểm toán.

Câu hỏi thường gặp

Dùng HTTPX async có giúp giải nhiều CAPTCHA nhanh hơn không?

Có, khi bạn giải nhiều CAPTCHA cùng lúc. Async gửi và polling song song thay vì tuần tự, nên tổng thời gian cho một batch giảm rõ. Với một CAPTCHA đơn lẻ thì chênh lệch không đáng kể.

Chạy nhiều request async cùng lúc thì CaptchaAI tính tiền thế nào?

Theo thread (luồng giải đồng thời), không theo từng lần giải. Mỗi thread giải không giới hạn trong tháng; số task async song song tối đa bằng số thread trong gói, ví dụ BASIC ($15/tháng, 5 thread) cho 5 CAPTCHA cùng lúc.

Nên dùng polling hay webhook với HTTPX?

Polling (gọi res.php định kỳ) như ví dụ trên là đủ cho hầu hết script. Chạy dịch vụ dài hạn thì callback/webhook tiết kiệm request hơn, nhưng cần một endpoint công khai để nhận kết quả.

Cùng class HTTPX này có giải được Cloudflare Turnstile không?

Có. Đổi method thành turnstile, truyền sitekey và pageurl; token trả về đặt vào field cf-turnstile-response. CaptchaAI hỗ trợ reCAPTCHA v2/v3, Turnstile, Cloudflare Challenge, GeeTest v3 và image/OCR nên cùng một client dùng lại được.

Hướng dẫn liên quan

• Tích hợp aiohttp với CaptchaAI (async)
• Axios + CaptchaAI: giải CAPTCHA không cần trình duyệt
• Hướng dẫn scraping có xử lý CAPTCHA bằng Python