Streaming kết quả batch CAPTCHA: xử lý từng token ngay khi giải xong

Nếu batch CAPTCHA của bạn chờ task cuối cùng xong mới bắt đầu xử lý, mỗi lần chạy tự cộng thêm vài phút chờ. Cách sửa gọn nhất là đọc kết quả theo luồng: token nào giải xong trước thì đẩy ngay xuống bước kế tiếp.

Dưới đây là hai mẫu code chạy được: async generator bằng Python (asyncio + aiohttp) và EventEmitter bằng Node.js. Số lần gọi in.php và res.php giữ nguyên; chỉ thời điểm bạn nhận kết quả đầu tiên là đổi.

Khi nào nên stream kết quả CAPTCHA, khi nào gom hết rồi xử lý

Tiêu chí: bước xuôi dòng xử lý được từng kết quả độc lập thì stream, cần cả batch mới làm được việc thì gom hết.

Ví dụ quen thuộc với team dữ liệu ở TP.HCM và Hà Nội: pipeline theo dõi giá chạy đêm trên danh mục công khai của Shopee, Lazada hay Tiki, khoảng 500 trang mỗi đêm. Gom hết thì báo cáo sáng hôm sau trễ đúng bằng task chậm nhất; stream thì bản ghi đầu tiên đã vào database.

Ba kiểu nhận kết quả batch và độ trễ tương ứng

Trong một batch 500 task, có task xong sau 8 giây, có task mất 45 giây. Khoảng chênh đó là thời gian chết nếu bạn chờ task chậm nhất.

Python: async generator trả kết quả CAPTCHA ngay khi giải xong

stream_results gom mọi task vào một set, dùng asyncio.wait với FIRST_COMPLETED và yield từng kết quả ngay khi nó xong.

import asyncio
import aiohttp
import time

API_KEY = "YOUR_API_KEY"
SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"

async def submit_task(session, task_data):
    """Submit a single CAPTCHA task."""
    params = {
        "key": API_KEY,
        "method": task_data.get("method", "userrecaptcha"),
        "json": 1,
    }
    if params["method"] == "userrecaptcha":
        params["googlekey"] = task_data["sitekey"]
        params["pageurl"] = task_data["pageurl"]
    elif params["method"] == "turnstile":
        params["sitekey"] = task_data["sitekey"]
        params["pageurl"] = task_data["pageurl"]

    async with session.post(SUBMIT_URL, data=params) as resp:
        result = await resp.json(content_type=None)
        if result.get("status") != 1:
            return None, result.get("request", "unknown")
        return result["request"], None

async def poll_task(session, task_id, timeout=300):
    """Poll until solved or timeout."""
    start = time.monotonic()
    while time.monotonic() - start < timeout:
        await asyncio.sleep(5)
        params = {"key": API_KEY, "action": "get", "id": task_id, "json": 1}
        async with session.get(RESULT_URL, params=params) as resp:
            result = await resp.json(content_type=None)

        if result.get("request") == "CAPCHA_NOT_READY":
            continue
        if result.get("status") == 1:
            return result["request"], None
        return None, result.get("request", "unknown")

    return None, "TIMEOUT"

async def solve_one(session, index, task_data, semaphore):
    """Solve a single task within concurrency limits."""
    async with semaphore:
        start = time.monotonic()
        task_id, error = await submit_task(session, task_data)
        if error:
            return {"index": index, "status": "failed", "error": error, "time": 0}

        token, error = await poll_task(session, task_id)
        elapsed = time.monotonic() - start

        if token:
            return {"index": index, "status": "solved", "token": token, "time": round(elapsed, 1)}
        return {"index": index, "status": "failed", "error": error, "time": round(elapsed, 1)}

async def stream_results(tasks, max_concurrent=20):
    """
    Async generator that yields each result as it completes.
    Results arrive in completion order, not submission order.
    """
    semaphore = asyncio.Semaphore(max_concurrent)

    async with aiohttp.ClientSession() as session:
        pending = set()
        for i, task in enumerate(tasks):
            coro = solve_one(session, i, task, semaphore)
            pending.add(asyncio.ensure_future(coro))

        while pending:
            done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
            for future in done:
                yield future.result()

async def main():
    tasks = [
        {"sitekey": "SITE_KEY", "pageurl": f"https://example.com/page{i}"}
        for i in range(50)
    ]

    solved = 0
    failed = 0

    async for result in stream_results(tasks, max_concurrent=15):
        # Process each result immediately
        if result["status"] == "solved":
            solved += 1
            print(f"  [{solved + failed}/{len(tasks)}] Task {result['index']} SOLVED in {result['time']}s")

            # Use token immediately — don't wait for batch
            # await submit_form(result["token"])
            # await save_to_database(result)
        else:
            failed += 1
            print(f"  [{solved + failed}/{len(tasks)}] Task {result['index']} FAILED: {result['error']}")

    print(f"\nDone: {solved} solved, {failed} failed")

asyncio.run(main())

Vòng async for trong main() là chỗ cắm logic xuôi dòng: submit form, ghi database, đẩy vào hàng đợi. Kết quả về theo thứ tự giải xong, nên map ngược bằng result["index"].

Cài thư viện:

pip install aiohttp

Node.js: stream kết quả CAPTCHA bằng EventEmitter

Bên Node.js, mẫu tự nhiên hơn là hướng sự kiện: CaptchaStream giữ hàng đợi nội bộ, chạy tối đa maxConcurrent task cùng lúc và phát sự kiện result khi có task xong.

const { EventEmitter } = require("events");

const API_KEY = "YOUR_API_KEY";
const SUBMIT_URL = "https://ocr.captchaai.com/in.php";
const RESULT_URL = "https://ocr.captchaai.com/res.php";

class CaptchaStream extends EventEmitter {
  constructor(maxConcurrent = 15) {
    super();
    this.maxConcurrent = maxConcurrent;
    this.active = 0;
    this.queue = [];
    this.total = 0;
    this.completed = 0;
  }

  async submitAndPoll(index, taskData) {
    const params = new URLSearchParams({
      key: API_KEY,
      method: taskData.method || "userrecaptcha",
      googlekey: taskData.sitekey,
      pageurl: taskData.pageurl,
      json: "1",
    });

    const start = Date.now();
    const submitResp = await (await fetch(SUBMIT_URL, { method: "POST", body: params })).json();

    if (submitResp.status !== 1) {
      return { index, status: "failed", error: submitResp.request, time: 0 };
    }

    const taskId = submitResp.request;
    for (let i = 0; i < 60; i++) {
      await new Promise((r) => setTimeout(r, 5000));
      const url = `${RESULT_URL}?key=${API_KEY}&action=get&id=${taskId}&json=1`;
      const poll = await (await fetch(url)).json();

      if (poll.request === "CAPCHA_NOT_READY") continue;
      const elapsed = ((Date.now() - start) / 1000).toFixed(1);
      if (poll.status === 1) return { index, status: "solved", token: poll.request, time: elapsed };
      return { index, status: "failed", error: poll.request, time: elapsed };
    }
    return { index, status: "failed", error: "TIMEOUT", time: ((Date.now() - start) / 1000).toFixed(1) };
  }

  async processNext() {
    if (this.queue.length === 0 || this.active >= this.maxConcurrent) return;

    const { index, taskData } = this.queue.shift();
    this.active++;

    try {
      const result = await this.submitAndPoll(index, taskData);
      this.emit("result", result);
    } catch (err) {
      this.emit("result", { index, status: "failed", error: err.message });
    } finally {
      this.active--;
      this.completed++;

      if (this.completed === this.total) {
        this.emit("done");
      } else {
        this.processNext();
      }
    }
  }

  start(tasks) {
    this.total = tasks.length;
    this.queue = tasks.map((taskData, index) => ({ index, taskData }));

    // Launch initial batch
    const initial = Math.min(this.maxConcurrent, tasks.length);
    for (let i = 0; i < initial; i++) {
      this.processNext();
    }
    return this;
  }
}

// Usage
const tasks = Array.from({ length: 50 }, (_, i) => ({
  sitekey: "SITE_KEY",
  pageurl: `https://example.com/page${i}`,
}));

const stream = new CaptchaStream(15);
let solved = 0, failed = 0;

stream.on("result", (result) => {
  if (result.status === "solved") {
    solved++;
    console.log(`[${solved + failed}/${tasks.length}] Task ${result.index} SOLVED (${result.time}s)`);
    // Use token immediately
    // submitForm(result.token);
  } else {
    failed++;
    console.log(`[${solved + failed}/${tasks.length}] Task ${result.index} FAILED: ${result.error}`);
  }
});

stream.on("done", () => {
  console.log(`\nComplete: ${solved} solved, ${failed} failed`);
});

stream.start(tasks);

Listener result chạy ngay khi từng task xong; sự kiện done chỉ bắn một lần khi cả batch kết thúc — dùng để đóng file hoặc flush buffer.

Đặt mức đồng thời khớp với số thread trong gói

CaptchaAI tính giá theo thread (luồng giải đồng thời), không theo lượt giải: mỗi thread là một CAPTCHA đang chạy, giải xong là nhận task tiếp. Vì vậy max_concurrent và maxConcurrent nên bằng đúng số thread của gói bạn dùng.

• Gói BASIC ($15/tháng, 5 thread): đặt mức đồng thời là 5.
• Gói ADVANCE ($90/tháng, 50 thread): đủ cho ví dụ 50 task trong bài.

Đặt cao hơn số thread không rút ngắn tổng thời gian, phần dư chỉ nằm chờ. Nếu batch chạm vào dữ liệu cá nhân, ghi log ID task và thời điểm xử lý ngay trong handler stream — tiện truy vết và hợp với yêu cầu tối thiểu hóa dữ liệu theo Nghị định 13/2023/NĐ-CP.

Lỗi hay gặp khi chạy stream và cách xử lý

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

Đặt max_concurrent bao nhiêu là hợp lý?

• Bằng đúng số thread của gói bạn đang dùng, không hơn.
• BASIC ($15/tháng, 5 thread) → max_concurrent=5.
• ADVANCE ($90/tháng, 50 thread) → đủ cho ví dụ 50 task trong bài.

Stream có làm tăng số lần gọi API không?

Không. Số request tới in.php và res.php giữ nguyên dù gom hết hay stream. Thay đổi duy nhất là thời điểm ứng dụng của bạn cầm được token đầu tiên.

Một task timeout thì cả stream có dừng không?

Không, miễn là bạn bắt lỗi trong coroutine như ví dụ trên. solve_one luôn trả về dict với status là solved hoặc failed, nên vòng asyncio.wait chạy tiếp và batch vẫn kết thúc đủ số task.

Mẫu stream này dùng được cho những loại CAPTCHA nào?

Mọi loại CaptchaAI hỗ trợ:

• Đầy đủ: reCAPTCHA v2/v3 kể cả Enterprise, Cloudflare Turnstile, Cloudflare Challenge, GeeTest v3, CAPTCHA ảnh/OCR, grid-image, BLS.
• Beta: CaptchaFox, Friendly Captcha, Lemin.
• Chưa hỗ trợ: hCaptcha, FunCaptcha, GeeTest v4.

Bài viết liên quan

• Xử lý CAPTCHA theo luồng với Kafka
• Xếp hàng đợi và ưu tiên cho batch CAPTCHA
• Giải CAPTCHA hàng loạt từ file CSV
• Giải CAPTCHA song song nhiều luồng

Bước tiếp theo

Đưa từng token vào pipeline ngay khi giải xong — lấy API key CaptchaAI và dựng luồng xử lý của bạn.