Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions python_anticaptcha/__init__.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,23 @@
"""Python client library for the `Anticaptcha.com <https://anti-captcha.com>`_ API.

Solve ReCAPTCHA v2/v3, hCaptcha, FunCaptcha, GeeTest, image-to-text, and
AntiGate tasks using human workers. Supports both synchronous (``requests``)
and asynchronous (``httpx``) usage.

Quick start::

from python_anticaptcha import AnticaptchaClient, NoCaptchaTaskProxylessTask

with AnticaptchaClient("my-api-key") as client:
task = NoCaptchaTaskProxylessTask(website_url, site_key)
job = client.create_task(task)
job.join()
print(job.get_solution_response())

For async usage, install with ``pip install python-anticaptcha[async]`` and use
:class:`AsyncAnticaptchaClient`.
"""

import contextlib
from importlib.metadata import PackageNotFoundError, version

Expand Down
133 changes: 130 additions & 3 deletions python_anticaptcha/async_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,19 @@


class AsyncJob:
"""An async handle to a submitted captcha-solving task.

Returned by :meth:`AsyncAnticaptchaClient.create_task`. Use :meth:`join`
to wait for completion, then call one of the ``get_*`` methods to
retrieve the solution.

Example::

job = await client.create_task(task)
await job.join()
print(job.get_solution_response()) # for ReCAPTCHA / hCaptcha
"""

client: AsyncAnticaptchaClient
task_id: int
_last_result: dict[str, Any] | None = None
Expand All @@ -31,38 +44,80 @@ async def _update(self) -> None:
self._last_result = await self.client.getTaskResult(self.task_id)

async def check_is_ready(self) -> bool:
"""Poll the API once and return whether the task is complete.

:returns: ``True`` if the solution is ready, ``False`` otherwise.
"""
await self._update()
assert self._last_result is not None
return self._last_result["status"] == "ready"

def get_solution_response(self) -> str: # Recaptcha
def get_solution_response(self) -> str:
"""Return the ``gRecaptchaResponse`` token.

Use after solving ReCAPTCHA v2, ReCAPTCHA v3, or hCaptcha tasks.
Call this only after :meth:`join` has returned.
"""
assert self._last_result is not None
return self._last_result["solution"]["gRecaptchaResponse"]

def get_solution(self) -> dict[str, Any]:
"""Return the full solution dictionary from the API response.

Useful for task types where the solution has multiple fields
(e.g. GeeTest returns ``challenge``, ``validate``, ``seccode``).
Call this only after :meth:`join` has returned.
"""
assert self._last_result is not None
return self._last_result["solution"]

def get_token_response(self) -> str: # Funcaptcha
def get_token_response(self) -> str:
"""Return the ``token`` string from the solution.

Use after solving FunCaptcha tasks.
Call this only after :meth:`join` has returned.
"""
assert self._last_result is not None
return self._last_result["solution"]["token"]

def get_answers(self) -> dict[str, str]:
"""Return the ``answers`` dictionary from the solution.

Use after solving AntiGate tasks.
Call this only after :meth:`join` has returned.
"""
assert self._last_result is not None
return self._last_result["solution"]["answers"]

def get_captcha_text(self) -> str: # Image
def get_captcha_text(self) -> str:
"""Return the recognized text from an image captcha.

Use after solving :class:`ImageToTextTask` tasks.
Call this only after :meth:`join` has returned.
"""
assert self._last_result is not None
return self._last_result["solution"]["text"]

def get_cells_numbers(self) -> list[int]:
"""Return the list of selected cell numbers from a grid captcha.

Call this only after :meth:`join` has returned.
"""
assert self._last_result is not None
return self._last_result["solution"]["cellNumbers"]

async def report_incorrect_image(self) -> bool:
"""Report that an image captcha was solved incorrectly.

:returns: ``True`` if the report was accepted.
"""
return await self.client.reportIncorrectImage(self.task_id)

async def report_incorrect_recaptcha(self) -> bool:
"""Report that a ReCAPTCHA was solved incorrectly.

:returns: ``True`` if the report was accepted.
"""
return await self.client.reportIncorrectRecaptcha(self.task_id)

def __repr__(self) -> str:
Expand Down Expand Up @@ -109,6 +164,28 @@ async def join(


class AsyncAnticaptchaClient:
"""Asynchronous client for the Anticaptcha.com API.

Mirrors :class:`AnticaptchaClient` but all network methods are coroutines.
Requires the ``httpx`` package — install with
``pip install python-anticaptcha[async]``.

Can be used as an async context manager::

async with AsyncAnticaptchaClient("my-api-key") as client:
job = await client.create_task(task)
await job.join()

:param client_key: Your Anticaptcha API key. If omitted, the
``ANTICAPTCHA_API_KEY`` environment variable is used.
:param language_pool: Language pool for workers — ``"en"`` (default)
or ``"rn"`` (Russian).
:param host: API hostname (default: ``"api.anti-captcha.com"``).
:param use_ssl: Use HTTPS (default: ``True``).
:raises ImportError: If ``httpx`` is not installed.
:raises AnticaptchaException: If no API key is provided.
"""

client_key = None
CREATE_TASK_URL = "/createTask"
TASK_RESULT_URL = "/getTaskResult"
Expand Down Expand Up @@ -155,6 +232,10 @@ async def __aexit__(
return False

async def close(self) -> None:
"""Close the underlying HTTP session.

Called automatically when using the client as an async context manager.
"""
await self.session.aclose()

def __repr__(self) -> str:
Expand All @@ -179,6 +260,12 @@ async def _check_response(self, response: dict[str, Any]) -> None:
raise AnticaptchaException(response["errorId"], response["errorCode"], response["errorDescription"])

async def createTask(self, task: BaseTask) -> AsyncJob:
"""Submit a captcha task and return an :class:`AsyncJob` handle.

:param task: A task instance (e.g. :class:`NoCaptchaTaskProxylessTask`).
:returns: An :class:`AsyncJob` that can be polled with :meth:`AsyncJob.join`.
:raises AnticaptchaException: If the API returns an error.
"""
request = {
"clientKey": self.client_key,
"task": task.serialize(),
Expand All @@ -196,12 +283,23 @@ async def createTask(self, task: BaseTask) -> AsyncJob:
return AsyncJob(self, response["taskId"])

async def getTaskResult(self, task_id: int) -> dict[str, Any]:
"""Fetch the current result/status of a task.

:param task_id: The task ID returned when the task was created.
:returns: Raw API response dictionary with ``status`` and ``solution`` keys.
:raises AnticaptchaException: If the API returns an error.
"""
request = {"clientKey": self.client_key, "taskId": task_id}
response = (await self.session.post(urljoin(self.base_url, self.TASK_RESULT_URL), json=request)).json()
await self._check_response(response)
return response

async def getBalance(self) -> float:
"""Return the current account balance in USD.

:returns: Account balance as a float (e.g. ``3.50``).
:raises AnticaptchaException: If the API returns an error.
"""
request = {
"clientKey": self.client_key,
"softId": self.SOFT_ID,
Expand All @@ -211,27 +309,56 @@ async def getBalance(self) -> float:
return response["balance"]

async def getAppStats(self, soft_id: int, mode: str) -> dict[str, Any]:
"""Retrieve application statistics.

:param soft_id: Application ID.
:param mode: Statistics mode (e.g. ``"errors"``, ``"views"``, ``"downloads"``).
:returns: Raw API response dictionary with statistics data.
:raises AnticaptchaException: If the API returns an error.
"""
request = {"clientKey": self.client_key, "softId": soft_id, "mode": mode}
response = (await self.session.post(urljoin(self.base_url, self.APP_STAT_URL), json=request)).json()
await self._check_response(response)
return response

async def reportIncorrectImage(self, task_id: int) -> bool:
"""Report that an image captcha was solved incorrectly.

Use this to get a refund and improve solver accuracy.

:param task_id: The task ID of the incorrectly solved task.
:returns: ``True`` if the report was accepted.
:raises AnticaptchaException: If the API returns an error.
"""
request = {"clientKey": self.client_key, "taskId": task_id}
response = (await self.session.post(urljoin(self.base_url, self.REPORT_IMAGE_URL), json=request)).json()
await self._check_response(response)
return bool(response.get("status", False))

async def reportIncorrectRecaptcha(self, task_id: int) -> bool:
"""Report that a ReCAPTCHA was solved incorrectly.

Use this to get a refund and improve solver accuracy.

:param task_id: The task ID of the incorrectly solved task.
:returns: ``True`` if the report was accepted.
:raises AnticaptchaException: If the API returns an error.
"""
request = {"clientKey": self.client_key, "taskId": task_id}
response = (await self.session.post(urljoin(self.base_url, self.REPORT_RECAPTCHA_URL), json=request)).json()
await self._check_response(response)
return response["status"] == "success"

# Snake_case aliases
#: Alias for :meth:`createTask`.
create_task = createTask
#: Alias for :meth:`getTaskResult`.
get_task_result = getTaskResult
#: Alias for :meth:`getBalance`.
get_balance = getBalance
#: Alias for :meth:`getAppStats`.
get_app_stats = getAppStats
#: Alias for :meth:`reportIncorrectImage`.
report_incorrect_image = reportIncorrectImage
#: Alias for :meth:`reportIncorrectRecaptcha`.
report_incorrect_recaptcha = reportIncorrectRecaptcha
23 changes: 23 additions & 0 deletions python_anticaptcha/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,24 @@


class AnticaptchaException(Exception):
"""Base exception for all Anticaptcha API errors.

Raised when the API returns a non-zero ``errorId``. Inspect
:attr:`error_code` to determine the cause::

try:
job = client.create_task(task)
except AnticaptchaException as e:
if e.error_code == "ERROR_ZERO_BALANCE":
print("Please top up your balance")
else:
raise

:param error_id: Numeric error ID from the API (or a local identifier).
:param error_code: Error code string (e.g. ``"ERROR_ZERO_BALANCE"``).
:param error_description: Human-readable error description.
"""

def __init__(
self,
error_id: int | str | None,
Expand All @@ -16,16 +34,21 @@ def __init__(


AnticatpchaException = AnticaptchaException
"""Backward-compatible alias (legacy misspelling)."""


class InvalidWidthException(AnticaptchaException):
"""Raised when an invalid grid width is specified."""

def __init__(self, width: int) -> None:
self.width = width
msg = f"Invalid width ({self.width}). Can be one of these: 100, 50, 33, 25."
super().__init__("AC-1", 1, msg)


class MissingNameException(AnticaptchaException):
"""Raised when a required ``name`` parameter is missing during serialization."""

def __init__(self, cls: type) -> None:
self.cls = cls
msg = 'Missing name data in {0}. Provide {0}.__init__(name="X") or {0}.serialize(name="X")'.format(
Expand Down
32 changes: 32 additions & 0 deletions python_anticaptcha/proxy.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,21 @@

@dataclass(frozen=True)
class Proxy:
"""Immutable representation of a proxy server.

Use :meth:`parse_url` to build from a URL string, then pass the proxy
parameters to a proxy-enabled task class with :meth:`to_kwargs`::

proxy = Proxy.parse_url("socks5://user:pass@host:1080")
task = NoCaptchaTask(url, key, user_agent=UA, **proxy.to_kwargs())

:param proxy_type: Protocol — ``"http"``, ``"socks4"``, or ``"socks5"``.
:param proxy_address: Hostname or IP address.
:param proxy_port: Port number.
:param proxy_login: Username for authentication (default: ``""``).
:param proxy_password: Password for authentication (default: ``""``).
"""

proxy_type: str
proxy_address: str
proxy_port: int
Expand All @@ -14,6 +29,13 @@ class Proxy:

@classmethod
def parse_url(cls, url: str) -> Proxy:
"""Create a :class:`Proxy` from a URL string.

:param url: Proxy URL, e.g. ``"socks5://user:pass@host:1080"``
or ``"http://host:8080"``.
:returns: A new :class:`Proxy` instance.
:raises ValueError: If the URL is missing a hostname or port.
"""
parsed = urlparse(url)
if not parsed.hostname or not parsed.port:
raise ValueError(f"Invalid proxy URL: {url}")
Expand All @@ -26,6 +48,16 @@ def parse_url(cls, url: str) -> Proxy:
)

def to_kwargs(self) -> dict[str, str | int]:
"""Convert to a keyword-arguments dictionary for task constructors.

The returned dictionary can be unpacked directly into any
proxy-enabled task class::

task = NoCaptchaTask(url, key, user_agent=UA, **proxy.to_kwargs())

:returns: Dictionary with ``proxy_type``, ``proxy_address``,
``proxy_port``, ``proxy_login``, and ``proxy_password`` keys.
"""
return {
"proxy_type": self.proxy_type,
"proxy_address": self.proxy_address,
Expand Down
Loading
Loading