Add CaptchaAI as a provider (classic in.php/res.php adapter)

README.md

@@ -5,7 +5,22 @@
 [![Downloads](https://static.pepy.tech/badge/python-rucaptcha/month)](https://pepy.tech/project/python-rucaptcha)
 [![Documentation](https://img.shields.io/badge/docs-Sphinx-green)](https://andreidrang.github.io/python-rucaptcha/)
 
-**Python 3.9+ library to solve CAPTCHAs automatically using RuCaptcha, 2Captcha, or DeathByCaptcha services.**
+**Python 3.9+ library to solve CAPTCHAs automatically using RuCaptcha, 2Captcha, DeathByCaptcha, or CaptchaAI services.**
+
+> **Using CaptchaAI:** pass `service_type=ServiceEnm.CAPTCHAAI` to any captcha class. CaptchaAI is
+> 2Captcha-API-compatible via the classic `in.php`/`res.php` endpoints; supported task types: Turnstile,
+> reCAPTCHA v2 (incl. Enterprise), reCAPTCHA v3, and ImageToText.
+> ```python
+> from python_rucaptcha.turnstile import Turnstile
+> from python_rucaptcha.core.enums import ServiceEnm
+> result = Turnstile(
+>     rucaptcha_key="CAPTCHAAI_KEY",
+>     service_type=ServiceEnm.CAPTCHAAI,
+>     websiteURL="https://example.com",
+>     websiteKey="0x4AAAAAAA...",
+>     userAgent="Mozilla/5.0 ...",
+> ).captcha_handler()
+> ```
 
 ## What is this?
 

src/python_rucaptcha/core/base.py

@@ -82,6 +82,16 @@ def _processing_response(self, **kwargs: dict[str, Any]) -> dict[str, Any]:
         Method processing captcha solving task creation result
         :param kwargs: additional params for Requests library
         """
+        # CaptchaAI speaks the classic in.php/res.php API, not the JSON createTask API.
+        if self.params.service_type == ServiceEnm.CAPTCHAAI.value:
+            from . import captchaai
+
+            return captchaai.solve(
+                create_task_payload=self.create_task_payload,
+                url_request=self.params.url_request,
+                url_response=self.params.url_response,
+                sleep_time=self.params.sleep_time,
+            )
         try:
             response = GetTaskResultResponseSer(
                 **self.session.post(self.params.url_request, json=self.create_task_payload, **kwargs).json()
@@ -126,6 +136,16 @@ async def _aio_processing_response(self) -> dict[str, Any]:
         """
         Method processing async captcha solving task creation result
         """
+        # CaptchaAI speaks the classic in.php/res.php API, not the JSON createTask API.
+        if self.params.service_type == ServiceEnm.CAPTCHAAI.value:
+            from . import captchaai
+
+            return await captchaai.aio_solve(
+                create_task_payload=self.create_task_payload,
+                url_request=self.params.url_request,
+                url_response=self.params.url_response,
+                sleep_time=self.params.sleep_time,
+            )
         try:
             # make async or sync request
             response = await self.__aio_create_task()

src/python_rucaptcha/core/captchaai.py

@@ -0,0 +1,197 @@
+"""
+CaptchaAI provider adapter.
+
+CaptchaAI (https://captchaai.com) is 2Captcha-API-compatible but exposes the
+**classic** ``in.php`` / ``res.php`` form-parameter API rather than the newer
+JSON ``createTask`` / ``getTaskResult`` API the rest of this library speaks.
+
+This module translates the library's internal ``create_task_payload`` (the v2
+JSON task dict) into the classic form parameters CaptchaAI expects, submits the
+task, polls for the result, and returns a response shaped like
+``GetTaskResultResponseSer.to_dict()`` so the public ``Turnstile`` / ``ReCaptcha``
+/ ``ImageCaptcha`` classes work transparently when ``service_type`` is
+``ServiceEnm.CAPTCHAAI``.
+
+Supported task types (the ones CaptchaAI solves through the classic API):
+``TurnstileTaskProxyless``/``TurnstileTask``, ``RecaptchaV2TaskProxyless``/
+``RecaptchaV2Task``, ``RecaptchaV3TaskProxyless``, ``ImageToTextTask``.
+Other task types raise a clear ``ValueError`` so callers fail fast rather than
+silently sending an unsupported method.
+"""
+
+from __future__ import annotations
+
+import time
+import asyncio
+import logging
+from typing import Any
+
+import aiohttp
+import requests
+
+from .config import attempts_generator
+from .serializer import GetTaskResultResponseSer
+
+# v2 task ``type`` -> classic ``method``
+_TURNSTILE = {"TurnstileTaskProxyless", "TurnstileTask"}
+_RECAPTCHA_V2 = {
+    "RecaptchaV2TaskProxyless",
+    "RecaptchaV2Task",
+    "RecaptchaV2EnterpriseTaskProxyless",
+    "RecaptchaV2EnterpriseTask",
+}
+_RECAPTCHA_V3 = {"RecaptchaV3TaskProxyless"}
+_IMAGE = {"ImageToTextTask"}
+
+
+def _err(request: str | None) -> dict[str, Any]:
+    return GetTaskResultResponseSer(
+        status="failed",
+        errorId=12,
+        errorCode=request or "ERROR_UNKNOWN",
+        errorDescription=f"CaptchaAI returned: {request}",
+    ).to_dict()
+
+
+def build_classic_params(task: dict[str, Any]) -> tuple[str, dict[str, Any]]:
+    """Map a v2 task dict to (classic ``method``, classic params)."""
+    ctype = task.get("type")
+    extra: dict[str, Any] = {}
+
+    if ctype in _TURNSTILE:
+        method = "turnstile"
+        extra["sitekey"] = task["websiteKey"]
+        extra["pageurl"] = task["websiteURL"]
+        if task.get("action"):
+            extra["action"] = task["action"]
+        if task.get("data"):
+            extra["data"] = task["data"]
+    elif ctype in _RECAPTCHA_V2:
+        method = "userrecaptcha"
+        extra["googlekey"] = task["websiteKey"]
+        extra["pageurl"] = task["websiteURL"]
+        if "Enterprise" in (ctype or ""):
+            extra["enterprise"] = 1
+    elif ctype in _RECAPTCHA_V3:
+        method = "userrecaptcha"
+        extra["googlekey"] = task["websiteKey"]
+        extra["pageurl"] = task["websiteURL"]
+        extra["version"] = "v3"
+        if task.get("pageAction"):
+            extra["action"] = task["pageAction"]
+        if task.get("minScore"):
+            extra["min_score"] = task["minScore"]
+    elif ctype in _IMAGE:
+        method = "base64"
+        extra["body"] = task["body"]
+    else:
+        raise ValueError(
+            f"CaptchaAI provider does not support task type {ctype!r}. "
+            f"Supported: Turnstile, ReCaptchaV2, ReCaptchaV3, ImageToText."
+        )
+
+    # pass-through proxy / user-agent if present
+    if task.get("userAgent"):
+        extra["userAgent"] = task["userAgent"]
+    return method, extra
+
+
+def _solution(method: str, token: str) -> dict[str, str]:
+    if method == "base64":
+        return {"text": token}
+    return {"token": token, "gRecaptchaResponse": token}
+
+
+def solve(
+    create_task_payload: dict[str, Any],
+    url_request: str,
+    url_response: str,
+    sleep_time: int,
+) -> dict[str, Any]:
+    """Synchronous classic in.php/res.php solve."""
+    key = create_task_payload["clientKey"]
+    task = create_task_payload["task"]
+    try:
+        method, extra = build_classic_params(task)
+    except (KeyError, ValueError) as error:
+        return _err(str(error))
+
+    data = {"key": key, "method": method, "json": 1, **extra}
+    try:
+        created = requests.post(url_request, data=data).json()
+    except (requests.RequestException, ValueError) as error:
+        return _err(f"ERROR_SUBMIT {error}")
+    if created.get("status") != 1:
+        return _err(created.get("request"))
+
+    captcha_id = created["request"]
+    time.sleep(sleep_time)
+    for _ in attempts_generator():
+        try:
+            res = requests.get(
+                url_response,
+                params={"key": key, "action": "get", "id": captcha_id, "json": 1},
+            ).json()
+            logging.info(f"CaptchaAI sync result - {res = }")
+        except (requests.RequestException, ValueError) as error:
+            return _err(f"ERROR_POLL {error}")
+        if res.get("status") == 1:
+            return GetTaskResultResponseSer(
+                status="ready",
+                solution=_solution(method, res["request"]),
+                taskId=captcha_id if str(captcha_id).isdigit() else None,
+            ).to_dict()
+        if res.get("request") == "CAPCHA_NOT_READY":
+            time.sleep(sleep_time)
+            continue
+        return _err(res.get("request"))
+    return _err("ERROR_TIMEOUT")
+
+
+async def aio_solve(
+    create_task_payload: dict[str, Any],
+    url_request: str,
+    url_response: str,
+    sleep_time: int,
+) -> dict[str, Any]:
+    """Asynchronous classic in.php/res.php solve."""
+    key = create_task_payload["clientKey"]
+    task = create_task_payload["task"]
+    try:
+        method, extra = build_classic_params(task)
+    except (KeyError, ValueError) as error:
+        return _err(str(error))
+
+    data = {"key": key, "method": method, "json": 1, **extra}
+    async with aiohttp.ClientSession() as session:
+        try:
+            async with session.post(url_request, data=data) as resp:
+                created = await resp.json(content_type=None)
+        except (aiohttp.ClientError, ValueError) as error:
+            return _err(f"ERROR_SUBMIT {error}")
+        if created.get("status") != 1:
+            return _err(created.get("request"))
+
+        captcha_id = created["request"]
+        await asyncio.sleep(sleep_time)
+        for _ in attempts_generator():
+            try:
+                async with session.get(
+                    url_response,
+                    params={"key": key, "action": "get", "id": captcha_id, "json": 1},
+                ) as resp:
+                    res = await resp.json(content_type=None)
+                    logging.info(f"CaptchaAI async result - {res = }")
+            except (aiohttp.ClientError, ValueError) as error:
+                return _err(f"ERROR_POLL {error}")
+            if res.get("status") == 1:
+                return GetTaskResultResponseSer(
+                    status="ready",
+                    solution=_solution(method, res["request"]),
+                    taskId=captcha_id if str(captcha_id).isdigit() else None,
+                ).to_dict()
+            if res.get("request") == "CAPCHA_NOT_READY":
+                await asyncio.sleep(sleep_time)
+                continue
+            return _err(res.get("request"))
+    return _err("ERROR_TIMEOUT")

src/python_rucaptcha/core/enums.py

@@ -24,6 +24,7 @@ class ServiceEnm(str, MyEnum):
     TWOCAPTCHA = "2captcha"
     RUCAPTCHA = "rucaptcha"
     DEATHBYCAPTCHA = "deathbycaptcha"
+    CAPTCHAAI = "captchaai"
 
 
 class SaveFormatsEnm(str, MyEnum):

src/python_rucaptcha/core/serializer.py

@@ -68,7 +68,11 @@ def urls_set(self):
         if isinstance(self.service_type, enums.ServiceEnm):
             self.service_type = self.service_type.value
 
-        if self.service_type == enums.ServiceEnm.DEATHBYCAPTCHA:
+        if self.service_type == enums.ServiceEnm.CAPTCHAAI:
+            # CaptchaAI exposes the classic 2captcha in.php/res.php API.
+            self.url_request = "https://ocr.captchaai.com/in.php"
+            self.url_response = "https://ocr.captchaai.com/res.php"
+        elif self.service_type == enums.ServiceEnm.DEATHBYCAPTCHA:
             self.url_request = f"http://api.{self.service_type}.com/2captcha/in.php"
             self.url_response = f"http://api.{self.service_type}.com/2captcha/res.php"
         else:

tests/test_captchaai.py

@@ -0,0 +1,73 @@
+import pytest
+
+from python_rucaptcha.core import captchaai
+from python_rucaptcha.core.enums import ServiceEnm
+from python_rucaptcha.core.serializer import CaptchaOptionsSer
+
+
+class TestCaptchaAIService:
+    """Unit tests for the CaptchaAI classic-API provider adapter (no network)."""
+
+    def test_service_enum(self):
+        assert ServiceEnm.CAPTCHAAI.value == "captchaai"
+        assert "captchaai" in ServiceEnm.list_values()
+
+    def test_urls_set(self):
+        opts = CaptchaOptionsSer(service_type=ServiceEnm.CAPTCHAAI)
+        opts.urls_set()
+        assert opts.url_request == "https://ocr.captchaai.com/in.php"
+        assert opts.url_response == "https://ocr.captchaai.com/res.php"
+
+    def test_map_turnstile(self):
+        method, extra = captchaai.build_classic_params(
+            {"type": "TurnstileTaskProxyless", "websiteKey": "0xAAA", "websiteURL": "https://x.com"}
+        )
+        assert method == "turnstile"
+        assert extra == {"sitekey": "0xAAA", "pageurl": "https://x.com"}
+
+    def test_map_recaptcha_v2(self):
+        method, extra = captchaai.build_classic_params(
+            {"type": "RecaptchaV2TaskProxyless", "websiteKey": "k", "websiteURL": "https://x"}
+        )
+        assert method == "userrecaptcha"
+        assert extra["googlekey"] == "k"
+        assert extra["pageurl"] == "https://x"
+
+    def test_map_recaptcha_v2_enterprise(self):
+        method, extra = captchaai.build_classic_params(
+            {"type": "RecaptchaV2EnterpriseTaskProxyless", "websiteKey": "k", "websiteURL": "https://x"}
+        )
+        assert method == "userrecaptcha"
+        assert extra["enterprise"] == 1
+
+    def test_map_recaptcha_v3(self):
+        method, extra = captchaai.build_classic_params(
+            {
+                "type": "RecaptchaV3TaskProxyless",
+                "websiteKey": "k",
+                "websiteURL": "https://x",
+                "pageAction": "login",
+                "minScore": 0.7,
+            }
+        )
+        assert method == "userrecaptcha"
+        assert extra["version"] == "v3"
+        assert extra["action"] == "login"
+        assert extra["min_score"] == 0.7
+
+    def test_map_image(self):
+        method, extra = captchaai.build_classic_params({"type": "ImageToTextTask", "body": "B64"})
+        assert method == "base64"
+        assert extra == {"body": "B64"}
+
+    def test_unsupported_type_raises(self):
+        with pytest.raises(ValueError):
+            captchaai.build_classic_params(
+                {"type": "HCaptchaTaskProxyless", "websiteKey": "k", "websiteURL": "u"}
+            )
+
+    def test_solution_shapes(self):
+        assert captchaai._solution("base64", "abc") == {"text": "abc"}
+        token = captchaai._solution("turnstile", "tok")
+        assert token["token"] == "tok"
+        assert token["gRecaptchaResponse"] == "tok"