Tutorials

使用 Node.js 和 CaptchaAI 构建 CAPTCHA 解决事件总线

Python Automation All CAPTCHA Types Node.js
作者 CaptchaAI 中文团队
发布时间 May 02, 2026
使用 Node.js 和 CaptchaAI 构建 CAPTCHA 解决事件总线
特色图片不可用

回调和轮询处理结果，但它们不会让您的应用程序了解完整的验证码生命周期。事件总线广播状态更改——已提交、待处理、已解决、失败、超时——因此应用程序的任何部分都可以在没有紧密耦合的情况下做出反应。

事件总线架构

[CaptchaBus]
   ├── emit("submitted", { taskId, type, pageurl })
   ├── emit("pending", { taskId, elapsed })
   ├── emit("solved", { taskId, solution, duration })
   ├── emit("failed", { taskId, error, duration })
   └── emit("timeout", { taskId, elapsed })
        ↓          ↓           ↓
   [Logger]    [Metrics]   [Retry Handler]

听众独立登记。添加新功能（例如，指标收集）需要对求解代码进行零更改。

CaptchaBus 类 – JavaScript

const EventEmitter = require("events");
const axios = require("axios");

class CaptchaBus extends EventEmitter {
  constructor(apiKey, options = {}) {
    super();
    this.apiKey = apiKey;
    this.pollInterval = options.pollInterval || 5000;
    this.maxWait = options.maxWait || 300000; // 5 minutes
    this.pending = new Map();
  }

  async submit(params) {
    const { method, sitekey, pageurl, ...extra } = params;
    const taskId = `task_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`;

    const submitParams = {
      key: this.apiKey,
      method: method || "userrecaptcha",
      googlekey: sitekey,
      pageurl: pageurl,
      json: 1,
      ...extra,
    };

    try {
      const resp = await axios.post(
        "https://ocr.captchaai.com/in.php",
        null,
        { params: submitParams }
      );

      if (resp.data.status !== 1) {
        this.emit("failed", {
          taskId,
          error: resp.data.request,
          duration: 0,
        });
        return null;
      }

      const captchaId = resp.data.request;
      const startTime = Date.now();

      this.emit("submitted", {
        taskId,
        captchaId,
        method: method || "userrecaptcha",
        pageurl,
      });

      // Start polling
      this._poll(taskId, captchaId, startTime);
      return taskId;
    } catch (err) {
      this.emit("failed", { taskId, error: err.message, duration: 0 });
      return null;
    }
  }

  async _poll(taskId, captchaId, startTime) {
    const check = async () => {
      const elapsed = Date.now() - startTime;

      if (elapsed > this.maxWait) {
        this.emit("timeout", { taskId, elapsed });
        return;
      }

      this.emit("pending", { taskId, elapsed });

      try {
        const resp = await axios.get("https://ocr.captchaai.com/res.php", {
          params: {
            key: this.apiKey,
            action: "get",
            id: captchaId,
            json: 1,
          },
        });

        if (resp.data.status === 1) {
          this.emit("solved", {
            taskId,
            captchaId,
            solution: resp.data.request,
            duration: Date.now() - startTime,
          });
        } else if (resp.data.request === "CAPCHA_NOT_READY") {
          setTimeout(check, this.pollInterval);
        } else {
          this.emit("failed", {
            taskId,
            error: resp.data.request,
            duration: Date.now() - startTime,
          });
        }
      } catch (err) {
        this.emit("failed", {
          taskId,
          error: err.message,
          duration: Date.now() - startTime,
        });
      }
    };

    setTimeout(check, this.pollInterval);
  }
}

module.exports = CaptchaBus;

注册事件监听器

const CaptchaBus = require("./captcha-bus");

const bus = new CaptchaBus(process.env.CAPTCHAAI_API_KEY, {
  pollInterval: 5000,
  maxWait: 120000,
});

// Logging listener
bus.on("submitted", (e) => {
  console.log(`[SUBMIT] ${e.taskId} → ${e.method} on ${e.pageurl}`);
});

bus.on("pending", (e) => {
  console.log(`[PENDING] ${e.taskId} — ${(e.elapsed / 1000).toFixed(1)}s`);
});

bus.on("solved", (e) => {
  console.log(
    `[SOLVED] ${e.taskId} in ${(e.duration / 1000).toFixed(1)}s — ${e.solution.substring(0, 30)}...`
  );
});

bus.on("failed", (e) => {
  console.error(`[FAILED] ${e.taskId} — ${e.error}`);
});

bus.on("timeout", (e) => {
  console.error(
    `[TIMEOUT] ${e.taskId} after ${(e.elapsed / 1000).toFixed(1)}s`
  );
});

// Metrics listener
const metrics = { submitted: 0, solved: 0, failed: 0, totalDuration: 0 };

bus.on("submitted", () => metrics.submitted++);
bus.on("solved", (e) => {
  metrics.solved++;
  metrics.totalDuration += e.duration;
});
bus.on("failed", () => metrics.failed++);

// Submit a CAPTCHA
bus.submit({
  sitekey: "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
  pageurl: "https://example.com",
});

Python 等效项

import os
import time
import threading
from collections import defaultdict
import requests


class CaptchaBus:
    def __init__(self, api_key, poll_interval=5, max_wait=300):
        self.api_key = api_key
        self.poll_interval = poll_interval
        self.max_wait = max_wait
        self._listeners = defaultdict(list)

    def on(self, event, callback):
        """Register a listener for an event."""
        self._listeners[event].append(callback)
        return self

    def emit(self, event, data):
        """Emit an event to all registered listeners."""
        for callback in self._listeners.get(event, []):
            try:
                callback(data)
            except Exception as e:
                print(f"Listener error on {event}: {e}")

    def submit(self, sitekey, pageurl, method="userrecaptcha", **extra):
        """Submit a CAPTCHA and begin tracking."""
        task_id = f"task_{int(time.time())}_{id(sitekey) % 10000}"

        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key,
            "method": method,
            "googlekey": sitekey,
            "pageurl": pageurl,
            "json": 1,
            **extra
        })
        data = resp.json()

        if data.get("status") != 1:
            self.emit("failed", {
                "task_id": task_id,
                "error": data.get("request"),
                "duration": 0
            })
            return None

        captcha_id = data["request"]
        start_time = time.time()

        self.emit("submitted", {
            "task_id": task_id,
            "captcha_id": captcha_id,
            "method": method,
            "pageurl": pageurl
        })

        # Poll in a background thread
        thread = threading.Thread(
            target=self._poll,
            args=(task_id, captcha_id, start_time),
            daemon=True
        )
        thread.start()
        return task_id

    def _poll(self, task_id, captcha_id, start_time):
        while True:
            elapsed = time.time() - start_time

            if elapsed > self.max_wait:
                self.emit("timeout", {"task_id": task_id, "elapsed": elapsed})
                return

            time.sleep(self.poll_interval)
            self.emit("pending", {"task_id": task_id, "elapsed": elapsed})

            resp = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": captcha_id,
                "json": 1
            })
            data = resp.json()

            if data.get("status") == 1:
                self.emit("solved", {
                    "task_id": task_id,
                    "solution": data["request"],
                    "duration": time.time() - start_time
                })
                return
            elif data.get("request") != "CAPCHA_NOT_READY":
                self.emit("failed", {
                    "task_id": task_id,
                    "error": data.get("request"),
                    "duration": time.time() - start_time
                })
                return


# Usage
bus = CaptchaBus(os.environ["CAPTCHAAI_API_KEY"])

bus.on("submitted", lambda e: print(f"[SUBMIT] {e['task_id']}"))
bus.on("solved", lambda e: print(f"[SOLVED] {e['task_id']} in {e['duration']:.1f}s"))
bus.on("failed", lambda e: print(f"[FAILED] {e['task_id']} — {e['error']}"))
bus.on("timeout", lambda e: print(f"[TIMEOUT] {e['task_id']}"))

bus.submit("6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-", "https://example.com")

高级：作为侦听器重试处理程序

// Automatic retry on failure
bus.on("failed", async (e) => {
  if (e.retryCount >= 3) {
    console.error(`[GIVE UP] ${e.taskId} after 3 retries`);
    return;
  }

  console.log(`[RETRY] ${e.taskId} — attempt ${(e.retryCount || 0) + 1}`);
  await bus.submit({
    ...e.originalParams,
    _retryCount: (e.retryCount || 0) + 1,
  });
});

高级：Promise 包装器

在事件总线之上获取基于承诺的 API：

function solveCaptcha(bus, params) {
  return new Promise((resolve, reject) => {
    const taskId = bus.submit(params);

    function onSolved(e) {
      if (e.taskId === taskId) {
        cleanup();
        resolve(e.solution);
      }
    }

    function onFailed(e) {
      if (e.taskId === taskId) {
        cleanup();
        reject(new Error(e.error));
      }
    }

    function cleanup() {
      bus.removeListener("solved", onSolved);
      bus.removeListener("failed", onFailed);
      bus.removeListener("timeout", onFailed);
    }

    bus.on("solved", onSolved);
    bus.on("failed", onFailed);
    bus.on("timeout", onFailed);
  });
}

// Usage
const solution = await solveCaptcha(bus, {
  sitekey: "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
  pageurl: "https://example.com",
});

故障排除

问题 原因 处理方式
监听器不触发 事件名称不匹配（例如“解决”与“已解决”） 检查emit/on中使用的确切事件名称
内存泄漏警告 一项活动的听众过多 使用 setMaxListeners() 或使用后清理监听器
待处理事件淹没控制台 轮询间隔太短 将 pollInterval 增加到 5000+ 毫秒
重试中丢失的事件 重试时生成新任务 ID 传递原始参数以重新连接状态

常问问题

我应该使用外部消息代理吗？

对于单进程应用程序，进程内事件总线（EventEmitter）更简单、更快。当您有多个进程或服务需要对验证码事件做出反应时，请使用 Kafka、RabbitMQ 或 Redis。

我可以保留事件以进行调试吗？

是的。添加将事件写入 JSONL 文件或数据库的侦听器。这会在不修改求解逻辑的情况下创建审计跟踪。

如何在不调用 CaptchaAI 的情况下测试事件总线？

模拟 HTTP 调用。事件总线只是一个 EventEmitter - 您可以在测试中直接调用 bus.emit("solved", {...}) 来验证侦听器行为。

相关文章

下一步

构建事件驱动的验证码管道 -获取您的 CaptchaAI API 密钥并连接您的事件总线。

相关指南：

  • 回调 URL 和 Webhook 指南
  • SSE 实时通知
  • 回调错误处理模式
该文章已禁用评论。

相关文章

DevOps & Scaling 用于 CaptchaAI Worker 部署的 Ansible Playbook
使用 Captcha AI Worker 部署 Ansible Playbook 的 Dev Ops 指南，包括生产中 Captcha AI 工作流程的架构决策、操作注意事项和自动化模式。

用于 CaptchaAI Worker 部署的 Ansible Playbook

使用 Captcha AI Worker 部署 Ansible Playbook 的 Dev Ops 指南，包括生产中 Captcha AI 工作流程的架构决策、操作注...

Apr 19, 2026
DevOps & Scaling AWS Lambda + CaptchaAI：无服务器验证码解决
AWS Lambda + Captcha AI 的开发运营指南：无服务器验证码解决方案，包含生产中 Captcha AI 工作流程的架构决策、操作注意事项和自动化模式。

AWS Lambda + CaptchaAI：无服务器验证码解决

AWS Lambda + Captcha AI 的开发运营指南：无服务器验证码解决方案，包含生产中 Captcha AI 工作流程的架构决策、操作...

Apr 21, 2026
DevOps & Scaling 使用 AWS SNS 和 CaptchaAI 构建事件驱动的验证码解决方案
使用 AWS SNS 和 Captcha AI 构建事件驱动的验证码解决方案的开发运营指南，包括生产中 Captcha AI 工作流程的架构决策、操作注意事项和自动化模式。

使用 AWS SNS 和 CaptchaAI 构建事件驱动的验证码解决方案

使用 AWS SNS 和 Captcha AI 构建事件驱动的验证码解决方案的开发运营指南，包括生产中 Captcha AI 工作流程的架构决...

Apr 22, 2026