CaptchaAI Pingback 和任务通知模式

轮询会拖慢生产环境的验证码处理流程——脚本每 5 秒问一次“任务解决了没有”，并发一高，连接和延迟很快顶不住。CaptchaAI 的 pingback 参数把流程倒过来：任务一旦解决，CaptchaAI 主动把 token 推送到你指定的地址。

本文覆盖：

• Pingback 工作原理，以及何时该用它、何时轮询更合适
• 三种生产模式：结果存储、多任务扇出、按业务路由
• 端点加固与本地到线上的部署路径

Pingback 工作原理

1. Submit task with pingback=YOUR_CALLBACK_URL
2. CaptchaAI solves the CAPTCHA
3. CaptchaAI sends GET request to your callback:
   YOUR_CALLBACK_URL?id=TASK_ID&code=TOKEN

4. Your server processes the result

整个过程只有一次网络往返：

• 提交任务时带上 pingback 参数
• CaptchaAI 解决后主动 GET 请求，把 task_id 和 token 送过来

不需要轮询循环，也不需要反复调用 res.php。

Pingback 还是轮询：怎么选

两种方式各有适用场景。

怎么选，看两条：

• 本地脚本、临时任务、没有公网 IP：轮询更省事
• 并发上到两位数、追求低延迟：pingback 更合适

下面三种模式对应单机脚本、批量任务、多业务线路由，按架构挑一种落地即可。

模式 1：即发即忘，用字典存结果

提交任务，让回调把结果写进一个线程安全的字典：

import requests
import threading
import time
from flask import Flask, request


class PingbackStore:
    """Store for results received via pingback."""

    def __init__(self):
        self.results = {}
        self.events = {}
        self.lock = threading.Lock()

    def register(self, task_id):
        """Register a task ID we expect results for."""
        with self.lock:
            self.events[task_id] = threading.Event()

    def store(self, task_id, token):
        """Store result from pingback callback."""
        with self.lock:
            self.results[task_id] = token
            if task_id in self.events:
                self.events[task_id].set()

    def wait(self, task_id, timeout=120):
        """Wait for a specific result."""
        event = self.events.get(task_id)
        if not event:
            return None
        event.wait(timeout=timeout)
        return self.results.get(task_id)

    def get(self, task_id):
        """Get result without waiting (non-blocking)."""
        return self.results.get(task_id)


# Global store
store = PingbackStore()

# Flask app for receiving callbacks
app = Flask(__name__)


@app.route("/pingback")
def receive_pingback():
    """Handle CaptchaAI pingback callback."""
    task_id = request.args.get("id")
    code = request.args.get("code")

    if not task_id or not code:
        return "Bad request", 400

    store.store(task_id, code)
    return "OK", 200


def submit_with_pingback(api_key, method, callback_url, **params):
    """Submit a task with pingback enabled."""
    data = {
        "key": api_key,
        "method": method,
        "pingback": callback_url,
        "json": 1,
    }
    data.update(params)

    resp = requests.post(
        "https://ocr.captchaai.com/in.php",
        data=data,
        timeout=30,
    )
    result = resp.json()

    if result.get("status") != 1:
        raise RuntimeError(f"Submit error: {result.get('request')}")

    task_id = result["request"]
    store.register(task_id)
    return task_id


# Usage
# Start Flask server in background thread
server = threading.Thread(
    target=lambda: app.run(port=8080, debug=False),
    daemon=True,
)
server.start()

# Submit task
task_id = submit_with_pingback(
    "YOUR_API_KEY",
    "userrecaptcha",
    "https://yourserver.com/pingback",
    googlekey="SITE_KEY",
    pageurl="https://example.com",
)

# Wait for result via pingback
token = store.wait(task_id, timeout=120)
print(f"Token: {token[:50]}...")

模式 2：多任务扇出与结果收集

一次提交一批任务，结果到达一个记一个：

import requests
import threading
import time


class FanOutSolver:
    """Submit many tasks, collect results via pingback."""

    def __init__(self, api_key, callback_url):
        self.api_key = api_key
        self.callback_url = callback_url
        self.store = PingbackStore()
        self.pending = []

    def submit(self, method, **params):
        """Submit a task and track it."""
        data = {
            "key": self.api_key,
            "method": method,
            "pingback": self.callback_url,
            "json": 1,
        }
        data.update(params)

        resp = requests.post(
            "https://ocr.captchaai.com/in.php",
            data=data,
            timeout=30,
        )
        result = resp.json()

        if result.get("status") != 1:
            raise RuntimeError(f"Submit error: {result.get('request')}")

        task_id = result["request"]
        self.store.register(task_id)
        self.pending.append(task_id)
        return task_id

    def submit_batch(self, tasks):
        """Submit multiple tasks.

        tasks: list of dicts with 'method' and params
        """
        task_ids = []
        for task in tasks:
            method = task.pop("method")
            task_id = self.submit(method, **task)
            task_ids.append(task_id)
            time.sleep(0.1)  # Avoid rate limits
        return task_ids

    def collect_all(self, timeout=180):
        """Wait for all pending results."""
        results = {}
        deadline = time.time() + timeout

        for task_id in self.pending:
            remaining = max(1, deadline - time.time())
            token = self.store.wait(task_id, timeout=remaining)
            results[task_id] = token

        self.pending.clear()
        return results


# Usage
solver = FanOutSolver("YOUR_API_KEY", "https://yourserver.com/pingback")

# Submit 5 tasks
tasks = [
    {
        "method": "userrecaptcha",
        "googlekey": "SITE_KEY",
        "pageurl": f"https://example.com/page{i}",
    }
    for i in range(5)
]

task_ids = solver.submit_batch(tasks)
print(f"Submitted {len(task_ids)} tasks")

# Wait for all results
results = solver.collect_all(timeout=180)
for tid, token in results.items():
    status = "solved" if token else "failed"
    print(f"  {tid}: {status}")

批量场景下：

• submit_batch 把并发控制交给 CaptchaAI 服务端
• 脚本只管发任务、等结果

模式 3：按业务路由通知

同一个回调地址接住多条业务线时，用路由器把结果分发到对应处理函数：

import threading
from collections import defaultdict


class NotificationRouter:
    """Route pingback results to registered handlers."""

    def __init__(self):
        self.handlers = {}
        self.default_handler = None
        self.task_routes = {}
        self.lock = threading.Lock()

    def register_handler(self, name, handler_fn):
        """Register a named handler function."""
        self.handlers[name] = handler_fn

    def set_default(self, handler_fn):
        """Set a default handler for unrouted tasks."""
        self.default_handler = handler_fn

    def route(self, task_id, handler_name):
        """Route a task ID to a specific handler."""
        with self.lock:
            self.task_routes[task_id] = handler_name

    def dispatch(self, task_id, token):
        """Dispatch a result to the correct handler."""
        handler_name = self.task_routes.get(task_id)

        if handler_name and handler_name in self.handlers:
            self.handlers[handler_name](task_id, token)
        elif self.default_handler:
            self.default_handler(task_id, token)


# Usage
router = NotificationRouter()

# Register handlers
def login_handler(task_id, token):
    print(f"Login flow got token from {task_id}")
    # Submit token to login form

def scraping_handler(task_id, token):
    print(f"Scraping pipeline got token from {task_id}")
    # Continue scraping with token

router.register_handler("login", login_handler)
router.register_handler("scraping", scraping_handler)

# When submitting
task_id = submit_with_pingback(
    "YOUR_API_KEY", "userrecaptcha",
    "https://yourserver.com/pingback",
    googlekey="KEY", pageurl="https://example.com",
)
router.route(task_id, "login")

# In pingback handler
# router.dispatch(task_id, token)

加固你的 Pingback 接口

回调接口暴露在公网上，任何人都能对着这个地址发请求，所以校验参数、限制来源、做好幂等缺一不可：

import hmac
import hashlib
from flask import Flask, request, abort

app = Flask(__name__)
API_KEY = "YOUR_API_KEY"


@app.route("/pingback")
def secure_pingback():
    """Validate pingback requests."""
    task_id = request.args.get("id")
    code = request.args.get("code")
    ip = request.remote_addr

    # Validate required parameters
    if not task_id or not code:
        abort(400)

    # Validate IP (CaptchaAI server IPs)
    # Add actual CaptchaAI IPs to allowlist
    ALLOWED_IPS = {"0.0.0.0/0"}  # Replace with real IPs

    # Validate task ID format (numeric)
    if not task_id.isdigit():
        abort(400)

    # Store result
    store.store(task_id, code)
    return "OK", 200

生产环境里再做两件事：

• 把 ALLOWED_IPS 换成 CaptchaAI 真实的服务器 IP 段
• 叠加一层速率限制，挡掉无关请求

部署提示：从本地联调到线上

localhost 收不到回调，实际落地分三步：

1. 用 ngrok、frp 或国内的花生壳做内网穿透，把本地端口映射成临时公网地址
2. 把这个地址填进 pingback 参数联调
3. 上线后部署到有固定公网 IP 的服务器（阿里云、腾讯云 ECS 均可），安全组放行端口，换回正式域名

故障排查

常见问题

所有验证码类型都支持 pingback 吗？

支持。pingback 适用于 reCAPTCHA、Turnstile、GeeTest v3、图片验证码、BLS 等所有支持的方法，无需按类型单独配置。

本地开发环境怎么测试 pingback？

不能直接填 localhost。用 ngrok、frp 或花生壳做内网穿透，映射成临时公网地址，联调后换回正式域名。

pingback 和轮询可以一起用吗？

可以，而且推荐。pingback 处理即时通知，再加一层超时轮询兜底，避免网络抖动错过回调。

回调地址一定要用 HTTPS 吗？

官方没有强制要求，但生产环境建议启用：

• token 被截获等于泄露识别结果
• 用 Let's Encrypt 免费证书即可快速接入

相关指南

• 回调 URL 与 Webhook 配置指南
• 批量验证码识别指南

搭建事件驱动的验证码处理流程——立即获取你的 CaptchaAI API Key。