File size: 4,116 Bytes
3a04f21 c00180e d272ff0 c00180e 3a04f21 d272ff0 3a04f21 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | """FastAPI application with Playwright lifecycle management."""
from __future__ import annotations
import logging
from contextlib import asynccontextmanager
from typing import AsyncIterator
from fastapi import FastAPI
from .api.routes import router
from .core.config import config
from .services.classification import ClassificationSolver
from .services.hcaptcha import HCaptchaSolver
from .services.recognition import CaptchaRecognizer
from .services.recaptcha_v2 import RecaptchaV2Solver
from .services.recaptcha_v3 import RecaptchaV3Solver
from .services.task_manager import task_manager
from .services.turnstile import TurnstileSolver
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
log = logging.getLogger(__name__)
_RECAPTCHA_V3_TYPES = [
"RecaptchaV3TaskProxyless",
"RecaptchaV3TaskProxylessM1",
"RecaptchaV3TaskProxylessM1S7",
"RecaptchaV3TaskProxylessM1S9",
"RecaptchaV3EnterpriseTask",
"RecaptchaV3EnterpriseTaskM1",
]
_RECAPTCHA_V2_TYPES = [
"NoCaptchaTaskProxyless",
"RecaptchaV2TaskProxyless",
"RecaptchaV2EnterpriseTaskProxyless",
]
_HCAPTCHA_TYPES = [
"HCaptchaTaskProxyless",
]
_TURNSTILE_TYPES = [
"TurnstileTaskProxyless",
"TurnstileTaskProxylessM1",
]
_CLASSIFICATION_TYPES = [
"HCaptchaClassification",
"ReCaptchaV2Classification",
"FunCaptchaClassification",
"AwsClassification",
]
_IMAGE_TEXT_TYPES = [
"ImageToTextTask",
"ImageToTextTaskMuggle",
"ImageToTextTaskM1",
]
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
# ── startup ──
v3_solver = RecaptchaV3Solver(config)
for task_type in _RECAPTCHA_V3_TYPES:
task_manager.register_solver(task_type, v3_solver)
log.info(
"Registered reCAPTCHA v3 solver for types: %s (browser startup deferred)",
_RECAPTCHA_V3_TYPES,
)
v2_solver = RecaptchaV2Solver(config)
for task_type in _RECAPTCHA_V2_TYPES:
task_manager.register_solver(task_type, v2_solver)
log.info(
"Registered reCAPTCHA v2 solver for types: %s (browser startup deferred)",
_RECAPTCHA_V2_TYPES,
)
classifier = ClassificationSolver(config)
hcaptcha_solver = HCaptchaSolver(config, classifier=classifier)
for task_type in _HCAPTCHA_TYPES:
task_manager.register_solver(task_type, hcaptcha_solver)
log.info(
"Registered hCaptcha solver for types: %s (browser startup deferred)",
_HCAPTCHA_TYPES,
)
turnstile_solver = TurnstileSolver(config)
for task_type in _TURNSTILE_TYPES:
task_manager.register_solver(task_type, turnstile_solver)
log.info(
"Registered Turnstile solver for types: %s (browser startup deferred)",
_TURNSTILE_TYPES,
)
recognizer = CaptchaRecognizer(config)
for task_type in _IMAGE_TEXT_TYPES:
task_manager.register_solver(task_type, recognizer)
log.info("Registered image captcha recognizer for types: %s", _IMAGE_TEXT_TYPES)
for task_type in _CLASSIFICATION_TYPES:
task_manager.register_solver(task_type, classifier)
log.info("Registered classification solver for types: %s", _CLASSIFICATION_TYPES)
yield
# ── shutdown ──
await v3_solver.stop()
await v2_solver.stop()
await hcaptcha_solver.stop()
await turnstile_solver.stop()
app = FastAPI(
title="Captcha Solver Service",
version="3.0.0",
description="YesCaptcha-compatible captcha solving service for flow2api.",
lifespan=lifespan,
)
app.include_router(router)
@app.get("/")
async def root() -> dict[str, object]:
return {
"service": "captcha-solver",
"version": "3.0.0",
"endpoints": {
"createTask": "/createTask",
"getTaskResult": "/getTaskResult",
"getBalance": "/getBalance",
"health": "/api/v1/health",
},
"supported_task_types": task_manager.supported_types(),
}
|