Django 应用程序经常需要在两种情况下处理验证码：在自己的表单上验证验证码（防止机器人）和解决外部站点上的验证码（数据收集、测试、自动化）。本指南涵盖了使用 CaptchaAI 的两种模式。
场景 1：验证 Django 表单上的验证码
当您将 Turnstile 或 reCAPTCHA 添加到 Django 表单时，您需要在服务器端验证令牌。
将 Turnstile 添加到 Django 表单
# forms.py
from django import forms
class ContactForm(forms.Form):
name = forms.CharField(max_length=100)
email = forms.EmailField()
message = forms.CharField(widget=forms.Textarea)
cf_turnstile_response = forms.CharField(
widget=forms.HiddenInput(),
required=True,
)
# views.py
import requests
from django.conf import settings
from django.shortcuts import render, redirect
from .forms import ContactForm
def contact_view(request):
if request.method == "POST":
form = ContactForm(request.POST)
if form.is_valid():
# Verify Turnstile token with Cloudflare
token = form.cleaned_data["cf_turnstile_response"]
verification = requests.post(
"https://challenges.cloudflare.com/turnstile/v0/siteverify",
data={
"secret": settings.TURNSTILE_SECRET_KEY,
"response": token,
"remoteip": request.META.get("REMOTE_ADDR"),
},
).json()
if verification.get("success"):
# Process the form
return redirect("success")
else:
form.add_error(None, "CAPTCHA verification failed")
else:
form = ContactForm()
return render(request, "contact.html", {
"form": form,
"turnstile_sitekey": settings.TURNSTILE_SITE_KEY,
})
<!-- templates/contact.html -->
<form method="post">
{% csrf_token %}
{{ form.as_p }}
<div class="cf-turnstile" data-sitekey="{{ turnstile_sitekey }}"></div>
<button type="submit">Send</button>
</form>
<script src="https://challenges.cloudflare.com/turnstile/v0/api.js" async defer></script>
场景 2：解决外部站点上的验证码
这就是 CaptchaAI 的用武之地——当您的 Django 应用程序需要与受验证码保护的外部站点交互时。
CaptchaAI服务等级
# services/captcha_solver.py
import time
import requests
from django.conf import settings
class CaptchaSolverService:
"""Django service for solving CAPTCHAs via CaptchaAI."""
API_BASE = "https://ocr.captchaai.com"
def __init__(self):
self.api_key = settings.CAPTCHAAI_API_KEY
def solve_recaptcha_v2(self, sitekey, page_url, invisible=False):
"""Solve reCAPTCHA v2."""
params = {
"key": self.api_key,
"method": "userrecaptcha",
"googlekey": sitekey,
"pageurl": page_url,
"json": 1,
}
if invisible:
params["invisible"] = 1
return self._submit_and_poll(params)
def solve_turnstile(self, sitekey, page_url, action=None):
"""Solve Cloudflare Turnstile."""
params = {
"key": self.api_key,
"method": "turnstile",
"sitekey": sitekey,
"pageurl": page_url,
"json": 1,
}
if action:
params["action"] = action
return self._submit_and_poll(params)
def solve_image(self, image_base64):
"""Solve image/text CAPTCHA."""
return self._submit_and_poll({
"key": self.api_key,
"method": "base64",
"body": image_base64,
"json": 1,
})
def get_balance(self):
"""Check API balance."""
response = requests.get(f"{self.API_BASE}/res.php", params={
"key": self.api_key,
"action": "getbalance",
"json": 1,
}, timeout=30)
return float(response.json().get("request", 0))
def _submit_and_poll(self, params, timeout=120):
"""Submit task and poll for result."""
# Submit
response = requests.post(f"{self.API_BASE}/in.php", data=params, timeout=30)
response.raise_for_status()
data = response.json()
if data.get("status") != 1:
raise CaptchaSolveError(f"Submit failed: {data.get('request')}")
task_id = data["request"]
# Poll
start = time.time()
while time.time() - start < timeout:
time.sleep(5)
result = requests.get(f"{self.API_BASE}/res.php", params={
"key": self.api_key,
"action": "get",
"id": task_id,
"json": 1,
}, timeout=30).json()
if result.get("status") == 1:
return result["request"]
if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
raise CaptchaSolveError("CAPTCHA unsolvable")
raise CaptchaSolveError("Solve timed out")
class CaptchaSolveError(Exception):
pass
Django 设置
# settings.py
CAPTCHAAI_API_KEY = "YOUR_API_KEY"
TURNSTILE_SITE_KEY = "0x4AAAAAAAC3DHQhMMQ_Rxrg"
TURNSTILE_SECRET_KEY = "0x4AAAAAAAC3DHQhYYY_secret"
在视图中使用服务
查看外部数据收集
# views.py
from django.http import JsonResponse
from django.views.decorators.http import require_POST
from .services.captcha_solver import CaptchaSolverService, CaptchaSolveError
@require_POST
def scrape_external_data(request):
"""Solve CAPTCHA and fetch data from external CAPTCHA-protected site."""
url = request.POST.get("target_url")
if not url:
return JsonResponse({"error": "target_url required"}, status=400)
solver = CaptchaSolverService()
try:
# Solve the CAPTCHA
token = solver.solve_turnstile(
sitekey="0x4AAAAAAAC3DHQhMMQ_Rxrg",
page_url=url,
)
# Use token to access the protected resource
import requests as http_requests
response = http_requests.post(url, data={
"cf-turnstile-response": token,
}, timeout=30)
return JsonResponse({
"status": "success",
"data": response.text[:1000],
})
except CaptchaSolveError as e:
return JsonResponse({"error": str(e)}, status=500)
Django管理命令
# management/commands/solve_captcha.py
from django.core.management.base import BaseCommand
from myapp.services.captcha_solver import CaptchaSolverService
class Command(BaseCommand):
help = "Solve a CAPTCHA and print the token"
def add_arguments(self, parser):
parser.add_argument("--type", choices=["recaptcha", "turnstile"], required=True)
parser.add_argument("--sitekey", required=True)
parser.add_argument("--url", required=True)
def handle(self, *args, **options):
solver = CaptchaSolverService()
self.stdout.write(f"Solving {options['type']} for {options['url']}...")
if options["type"] == "recaptcha":
token = solver.solve_recaptcha_v2(options["sitekey"], options["url"])
else:
token = solver.solve_turnstile(options["sitekey"], options["url"])
self.stdout.write(self.style.SUCCESS(f"Token: {token[:50]}..."))
# Check balance
balance = solver.get_balance()
self.stdout.write(f"Remaining balance: ${balance:.2f}")
用法：
python manage.py solve_captcha --type turnstile --sitekey 0x4AAA... --url https://example.com
带有 CaptchaAI 的异步 Django
Django 4.1+ 支持异步视图：
# views.py (async)
import aiohttp
import asyncio
from django.http import JsonResponse
CAPTCHAAI_API_KEY = "YOUR_API_KEY"
async def solve_captcha_async(request):
"""Async view for solving CAPTCHAs."""
sitekey = request.GET.get("sitekey")
page_url = request.GET.get("url")
if not sitekey or not page_url:
return JsonResponse({"error": "sitekey and url required"}, status=400)
async with aiohttp.ClientSession() as session:
# Submit
async with session.post("https://ocr.captchaai.com/in.php", data={
"key": CAPTCHAAI_API_KEY,
"method": "turnstile",
"sitekey": sitekey,
"pageurl": page_url,
"json": 1,
}) as resp:
data = await resp.json()
if data.get("status") != 1:
return JsonResponse({"error": data.get("request")}, status=500)
task_id = data["request"]
# Poll
for _ in range(30):
await asyncio.sleep(5)
async with session.get("https://ocr.captchaai.com/res.php", params={
"key": CAPTCHAAI_API_KEY,
"action": "get",
"id": task_id,
"json": 1,
}) as resp:
result = await resp.json()
if result.get("status") == 1:
return JsonResponse({"token": result["request"]})
return JsonResponse({"error": "timeout"}, status=504)
Celery 集成用于后台解决
对于长时间运行的验证码解决方案，请使用 Celery：
# tasks.py
from celery import shared_task
from .services.captcha_solver import CaptchaSolverService, CaptchaSolveError
@shared_task(bind=True, max_retries=2, default_retry_delay=10)
def solve_captcha_task(self, captcha_type, sitekey, page_url):
"""Background CAPTCHA solving with Celery."""
solver = CaptchaSolverService()
try:
if captcha_type == "recaptcha_v2":
token = solver.solve_recaptcha_v2(sitekey, page_url)
elif captcha_type == "turnstile":
token = solver.solve_turnstile(sitekey, page_url)
else:
raise ValueError(f"Unknown type: {captcha_type}")
return {"success": True, "token": token}
except CaptchaSolveError as e:
self.retry(exc=e)
# Usage in views
from .tasks import solve_captcha_task
def start_solve(request):
result = solve_captcha_task.delay("turnstile", "0x4AAA...", "https://example.com")
return JsonResponse({"task_id": result.id})
def check_solve(request, task_id):
from celery.result import AsyncResult
result = AsyncResult(task_id)
if result.ready():
return JsonResponse(result.get())
return JsonResponse({"status": "pending"})
故障排除
| 症状 | 原因 | 处理方式 |
|---|---|---|
CaptchaSolveError 生产中 |
API 密钥不在设置中 | 将 CAPTCHAAI_API_KEY 添加到 Django 设置 |
| Celery 任务无限重试 | 无法解析的验证码或错误的站点密钥 | 设置 max_retries 并验证输入 |
| 异步视图挂起 | 在异步视图中同步代码 | 使用 aiohttp 代替 requests |
| 表单提交前令牌已过期 | 解决时间太长 | 及时解决，而不是提前解决 |
| 管理命令导入错误 | 服务不在 INSTALLED_APPS 中 | 检查应用程序注册情况 |
常见问题
验证码解决应该同步还是异步？
使用 Celery 进行面向 Web 的视图，这样用户就不会等待 15 秒以上。在管理命令和后台脚本中使用同步求解。
如何安全地存储 API 密钥？
使用环境变量或 Django 的 django-environ 包。切勿将 API 密钥提交给版本控制。
我可以缓存已解决的令牌吗？
reCAPTCHA 令牌将在 120 秒内过期，Turnstile 令牌将在 300 秒内过期。缓存不实用——在使用前解决。
我应该创建一个服务实例还是使用单例？
CaptchaSolverService 类是无状态的。每个请求创建一个新实例或使用 Django 的依赖注入模式。
概括
Django 应用程序集成CaptchaAI通过包装submit/poll 流程的服务类。在管理命令中使用同步求解，在 Django 4.1+ 异步视图中使用异步求解，以及用于后台处理的 Celery 任务。相同的服务处理 reCAPTCHA、Turnstile 和图像验证码。
相关文章
- Geetest 与 Cloudflare Turnstile 比较
- Cloudflare Turnstile 403 令牌修复后
- Cloudflare Turnstile 小部件模式解释
