From 4dd59aa8df30f88d0e44d4c894ca394914289112 Mon Sep 17 00:00:00 2001 From: bshahin Date: Wed, 17 Jun 2026 10:22:53 +0300 Subject: [PATCH 1/2] feat: add CaptchaAI as a provider (classic in.php/res.php adapter) CaptchaAI is 2Captcha-API-compatible via the classic in.php/res.php endpoints (rather than the JSON createTask API). Adds ServiceEnm.CAPTCHAAI, URL routing in CaptchaOptionsSer.urls_set(), and a small classic-API adapter (core/captchaai.py) that maps the library's createTask payloads to classic params and parses the classic responses, returning the usual GetTaskResultResponseSer shape so the public Turnstile / ReCaptcha / ImageCaptcha classes work transparently. Supported task types: Turnstile, reCAPTCHA v2 (incl. Enterprise), reCAPTCHA v3, ImageToText; unsupported types raise a clear ValueError. Adds README usage note and tests/test_captchaai.py (no-network unit tests). Refs #366 --- README.md | 17 +- src/python_rucaptcha/core/base.py | 20 +++ src/python_rucaptcha/core/captchaai.py | 197 ++++++++++++++++++++++++ src/python_rucaptcha/core/enums.py | 1 + src/python_rucaptcha/core/serializer.py | 6 +- tests/test_captchaai.py | 73 +++++++++ 6 files changed, 312 insertions(+), 2 deletions(-) create mode 100644 src/python_rucaptcha/core/captchaai.py create mode 100644 tests/test_captchaai.py diff --git a/README.md b/README.md index 6edea8e2..c1b94833 100644 --- a/README.md +++ b/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? diff --git a/src/python_rucaptcha/core/base.py b/src/python_rucaptcha/core/base.py index af5ff582..a6e95d95 100644 --- a/src/python_rucaptcha/core/base.py +++ b/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() diff --git a/src/python_rucaptcha/core/captchaai.py b/src/python_rucaptcha/core/captchaai.py new file mode 100644 index 00000000..73a77694 --- /dev/null +++ b/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 Exception as error: # noqa: BLE001 + 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 Exception as error: # noqa: BLE001 + 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 Exception as error: # noqa: BLE001 + 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 Exception as error: # noqa: BLE001 + 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") diff --git a/src/python_rucaptcha/core/enums.py b/src/python_rucaptcha/core/enums.py index 26315bff..12243d73 100644 --- a/src/python_rucaptcha/core/enums.py +++ b/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): diff --git a/src/python_rucaptcha/core/serializer.py b/src/python_rucaptcha/core/serializer.py index 1428c16e..e36aea67 100644 --- a/src/python_rucaptcha/core/serializer.py +++ b/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: diff --git a/tests/test_captchaai.py b/tests/test_captchaai.py new file mode 100644 index 00000000..cb4f2464 --- /dev/null +++ b/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" From abddcaa74b7061511d66a883b1c96d1c8e53c53a Mon Sep 17 00:00:00 2001 From: bshahin Date: Fri, 19 Jun 2026 21:47:45 +0300 Subject: [PATCH 2/2] style: narrow broad except clauses to specific exceptions (requests/aiohttp + ValueError) Addresses Codacy broad-exception-caught flags on the CaptchaAI adapter. --- src/python_rucaptcha/core/captchaai.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/python_rucaptcha/core/captchaai.py b/src/python_rucaptcha/core/captchaai.py index 73a77694..d6ee344a 100644 --- a/src/python_rucaptcha/core/captchaai.py +++ b/src/python_rucaptcha/core/captchaai.py @@ -119,7 +119,7 @@ def solve( data = {"key": key, "method": method, "json": 1, **extra} try: created = requests.post(url_request, data=data).json() - except Exception as error: # noqa: BLE001 + except (requests.RequestException, ValueError) as error: return _err(f"ERROR_SUBMIT {error}") if created.get("status") != 1: return _err(created.get("request")) @@ -133,7 +133,7 @@ def solve( params={"key": key, "action": "get", "id": captcha_id, "json": 1}, ).json() logging.info(f"CaptchaAI sync result - {res = }") - except Exception as error: # noqa: BLE001 + except (requests.RequestException, ValueError) as error: return _err(f"ERROR_POLL {error}") if res.get("status") == 1: return GetTaskResultResponseSer( @@ -167,7 +167,7 @@ async def aio_solve( try: async with session.post(url_request, data=data) as resp: created = await resp.json(content_type=None) - except Exception as error: # noqa: BLE001 + except (aiohttp.ClientError, ValueError) as error: return _err(f"ERROR_SUBMIT {error}") if created.get("status") != 1: return _err(created.get("request")) @@ -182,7 +182,7 @@ async def aio_solve( ) as resp: res = await resp.json(content_type=None) logging.info(f"CaptchaAI async result - {res = }") - except Exception as error: # noqa: BLE001 + except (aiohttp.ClientError, ValueError) as error: return _err(f"ERROR_POLL {error}") if res.get("status") == 1: return GetTaskResultResponseSer(