Django xử lý CAPTCHA theo hai hướng khác nhau, và bài này đi thẳng vào code cho cả hai:
- Xác minh — kiểm tra token gửi lên từ chính form Django của bạn (chặn bot đăng ký, spam liên hệ).
- Giải — chủ động vượt CAPTCHA trên site bên thứ ba bằng CaptchaAI (thu thập dữ liệu, test tự động).
Xác minh token CAPTCHA ngay trên form Django
Khi gắn Turnstile hoặc reCAPTCHA vào form, token phải được xác minh ở phía server. Xác minh riêng ở phía client không đủ an toàn vì request POST có thể bị giả mạo, bỏ qua widget:
# 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>
Giải CAPTCHA trên site bên ngoài bằng CaptchaAI
Đây là phần CaptchaAI thật sự cần thiết: khi Django app phải vượt qua CAPTCHA trên trang không do bạn kiểm soát — site nguồn dữ liệu cho scraping, hoặc portal đối tác cần test tự động. Đóng gói logic submit/poll vào một service class riêng:
# 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
Ba method solve đều gọi chung _submit_and_poll — gửi task tới in.php, nhận task_id, rồi polling res.php mỗi 5 giây tới khi có kết quả.
# settings.py
CAPTCHAAI_API_KEY = "YOUR_API_KEY"
TURNSTILE_SITE_KEY = "0x4AAAAAAAC3DHQhMMQ_Rxrg"
TURNSTILE_SECRET_KEY = "0x4AAAAAAAC3DHQhYYY_secret"
Đặt CAPTCHAAI_API_KEY qua biến môi trường, không hard-code trong file settings đưa lên Git.
Gọi service từ view và management command
Có hai điểm gọi service phổ biến trong một Django app thực tế:
- View xử lý request thu thập dữ liệu từ site ngoài.
- Management command chạy tay từ terminal để test nhanh sitekey.
View thu thập dữ liệu từ site ngoài
# 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)
Đội automation ở các công ty outsourcing tại TP.HCM/Hà Nội hay dựng pattern này để theo dõi giá trên Shopee, Lazada, Tiki.
Command dòng lệnh để test nhanh
# 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}")
Cách sử dụng:
python manage.py solve_captcha --type turnstile --sitekey 0x4AAA... --url https://example.com
Hữu ích để kiểm tra nhanh sitekey và số dư trước khi gắn logic vào view hoặc task nền.
Async view với Django 4.1+
Django 4.1 trở lên hỗ trợ async view trực tiếp, không cần thư viện ngoài:
# 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)
Lỗi hay gặp: gọi requests bên trong async view sẽ block event loop — luôn dùng aiohttp.
Giải CAPTCHA nền bằng Celery
Solve mất vài chục giây không nên chặn request HTTP — đẩy sang Celery task nền:
# 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"})
max_retries=2 chặn Celery retry vô hạn khi sitekey sai hoặc CAPTCHA không giải được — tránh tốn thread trong gói CaptchaAI vô ích.
Câu hỏi thường gặp
Nên giải CAPTCHA đồng bộ hay bất đồng bộ trong Django?
Dùng Celery cho view web để người dùng không phải chờ quá 15 giây. Dùng giải đồng bộ trong management command và script nền, nơi không ai đang đợi response.
CaptchaAI có hỗ trợ hCaptcha trong ứng dụng Django không?
Chưa. CaptchaAI hỗ trợ reCAPTCHA v2/v3, Cloudflare Turnstile, GeeTest v3 và CAPTCHA hình ảnh; hCaptcha và FunCaptcha (Arkose Labs) chưa được hỗ trợ.
Bao nhiêu thread là đủ cho một Django app traffic vừa?
Với lượng scrape hoặc test vừa phải, gói STANDARD ($30/tháng, 15 thread) thường đủ — thread chỉ giới hạn số CAPTCHA giải đồng thời, không giới hạn tổng số lần giải trong tháng.
Celery task cứ retry mãi thì kiểm tra gì trước?
Kiểm tra sitekey và page_url truyền vào task trước — sai sitekey khiến CaptchaAI luôn báo lỗi, Celery hiểu nhầm là cần retry.
Có nên cache token đã giải để dùng lại không?
Không nên. Token reCAPTCHA hết hạn sau 120 giây, Turnstile sau 300 giây — hãy giải ngay trước khi dùng.
Các lỗi thường gặp và cách xử lý
| Triệu chứng | Nguyên nhân | Cách xử lý |
|---|---|---|
CaptchaSolveError khi chạy production |
Thiếu CAPTCHAAI_API_KEY |
Thêm biến vào settings qua biến môi trường |
| Celery task retry liên tục | Sai sitekey hoặc CAPTCHA không giải được | Đặt max_retries hợp lý, validate input |
| Async view bị treo | Gọi requests trong async view |
Dùng aiohttp thay cho requests |
| Token hết hạn trước khi submit | Solve mất quá nhiều thời gian | Giải ngay trước khi dùng |
| Lỗi import trong management command | Service chưa có trong INSTALLED_APPS |
Kiểm tra đăng ký app |
Tóm tắt nhanh
Django tích hợp CaptchaAI qua một service class bao bọc luồng submit/poll:
- Đồng bộ trong management command và script chạy nền.
- Async view (Django 4.1+) khi cần không chặn event loop.
- Celery task cho job giải CAPTCHA chạy nền dài, có retry.
Cùng một service class xử lý reCAPTCHA, Turnstile và CAPTCHA hình ảnh.
