您可以提供回调 URL,而不是重复轮询 /res.php。 CaptchaAI 准备好后将解决的令牌直接发送到您的服务器。这消除了轮询开销并减少了延迟。
回调如何工作
Standard polling approach:
Submit ──▶ Wait ──▶ Poll ──▶ Poll ──▶ Poll ──▶ Result
(many requests, wasted time between polls)
Callback approach:
Submit (with callback URL) ──▶ ... CaptchaAI solves ...
│
Your server receives POST ◀───────────────┘
(one request, instant delivery)
使用回调 URL 提交
将 pingback 参数添加到您的提交请求中:
import requests
API_KEY = "YOUR_API_KEY"
resp = requests.post("https://ocr.captchaai.com/in.php", data={
"key": API_KEY,
"method": "userrecaptcha",
"googlekey": "SITE_KEY",
"pageurl": "https://example.com",
"pingback": "https://your-server.com/captcha-callback",
"json": 1,
})
result = resp.json()
task_id = result["request"]
print(f"Task submitted: {task_id}")
# No polling needed — result comes via webhook
支持所有验证码类型
# reCAPTCHA v2
data = {
"key": API_KEY,
"method": "userrecaptcha",
"googlekey": "SITE_KEY",
"pageurl": "https://example.com",
"pingback": "https://your-server.com/callback",
"json": 1,
}
# Turnstile
data = {
"key": API_KEY,
"method": "turnstile",
"sitekey": "SITE_KEY",
"pageurl": "https://example.com",
"pingback": "https://your-server.com/callback",
"json": 1,
}
# Image CAPTCHA
data = {
"key": API_KEY,
"method": "base64",
"body": base64_image,
"pingback": "https://your-server.com/callback",
"json": 1,
}
构建回调接收器(Flask)
from flask import Flask, request
import threading
import logging
app = Flask(__name__)
logger = logging.getLogger(__name__)
# Store results by task ID
results = {}
events = {}
@app.route("/captcha-callback", methods=["POST", "GET"])
def captcha_callback():
"""Receive solved CAPTCHA tokens from CaptchaAI."""
# CaptchaAI sends parameters as query string or form data
task_id = request.args.get("id") or request.form.get("id")
code = request.args.get("code") or request.form.get("code")
if not task_id or not code:
logger.warning("Callback missing id or code")
return "ERROR", 400
logger.info("Received result for task %s", task_id)
results[task_id] = code
# Notify waiting threads
event = events.get(task_id)
if event:
event.set()
return "OK"
def wait_for_result(task_id, timeout=120):
"""Wait for a callback result."""
event = threading.Event()
events[task_id] = event
if task_id in results:
return results.pop(task_id)
event.wait(timeout=timeout)
if task_id in results:
return results.pop(task_id)
raise TimeoutError(f"No callback received for task {task_id}")
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000)
完整的工作流程
import requests
import threading
import time
API_KEY = "YOUR_API_KEY"
CALLBACK_URL = "https://your-server.com/captcha-callback"
def solve_with_callback(sitekey, pageurl):
"""Submit CAPTCHA and wait for callback."""
# Submit with callback URL
resp = requests.post("https://ocr.captchaai.com/in.php", data={
"key": API_KEY,
"method": "userrecaptcha",
"googlekey": sitekey,
"pageurl": pageurl,
"pingback": CALLBACK_URL,
"json": 1,
})
result = resp.json()
if result.get("status") != 1:
raise RuntimeError(f"Submit failed: {result.get('request')}")
task_id = result["request"]
print(f"Task {task_id} submitted, waiting for callback...")
# Wait for callback on the receiver
token = wait_for_result(task_id, timeout=120)
print(f"Token received via callback: {token[:50]}...")
return token
回调与轮询比较
| 因素 | 轮询 | 打回来 |
|---|---|---|
| API请求 | 很多(提交+N次投票) | 2(提交+回调) |
| 延迟 | 轮询间隔延迟 | 即时交付 |
| 服务器负载 | 客户端持续轮询 | CaptchaAI 推送结果 |
| 复杂 | 简单(仅限客户端) | 需要公共端点 |
| 防火墙要求 | 仅限出境 | 必须接受入站 POST |
| 最适合 | 简单的脚本 | 生产管道 |
何时使用回调
| 设想 | 推荐方法 |
|---|---|
| 快速的脚本,很少的解决方案 | 轮询 |
| 生产流水线,大批量 | 打回来 |
| 无服务器(Lambda、云函数) | 轮询(无持久服务器) |
| 在防火墙后面,没有公共IP | 轮询 |
| 微服务架构 | 打回来 |
保护您的回调端点
验证回调实际上来自 CaptchaAI:
from flask import Flask, request, abort
app = Flask(__name__)
# Store submitted task IDs to validate callbacks
pending_tasks = set()
def submit_captcha(sitekey, pageurl):
"""Submit and track task ID."""
resp = requests.post("https://ocr.captchaai.com/in.php", data={
"key": API_KEY,
"method": "userrecaptcha",
"googlekey": sitekey,
"pageurl": pageurl,
"pingback": CALLBACK_URL,
"json": 1,
})
task_id = resp.json()["request"]
pending_tasks.add(task_id)
return task_id
@app.route("/captcha-callback", methods=["POST", "GET"])
def secure_callback():
"""Validate callback before processing."""
task_id = request.args.get("id") or request.form.get("id")
# Reject unknown task IDs
if task_id not in pending_tasks:
abort(403)
code = request.args.get("code") or request.form.get("code")
pending_tasks.discard(task_id)
results[task_id] = code
return "OK"
故障排除
| 问题 | 原因 | 处理方式 |
|---|---|---|
| 没有收到回电 | 服务器不可公开访问 | 使用ngrok进行测试,或部署到云端 |
| 回调 URL 被拒绝 | 无法从 CaptchaAI 访问 URL | 验证 URL 可公开访问 |
| 部分结果 | 部分回调失败 | 记录所有回调,添加 retry/polling 后备 |
| 重复回调 | 网络重试 | 使用任务 ID 重复数据删除 |
常问问题
回调 URL 是否需要 HTTPS?
建议在生产环境中使用 HTTPS。 HTTP 适用于测试,但会在传输过程中暴露令牌。
如果我的回调服务器宕机了怎么办?
CaptchaAI 可能会重试回调,但您应该实现轮询回退以确保可靠性。如果在预期时间内没有收到回调,请通过 /res.php 检查结果。
我可以为每个请求使用不同的回调 URL 吗?
是的。每个 pingback 参数都是针对每个请求的。不同的验证码类型或项目可以使用不同的回调端点。
相关指南
- Pingback 和通知模式
- 余额检查和自动充值
消除轮询开销 -尝试 CaptchaAI 回调用于即时代币交付。