返回博客教程
用Python解决reCAPTCHA v3:分步教程
学习如何使用异步请求、重试逻辑和生产就绪的错误处理将验证码解决集成到Python脚本中。
reGOTCHA团队2025年12月19日6分钟 阅读
前提条件
- Python 3.8 或更高版本
- 带有 API 密钥的 reGOTCHA 账户
- 对异步编程的基本理解
安装
terminal
pip install httpx asyncio基本同步示例
从简单的同步实现开始:
example.py
import httpx
import time
API_KEY = "your-api-key"
BASE_URL = "https://api.regotcha.com"
def solve_recaptcha(website_url: str, website_key: str, action: str = "submit"):
# 创建任务
response = httpx.post(
f"{BASE_URL}/createTask",
json={
"clientKey": API_KEY,
"task": {
"type": "ReCaptchaV3EnterpriseTaskProxyless",
"websiteURL": website_url,
"websiteKey": website_key,
"pageAction": action
}
}
)
task_id = response.json()["taskId"]
# 轮询结果
while True:
result = httpx.post(
f"{BASE_URL}/getTaskResult",
json={"clientKey": API_KEY, "taskId": task_id}
).json()
if result["status"] == "ready":
return result["solution"]["gRecaptchaResponse"]
time.sleep(2)
# 使用
token = solve_recaptcha(
"https://example.com",
"6LcR_RIpAAAAANGDwXhTz8nO4N_NijF_Pj2iO4L6",
"login"
)
print(f"Token: {token}")生产就绪的异步实现
对于生产工作负载,使用带有正确错误处理的异步:
example.py
import httpx
import asyncio
from typing import Optional
class RecaptchaSolver:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.regotcha.com"
async def solve(
self,
website_url: str,
website_key: str,
action: str = "submit",
timeout: int = 120
) -> Optional[str]:
async with httpx.AsyncClient() as client:
# 创建任务
response = await client.post(
f"{self.base_url}/createTask",
json={
"clientKey": self.api_key,
"task": {
"type": "ReCaptchaV3EnterpriseTaskProxyless",
"websiteURL": website_url,
"websiteKey": website_key,
"pageAction": action
}
}
)
data = response.json()
if data.get("errorId"):
raise Exception(f"API 错误: {data.get('errorDescription')}")
task_id = data["taskId"]
# 带超时轮询
start = asyncio.get_event_loop().time()
while asyncio.get_event_loop().time() - start < timeout:
result = await client.post(
f"{self.base_url}/getTaskResult",
json={"clientKey": self.api_key, "taskId": task_id}
)
result_data = result.json()
if result_data["status"] == "ready":
return result_data["solution"]["gRecaptchaResponse"]
if result_data.get("errorId"):
raise Exception(f"解决错误: {result_data.get('errorDescription')}")
await asyncio.sleep(2)
raise TimeoutError("验证码解决超时")
# 使用
async def main():
solver = RecaptchaSolver("your-api-key")
token = await solver.solve(
"https://example.com",
"6LcR_RIpAAAAANGDwXhTz8nO4N_NijF_Pj2iO4L6",
"checkout"
)
print(f"Token: {token}")
asyncio.run(main())批量处理
并发解决多个验证码:
example.py
async def solve_batch(solver, tasks):
"""并发解决多个验证码"""
return await asyncio.gather(
*[solver.solve(**task) for task in tasks],
return_exceptions=True
)
# 示例:并行解决 10 个验证码
tasks = [
{"website_url": "https://site1.com", "website_key": "key1", "action": "login"},
{"website_url": "https://site2.com", "website_key": "key2", "action": "signup"},
# ... 更多任务
]
results = await solve_batch(solver, tasks)性能提示:对于高容量工作负载,保持 AsyncClient 连接 并在请求之间复用以减少连接开销。
Python教程API代码