Tích Hợp

Xây dựng vi dịch vụ giải CAPTCHA với FastAPI và CaptchaAI

Tự động hóa Python Tất cả loại CAPTCHA
Por Đội ngũ CaptchaAI Việt Nam
Publicado em Jun 02, 2026
Xây dựng vi dịch vụ giải CAPTCHA với FastAPI và CaptchaAI
Imagem em destaque indisponível

Khi nhiều dịch vụ hoặc thành viên nhóm cần giải CAPTCHA, việc tập trung nó vào một vi dịch vụ sẽ tránh logic trùng lặp giữa các dự án. Khả năng hỗ trợ async của FastAPI khiến nó trở nên rất phù hợp - việc giải CAPTCHA bao gồm việc chờ phản hồi API bên ngoài, async xử lý hiệu quả mà không chặn các luồng.

Hướng dẫn này xây dựng một vi dịch vụ FastAPI chấp nhận các yêu cầu giải CAPTCHA qua REST và trả về mã thông báo đã giải quyết thông qua CaptchaAI.

Những gì bạn cần

Yêu cầu Chi tiết
Khóa API CaptchaAI captchaai.com
Python 3.9+
FastAPI + httpx Để xử lý HTTP không đồng bộ

Cài đặt phụ thuộc:

pip install fastapi uvicorn httpx

Cấu trúc dự án

captcha-service/
├── main.py          # FastAPI app with endpoints
├── solver.py        # CaptchaAI solving logic
└── requirements.txt

Mô-đun giải CaptchaAI

# solver.py
import httpx
import asyncio

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"

async def submit_task(params: dict) -> str:
    """Submit a CAPTCHA task and return the task ID."""
    params["key"] = API_KEY
    params["json"] = 1

    async with httpx.AsyncClient() as client:
        response = await client.post(f"{BASE_URL}/in.php", data=params)
        data = response.json()

    if data.get("status") != 1:
        raise ValueError(f"Submit error: {data.get('request')}")
    return data["request"]

async def poll_result(task_id: str, initial_wait: int = 15, max_attempts: int = 30) -> dict:
    """Poll for the CAPTCHA result."""
    await asyncio.sleep(initial_wait)

    async with httpx.AsyncClient() as client:
        for _ in range(max_attempts):
            response = await client.get(f"{BASE_URL}/res.php", params={
                "key": API_KEY, "action": "get", "id": task_id, "json": 1
            })
            data = response.json()

            if data.get("status") == 1:
                return {
                    "token": data["request"],
                    "user_agent": data.get("user_agent", "")
                }
            if data.get("request") != "CAPCHA_NOT_READY":
                raise ValueError(f"Solve error: {data['request']}")

            await asyncio.sleep(5)

    raise TimeoutError("Solve timed out")

async def solve_recaptcha_v2(sitekey: str, pageurl: str, enterprise: bool = False) -> dict:
    params = {"method": "userrecaptcha", "googlekey": sitekey, "pageurl": pageurl}
    if enterprise:
        params["enterprise"] = 1
    task_id = await submit_task(params)
    return await poll_result(task_id, initial_wait=20)

async def solve_recaptcha_v3(sitekey: str, pageurl: str, action: str, enterprise: bool = False) -> dict:
    params = {
        "method": "userrecaptcha", "version": "v3",
        "googlekey": sitekey, "pageurl": pageurl, "action": action
    }
    if enterprise:
        params["enterprise"] = 1
    task_id = await submit_task(params)
    return await poll_result(task_id, initial_wait=20)

async def solve_turnstile(sitekey: str, pageurl: str) -> dict:
    task_id = await submit_task({"method": "turnstile", "sitekey": sitekey, "pageurl": pageurl})
    return await poll_result(task_id, initial_wait=10)

async def solve_image(image_base64: str) -> dict:
    task_id = await submit_task({"method": "base64", "body": image_base64})
    return await poll_result(task_id, initial_wait=5, max_attempts=15)

Ứng dụng FastAPI

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import solver

app = FastAPI(title="CaptchaAI Solver Service")

class RecaptchaV2Request(BaseModel):
    sitekey: str
    pageurl: str
    enterprise: bool = False

class RecaptchaV3Request(BaseModel):
    sitekey: str
    pageurl: str
    action: str
    enterprise: bool = False

class TurnstileRequest(BaseModel):
    sitekey: str
    pageurl: str

class ImageRequest(BaseModel):
    image_base64: str

class SolveResponse(BaseModel):
    token: str
    user_agent: Optional[str] = ""

@app.post("/solve/recaptcha-v2", response_model=SolveResponse)
async def solve_recaptcha_v2(req: RecaptchaV2Request):
    try:
        result = await solver.solve_recaptcha_v2(req.sitekey, req.pageurl, req.enterprise)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))

@app.post("/solve/recaptcha-v3", response_model=SolveResponse)
async def solve_recaptcha_v3(req: RecaptchaV3Request):
    try:
        result = await solver.solve_recaptcha_v3(req.sitekey, req.pageurl, req.action, req.enterprise)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))

@app.post("/solve/turnstile", response_model=SolveResponse)
async def solve_turnstile(req: TurnstileRequest):
    try:
        result = await solver.solve_turnstile(req.sitekey, req.pageurl)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))

@app.post("/solve/image", response_model=SolveResponse)
async def solve_image(req: ImageRequest):
    try:
        result = await solver.solve_image(req.image_base64)
        return SolveResponse(**result)
    except (ValueError, TimeoutError) as e:
        raise HTTPException(status_code=502, detail=str(e))

@app.get("/health")
async def health():
    return {"status": "ok"}

Chạy dịch vụ

uvicorn main:app --host 0.0.0.0 --port 8000

Ví dụ sử dụng

Giải reCAPTCHA v2

curl -X POST http://localhost:8000/solve/recaptcha-v2 \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "6Le-wvkS...", "pageurl": "https://staging.example.com/qa-login"}'

Giải Cloudflare Turnstile

curl -X POST http://localhost:8000/solve/turnstile \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "0x4AAAA...", "pageurl": "https://example.com/form"}'

Trả lời:

{
  "token": "03AGdBq24PBCqLmOx2V4...",
  "user_agent": "Mozilla/5.0..."
}

Ghi chú tăng cường triển khai

  • Đọc khóa API từ môi trường và nhanh chóng bị lỗi trong quá trình khởi động nếu thiếu bí mật.
  • Tách biệt việc xác thực yêu cầu khỏi việc thực thi bộ giải để các tải trọng không hợp lệ không bao giờ đến được đường dẫn cuộc gọi đi.
  • Trả về các phản hồi lỗi có cấu trúc để phân biệt các lỗi xác thực, lỗi của bộ giải và từ chối ngược dòng.

Khắc phục sự cố

Vấn đề Nguyên nhân Cách xử lý
Phản hồi 502 CaptchaAI trả về lỗi Kiểm tra trường detail để biết lỗi cụ thể
Hết thời gian giải quyết CAPTCHA mất quá nhiều thời gian Tăng max_attempts hoặc kiểm tra trạng thái CaptchaAI
Kết nối bị từ chối Dịch vụ không chạy Xác minh uvicorn đang chạy trên cổng dự kiến
Phản hồi chậm Chặn I/O Đảm bảo httpx.AsyncClient được sử dụng, không phải requests

Câu hỏi thường gặp

Tại sao nên sử dụng FastAPI cho dịch vụ giải CAPTCHA?

FastAPI xử lý I/O không đồng bộ một cách tự nhiên, điều này lý tưởng cho việc giải CAPTCHA khi phần lớn thời gian được dành để chờ phản hồi của CaptchaAI. Nhiều yêu cầu có thể được xử lý đồng thời mà không cần phân luồng.

Tôi có thể thêm xác thực vào điểm cuối không?

Vâng. Thêm nội dung phụ thuộc của FastAPI bằng xác thực tiêu đề khóa API hoặc OAuth2 để hạn chế quyền truy cập.

Có bao nhiêu yêu cầu đồng thời có thể xử lý được?

Bị giới hạn bởi giới hạn nhiệm vụ đồng thời của gói CaptchaAI của bạn. Bản thân FastAPI có thể xử lý hàng nghìn kết nối đồng thời.

Tôi có nên dockerize cái này không?

Vâng. Thêm Dockerfile với FROM python:3.11-slim, cài đặt các phần phụ thuộc và hiển thị cổng 8000.

Tôi có thể thêm giới hạn tỷ lệ không?

Vâng. Sử dụng slowapi hoặc proxy ngược (nginx, Traefik) để giới hạn yêu cầu trên mỗi khách hàng.

Xây dựng microservice giải CAPTCHA của bạn

Nhận khóa API của bạn tạicaptchaai.com. Tập trung giải quyết CAPTCHA bằng vi dịch vụ FastAPI.

Hướng dẫn liên quan

Os comentários estão desativados para este artigo.