Когда нескольким службам или членам команды требуется решение CAPTCHA, централизация этого процесса в микросервисе позволяет избежать дублирования логики в разных проектах. Поддержка асинхронности FastAPI делает его идеальным решением — решение CAPTCHA предполагает ожидание ответов внешнего API, которые асинхронность эффективно обрабатывает, не блокируя потоки.
В этом руководстве создается микросервис FastAPI, который принимает запросы на решение CAPTCHA через REST и возвращает решенные токены через CaptchaAI.
Что вам нужно
| Требование | Подробности |
|---|---|
| CaptchaAI API-ключ | captchaai.com |
| Питон 3.9+ | |
| Быстрый API + httpx | Для асинхронной обработки HTTP |
Установите зависимости:
pip install fastapi uvicorn httpx
Структура проекта
captcha-service/
├── main.py # FastAPI app with endpoints
├── solver.py # CaptchaAI solving logic
└── requirements.txt
Решающий модуль 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)
Приложение 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"}
Запустите службу
uvicorn main:app --host 0.0.0.0 --port 8000
Примеры использования
Решить reCAPTCHA v2
curl -X POST http://localhost:8000/solve/recaptcha-v2 \
-H "Content-Type: application/json" \
-d '{"sitekey": "6Le-wvkS...", "pageurl": "https://https://staging.example.com/qa-login"}'
Решить Cloudflare Turnstile
curl -X POST http://localhost:8000/solve/turnstile \
-H "Content-Type: application/json" \
-d '{"sitekey": "0x4AAAA...", "pageurl": "https://example.com/form"}'
Ответ:
{
"token": "03AGdBq24PBCqLmOx2V4...",
"user_agent": "Mozilla/5.0..."
}
Поиск неисправностей
| Проблема | Причина | Исправить |
|---|---|---|
| 502 ответ | CaptchaAI вернул ошибку. | Проверьте поле detail на наличие конкретной ошибки. |
| Таймаут на решение | Капча занимала слишком много времени | Увеличьте max_attempts или проверьте статус CaptchaAI. |
| В соединении отказано | Служба не работает | Убедитесь, что uvicorn работает на ожидаемом порту. |
| Медленные ответы | Блокировка I/O | Убедитесь, что используется httpx.AsyncClient, а не requests. |
Замечания по усилению безопасности развертывания
- Считайте ключ API из среды и быстро выполните сбой во время запуска, если секрет отсутствует.
- Отделите проверку запроса от выполнения решателя, чтобы недействительные полезные данные никогда не доходили до пути исходящего вызова.
- Возвращайте структурированные ответы об ошибках, которые различают ошибки проверки, ошибки решателя и отклонение восходящего потока.
Часто задаваемые вопросы
Зачем использовать FastAPI для службы решения CAPTCHA?
FastAPI изначально обрабатывает асинхронный I/O, что идеально подходит для решения CAPTCHA, когда большая часть времени тратится на ожидание ответа CaptchaAI. Несколько запросов могут обрабатываться одновременно без потоковой обработки.
Могу ли я добавить аутентификацию к конечным точкам?
Да. Добавьте внедрение зависимостей FastAPI с проверкой заголовка ключа API или OAuth2 для ограничения доступа.
Сколько одновременных запросов это может обработать?
Ограничено лимитом одновременных задач вашего плана CaptchaAI. Сам FastAPI может обрабатывать тысячи одновременных соединений.
Должен ли я это докеризовать?
Да. Добавьте Dockerfile с FROM python:3.11-slim, установите зависимости и откройте порт 8000.
Могу ли я добавить ограничение скорости?
Да. Используйте slowapi или обратный прокси-сервер (nginx, Traefik), чтобы ограничить количество запросов для каждого клиента.
Создайте свой микросервис для решения CAPTCHA
Получите ключ API по адресуcaptchaai.com. Централизуйте решение CAPTCHA с помощью микросервиса FastAPI.
Связанные руководства
- Как решить reCAPTCHA v2 с помощью API
- Как решить Cloudflare Turnstile с помощью API
- Асинхронный CaptchaAI с aiohttp
- Параллельное решение CAPTCHA
