File size: 5,753 Bytes
3a04f21 5124452 3a04f21 5124452 3a04f21 5124452 3a04f21 5124452 | 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 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 | """YesCaptcha / AntiCaptcha compatible HTTP routes."""
from __future__ import annotations
import logging
from fastapi import APIRouter
from ..core.config import config
from ..models.task import (
CreateTaskRequest,
CreateTaskResponse,
GetBalanceRequest,
GetBalanceResponse,
GetTaskResultRequest,
GetTaskResultResponse,
SolutionObject,
)
from ..services.task_manager import TaskStatus, task_manager
log = logging.getLogger(__name__)
router = APIRouter()
_BROWSER_TASK_TYPES = {
"RecaptchaV3TaskProxyless",
"RecaptchaV3TaskProxylessM1",
"RecaptchaV3TaskProxylessM1S7",
"RecaptchaV3TaskProxylessM1S9",
"RecaptchaV3EnterpriseTask",
"RecaptchaV3EnterpriseTaskM1",
"NoCaptchaTaskProxyless",
"RecaptchaV2TaskProxyless",
"RecaptchaV2EnterpriseTaskProxyless",
"HCaptchaTaskProxyless",
"TurnstileTaskProxyless",
"TurnstileTaskProxylessM1",
}
_IMAGE_TASK_TYPES = {
"ImageToTextTask",
"ImageToTextTaskMuggle",
"ImageToTextTaskM1",
}
_CLASSIFICATION_TASK_TYPES = {
"HCaptchaClassification",
"ReCaptchaV2Classification",
"FunCaptchaClassification",
"AwsClassification",
}
def _check_client_key(client_key: str) -> CreateTaskResponse | None:
"""Return an error response if the client key is invalid, else None."""
if config.client_key and client_key != config.client_key:
return CreateTaskResponse(
errorId=1,
errorCode="ERROR_KEY_DOES_NOT_EXIST",
errorDescription="Invalid clientKey",
)
return None
@router.post(
"/createTask",
response_model=CreateTaskResponse,
response_model_exclude_none=True,
)
async def create_task(request: CreateTaskRequest) -> CreateTaskResponse:
err = _check_client_key(request.clientKey)
if err:
return err
supported = task_manager.supported_types()
if request.task.type not in supported:
return CreateTaskResponse(
errorId=1,
errorCode="ERROR_TASK_NOT_SUPPORTED",
errorDescription=f"Task type '{request.task.type}' is not supported. "
f"Supported: {supported}",
)
# Validate required fields for browser-based tasks
if request.task.type in _BROWSER_TASK_TYPES:
if not request.task.websiteURL or not request.task.websiteKey:
return CreateTaskResponse(
errorId=1,
errorCode="ERROR_TASK_PROPERTY_EMPTY",
errorDescription="websiteURL and websiteKey are required",
)
# Validate required fields for ImageToText tasks
if request.task.type in _IMAGE_TASK_TYPES:
if not request.task.body:
return CreateTaskResponse(
errorId=1,
errorCode="ERROR_TASK_PROPERTY_EMPTY",
errorDescription="body (base64 image) is required",
)
# Validate required fields for classification tasks
if request.task.type in _CLASSIFICATION_TASK_TYPES:
has_image = (
request.task.image
or request.task.images
or request.task.body
or request.task.queries
)
if not has_image:
return CreateTaskResponse(
errorId=1,
errorCode="ERROR_TASK_PROPERTY_EMPTY",
errorDescription="image data is required for classification tasks",
)
params = request.task.model_dump(exclude_none=True)
task_id = task_manager.create_task(request.task.type, params)
log.info("Created task %s (type=%s)", task_id, request.task.type)
return CreateTaskResponse(errorId=0, taskId=task_id)
@router.post(
"/getTaskResult",
response_model=GetTaskResultResponse,
response_model_exclude_none=True,
)
async def get_task_result(
request: GetTaskResultRequest,
) -> GetTaskResultResponse:
if config.client_key and request.clientKey != config.client_key:
return GetTaskResultResponse(
errorId=1,
errorCode="ERROR_KEY_DOES_NOT_EXIST",
errorDescription="Invalid clientKey",
)
task = task_manager.get_task(request.taskId)
if task is None:
return GetTaskResultResponse(
errorId=1,
errorCode="ERROR_NO_SUCH_CAPCHA_ID",
errorDescription="Task not found",
)
if task.status == TaskStatus.PROCESSING:
return GetTaskResultResponse(errorId=0, status="processing")
if task.status == TaskStatus.READY:
return GetTaskResultResponse(
errorId=0,
status="ready",
solution=SolutionObject(**(task.solution or {})),
)
return GetTaskResultResponse(
errorId=1,
errorCode=task.error_code or "ERROR_CAPTCHA_UNSOLVABLE",
errorDescription=task.error_description,
)
@router.post(
"/getBalance",
response_model=GetBalanceResponse,
response_model_exclude_none=True,
)
async def get_balance(request: GetBalanceRequest) -> GetBalanceResponse:
if config.client_key and request.clientKey != config.client_key:
return GetBalanceResponse(errorId=1, balance=0)
return GetBalanceResponse(errorId=0, balance=99999.0)
@router.get("/api/v1/health")
async def health() -> dict[str, object]:
return {
"status": "ok",
"supported_task_types": task_manager.supported_types(),
"browser_headless": config.browser_headless,
"browser_proxy_configured": bool(config.browser_proxy_url),
"cloud_model": config.cloud_model,
"local_model": config.local_model,
}
|