从国内网络测试 reCAPTCHA 或 Turnstile 时，一次跨境请求的往返延迟本来就不短——如果只是因为 sitekey 传空了才失败，那几秒钟纯属浪费。CaptchaAI 的 API 也是同样道理：空 sitekey 提交上去，要等接口返回 ERROR_WRONG_CAPTCHA_ID 才知道错了。用 Pydantic 在发请求前先校验参数，能把这类问题挡在本地：拿到的是清晰的字段级报错，而不是要查文档才看懂的错误码。
下面这套 Pydantic v2 模型和客户端封装可以直接照抄，覆盖 reCAPTCHA v2/v3、Turnstile 和图片验证码四种最常用类型。
为什么要给验证码 API 参数加 Pydantic 校验
不加校验的时候，问题都要等请求真正发出去才会暴露：
- sitekey 传空 → 得等 5 秒才收到 API 报错；加了 Pydantic 校验，提交前立刻抛出
ValidationError。 - 靠
dict["key"]手动解析响应 → 一不小心就 KeyError；换成带默认值、带校验的类型化模型，字段缺失在赋值阶段就报错。 - IDE 没有字段自动补全；模型定义好之后，所有字段都带完整类型提示。
请求与响应模型定义
下面这组模型覆盖四种请求参数（reCAPTCHA v2、v3、Turnstile、图片验证码），以及提交、轮询两类响应的解析结构。字段名对齐 CaptchaAI 官方参数（sitekey、pageurl、googlekey 等），校验通过后用 to_params() 直接拼出提交参数字典。
# models.py
from pydantic import BaseModel, Field, field_validator, HttpUrl
from enum import Enum
from typing import Optional
class CaptchaMethod(str, Enum):
RECAPTCHA_V2 = "userrecaptcha"
RECAPTCHA_V3 = "userrecaptcha" # Differentiated by version field
TURNSTILE = "turnstile"
HCAPTCHA = "hcaptcha"
IMAGE = "base64"
GEETEST = "geetest"
class RecaptchaV2Request(BaseModel):
"""Parameters for solving reCAPTCHA v2."""
sitekey: str = Field(min_length=20, max_length=100, description="Site's reCAPTCHA sitekey")
pageurl: HttpUrl = Field(description="URL where CAPTCHA appears")
invisible: bool = False
cookies: Optional[str] = None
@field_validator("sitekey")
@classmethod
def validate_sitekey(cls, v: str) -> str:
if v.strip() != v:
raise ValueError("Sitekey must not have leading/trailing whitespace")
return v
def to_params(self) -> dict:
params = {
"method": "userrecaptcha",
"googlekey": self.sitekey,
"pageurl": str(self.pageurl),
}
if self.invisible:
params["invisible"] = "1"
if self.cookies:
params["cookies"] = self.cookies
return params
class RecaptchaV3Request(BaseModel):
"""Parameters for solving reCAPTCHA v3."""
sitekey: str = Field(min_length=20, max_length=100)
pageurl: HttpUrl
action: str = Field(default="verify", min_length=1, max_length=100)
def to_params(self) -> dict:
return {
"method": "userrecaptcha",
"version": "v3",
"googlekey": self.sitekey,
"pageurl": str(self.pageurl),
"action": self.action,
}
class TurnstileRequest(BaseModel):
"""Parameters for solving Cloudflare Turnstile."""
sitekey: str = Field(min_length=10, max_length=100)
pageurl: HttpUrl
action: Optional[str] = None
cdata: Optional[str] = None
def to_params(self) -> dict:
params = {
"method": "turnstile",
"sitekey": self.sitekey,
"pageurl": str(self.pageurl),
}
if self.action:
params["action"] = self.action
if self.cdata:
params["data"] = self.cdata
return params
class ImageRequest(BaseModel):
"""Parameters for solving image/text CAPTCHA."""
base64_image: str = Field(min_length=100, description="Base64-encoded image")
case_sensitive: bool = False
min_length: Optional[int] = Field(default=None, ge=1, le=50)
max_length: Optional[int] = Field(default=None, ge=1, le=50)
@field_validator("base64_image")
@classmethod
def validate_base64(cls, v: str) -> str:
# Strip data URI prefix if present
if v.startswith("data:"):
parts = v.split(",", 1)
if len(parts) == 2:
return parts[1]
return v
def to_params(self) -> dict:
params = {
"method": "base64",
"body": self.base64_image,
}
if self.case_sensitive:
params["regsense"] = "1"
if self.min_length is not None:
params["min_len"] = str(self.min_length)
if self.max_length is not None:
params["max_len"] = str(self.max_length)
return params
class SubmitResponse(BaseModel):
"""Parsed API submit response."""
status: int
request: str
@property
def success(self) -> bool:
return self.status == 1
@property
def task_id(self) -> str:
if not self.success:
raise ValueError(f"No task ID — submission failed: {self.request}")
return self.request
class PollResponse(BaseModel):
"""Parsed API poll response."""
status: int
request: str
@property
def ready(self) -> bool:
return self.request != "CAPCHA_NOT_READY"
@property
def success(self) -> bool:
return self.status == 1
@property
def token(self) -> str:
if not self.success:
raise ValueError(f"No token — solve failed: {self.request}")
return self.request
class SolveResult(BaseModel):
"""Result of a successful solve."""
token: str
task_id: str
solve_time: float = Field(description="Solve time in seconds")
封装 CaptchaAI 客户端
客户端类把提交、轮询、错误处理串起来：调用方只传原始参数，模型校验、参数转换和轮询逻辑都在内部完成。_submit 负责提交并拿到 task_id，_poll 按 poll_interval 轮询直到出结果或超时，四个 solve_* 方法只是实例化对应请求模型后交给 _solve。
# client.py
import time
import requests
from pydantic import ValidationError
from models import (
RecaptchaV2Request,
RecaptchaV3Request,
TurnstileRequest,
ImageRequest,
SubmitResponse,
PollResponse,
SolveResult,
)
SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"
class CaptchaAIError(Exception):
def __init__(self, code: str, message: str = ""):
self.code = code
super().__init__(f"{code}: {message}" if message else code)
class CaptchaAI:
def __init__(self, api_key: str, poll_interval: int = 5, timeout: int = 180):
if not api_key or len(api_key) < 10:
raise ValueError("Invalid API key")
self.api_key = api_key
self.poll_interval = poll_interval
self.timeout = timeout
def _submit(self, params: dict) -> str:
params["key"] = self.api_key
params["json"] = 1
resp = requests.post(SUBMIT_URL, data=params, timeout=30)
result = SubmitResponse.model_validate(resp.json())
if not result.success:
raise CaptchaAIError(result.request, "Submit failed")
return result.task_id
def _poll(self, task_id: str) -> str:
start = time.monotonic()
while time.monotonic() - start < self.timeout:
time.sleep(self.poll_interval)
resp = requests.get(RESULT_URL, params={
"key": self.api_key,
"action": "get",
"id": task_id,
"json": 1,
}, timeout=15)
result = PollResponse.model_validate(resp.json())
if not result.ready:
continue
if result.success:
return result.token
raise CaptchaAIError(result.request, "Solve failed")
raise CaptchaAIError("TIMEOUT", f"Task {task_id} timed out after {self.timeout}s")
def _solve(self, params: dict) -> SolveResult:
start = time.monotonic()
task_id = self._submit(params)
token = self._poll(task_id)
elapsed = time.monotonic() - start
return SolveResult(
token=token,
task_id=task_id,
solve_time=round(elapsed, 1),
)
def solve_recaptcha_v2(self, sitekey: str, pageurl: str, **kwargs) -> SolveResult:
"""Solve reCAPTCHA v2 with validated parameters."""
req = RecaptchaV2Request(sitekey=sitekey, pageurl=pageurl, **kwargs)
return self._solve(req.to_params())
def solve_recaptcha_v3(self, sitekey: str, pageurl: str, **kwargs) -> SolveResult:
"""Solve reCAPTCHA v3 with validated parameters."""
req = RecaptchaV3Request(sitekey=sitekey, pageurl=pageurl, **kwargs)
return self._solve(req.to_params())
def solve_turnstile(self, sitekey: str, pageurl: str, **kwargs) -> SolveResult:
"""Solve Cloudflare Turnstile with validated parameters."""
req = TurnstileRequest(sitekey=sitekey, pageurl=pageurl, **kwargs)
return self._solve(req.to_params())
def solve_image(self, base64_image: str, **kwargs) -> SolveResult:
"""Solve image/text CAPTCHA with validated parameters."""
req = ImageRequest(base64_image=base64_image, **kwargs)
return self._solve(req.to_params())
def get_balance(self) -> float:
"""Get current account balance."""
resp = requests.get(RESULT_URL, params={
"key": self.api_key,
"action": "getbalance",
"json": 1,
}, timeout=10)
result = SubmitResponse.model_validate(resp.json())
return float(result.request)
调用示例
下面几段代码分别演示：参数校验通过、正常发起识别；sitekey 太短、在本地就被拦截；reCAPTCHA v3 的校验对照场景；以及请求真正发到 CaptchaAI 之后才暴露的业务错误（CaptchaAIError）。这几种情况接入时要分开处理，不能用同一段 except 一网打尽。
from pydantic import ValidationError
from client import CaptchaAI, CaptchaAIError
client = CaptchaAI("YOUR_API_KEY", timeout=120)
# Valid request — passes validation, calls API
result = client.solve_recaptcha_v2(
sitekey="6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
pageurl="https://staging.example.com/qa-login",
)
print(f"Token: {result.token[:40]}...")
print(f"Solved in {result.solve_time}s")
# Invalid sitekey — caught immediately, no API call
try:
client.solve_recaptcha_v2(sitekey="", pageurl="https://example.com")
except ValidationError as e:
print(e)
# sitekey: String should have at least 20 characters
# Invalid score — caught before API call
try:
client.solve_recaptcha_v3(
sitekey="6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
pageurl="https://example.com",
)
except ValidationError as e:
print(e)
# API error — caught during request
try:
result = client.solve_turnstile(
sitekey="0x4AAAAAAADnPIDROrmt1Wwj",
pageurl="https://example.com",
)
except CaptchaAIError as e:
print(f"API error: {e.code}")
安装依赖项：
pip install pydantic requests
国内网络安装较慢时可加镜像参数，例如
-i https://pypi.tuna.tsinghua.edu.cn/simple。
常见报错排查
把最容易踩的几类报错单独列出来，对着标题找。
看起来正常的 sitekey，也报 ValidationError
大概率是站点密钥长度不足 20 个字符。先确认目标站点 sitekey 的实际长度；如果它本来就短，调整模型里的 min_length 即可。
pageurl 报 ValidationError
多半是 URL 缺了协议前缀，补上 https:// 就能过。
Base64 图片校验失败
字符串太短，或者混入了 data: 前缀没剥干净。校验器会自动剥离 data: 前缀，确认真正的 Base64 内容超过 100 字符。
CaptchaAIError: ERROR_ZERO_BALANCE
账户余额不足，去 CaptchaAI 控制台充值即可解决。
明明装了 Pydantic，导入方式却在报错
大概率装的是 Pydantic v1。换成 Pydantic v2：pip install 'pydantic>=2.0'。
常见问题
pageurl 字段为什么用 HttpUrl 而不是普通字符串？
HttpUrl 在赋值阶段就会校验协议和格式，缺少 http(s):// 前缀或者格式不对的地址会立刻抛出 ValidationError，不用等提交任务之后才从 CaptchaAI 拿到错误码。序列化成参数时用 str(self.pageurl) 转回字符串即可。
Pydantic 校验会不会拖慢批量识别任务？
几乎不会。单次校验是微秒级开销，跟 API 往返的秒级延迟比可以忽略不计，省下来的时间远超验证本身的成本。
能不能换成异步的 httpx 客户端？
可以。把 requests 换成 httpx.AsyncClient，_submit、_poll 和各个 solve_* 方法改成 async def 即可。Pydantic 模型不用动——校验本来就发生在异步请求之前。
CaptchaMethod 里有 GEETEST，为什么客户端没有 solve_geetest 方法？
本文的模型只实现了最常用的四种类型。CaptchaAI 支持到 GeeTest v3（v4 暂不支持），要补的话，照同样模式新建 GeetestRequest 模型和 to_params()，再加一个 solve_geetest 方法即可。
相关文章
下一步
现在就动手搭建带校验的 CaptchaAI 客户端——获取你的 API 密钥，把这套 Pydantic 模型接进你自己的项目。
相关指南：
