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
38 changes: 21 additions & 17 deletions README.rst
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ python-anticaptcha
.. introduction-start

Client library for solve captchas with `Anticaptcha.com support`_.
The library supports both Python 2.7 and Python 3.
The library requires Python >= 3.9.

The library is cyclically and automatically tested for proper operation. We are constantly making the best efforts for its effective operation.

Expand Down Expand Up @@ -67,9 +67,9 @@ Example snippet for Recaptcha:

client = AnticaptchaClient(api_key)
task = NoCaptchaTaskProxylessTask(url, site_key)
job = client.createTask(task)
job = client.create_task(task)
job.join()
print job.get_solution_response()
print(job.get_solution_response())

The full integration example is available in file ``examples/recaptcha.py``.

Expand All @@ -89,9 +89,9 @@ measures for automated training and analysis. For provide that pass

client = AnticaptchaClient(api_key)
task = NoCaptchaTaskProxylessTask(url, site_key, is_invisible=True)
job = client.createTask(task)
job = client.create_task(task)
job.join()
print job.get_solution_response()
print(job.get_solution_response())


Solve text captcha
Expand All @@ -107,9 +107,9 @@ Example snippet for text captcha:
captcha_fp = open('examples/captcha_ms.jpeg', 'rb')
client = AnticaptchaClient(api_key)
task = ImageToTextTask(captcha_fp)
job = client.createTask(task)
job = client.create_task(task)
job.join()
print job.get_captcha_text()
print(job.get_captcha_text())

Solve funcaptcha
################
Expand All @@ -125,13 +125,13 @@ Example snippet for funcaptcha:
api_key = '174faff8fbc769e94a5862391ecfd010'
site_key = 'DE0B0BB7-1EE4-4D70-1853-31B835D4506B' # grab from site
url = 'https://www.google.com/recaptcha/api2/demo'
proxy = Proxy.parse_url("socks5://login:password@123.123.123.123")
proxy = Proxy.parse_url("socks5://login:password@123.123.123.123:1080")

client = AnticaptchaClient(api_key)
task = FunCaptchaTask(url, site_key, proxy=proxy, user_agent=user_agent)
job = client.createTask(task)
task = FunCaptchaTask(url, site_key, user_agent=UA, **proxy.to_kwargs())
job = client.create_task(task)
job.join()
print job.get_token_response()
print(job.get_token_response())

Report incorrect image
######################
Expand All @@ -146,10 +146,10 @@ Example snippet for reporting an incorrect image task:
captcha_fp = open('examples/captcha_ms.jpeg', 'rb')
client = AnticaptchaClient(api_key)
task = ImageToTextTask(captcha_fp)
job = client.createTask(task)
job = client.create_task(task)
job.join()
print job.get_captcha_text()
job.report_incorrect()
print(job.get_captcha_text())
job.report_incorrect_image()

Setup proxy
###########
Expand Down Expand Up @@ -181,20 +181,24 @@ We recommend entering IP-based access control for incoming addresses to proxy. I
Error handling
##############

In the event of an application error, the AnticaptchaException exception is thrown. To handle the exception, do the following:
In the event of an application error, the ``AnticaptchaException`` exception is thrown. To handle the exception, do the following:

.. code:: python

from python_anticaptcha import AnticatpchaException, ImageToTextTask
from python_anticaptcha import AnticaptchaException, ImageToTextTask

try:
# any actions
except AnticatpchaException as e:
except AnticaptchaException as e:
if e.error_code == 'ERROR_ZERO_BALANCE':
notify_about_no_funds(e.error_id, e.error_code, e.error_description)
else:
raise

.. note::

The legacy misspelled ``AnticatpchaException`` alias is still available for backward compatibility.

.. usage-end

Versioning
Expand Down
5 changes: 5 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,3 +1,8 @@
[project]
name = "python-anticaptcha"
requires-python = ">=3.9"
dynamic = ["version", "description", "readme", "dependencies", "optional-dependencies"]

[tool.pytest.ini_options]
testpaths = ["tests"]
markers = [
Expand Down
27 changes: 26 additions & 1 deletion python_anticaptcha/__init__.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from importlib.metadata import version, PackageNotFoundError

from .base import AnticaptchaClient
from .base import AnticaptchaClient, Job
from .proxy import Proxy
from .tasks import (
NoCaptchaTaskProxylessTask,
RecaptchaV2TaskProxyless,
Expand Down Expand Up @@ -28,3 +29,27 @@
except PackageNotFoundError:
# package is not installed
pass

__all__ = [
"AnticaptchaClient",
"Job",
"Proxy",
"NoCaptchaTaskProxylessTask",
"RecaptchaV2TaskProxyless",
"NoCaptchaTask",
"RecaptchaV2Task",
"FunCaptchaProxylessTask",
"FunCaptchaTask",
"ImageToTextTask",
"RecaptchaV3TaskProxyless",
"HCaptchaTaskProxyless",
"HCaptchaTask",
"RecaptchaV2EnterpriseTaskProxyless",
"RecaptchaV2EnterpriseTask",
"GeeTestTaskProxyless",
"GeeTestTask",
"AntiGateTaskProxyless",
"AntiGateTask",
"AnticaptchaException",
"AnticatpchaException",
]
60 changes: 36 additions & 24 deletions python_anticaptcha/base.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,10 @@
from __future__ import annotations

import requests
import time
import json
import warnings
from typing import Any

from urllib.parse import urljoin
from .exceptions import AnticaptchaException
Expand All @@ -15,49 +18,49 @@ class Job:
task_id = None
_last_result = None

def __init__(self, client, task_id):
def __init__(self, client: AnticaptchaClient, task_id: int) -> None:
self.client = client
self.task_id = task_id

def _update(self):
def _update(self) -> None:
self._last_result = self.client.getTaskResult(self.task_id)

def check_is_ready(self):
def check_is_ready(self) -> bool:
self._update()
return self._last_result["status"] == "ready"

def get_solution_response(self): # Recaptcha
def get_solution_response(self) -> str: # Recaptcha
return self._last_result["solution"]["gRecaptchaResponse"]

def get_solution(self):
def get_solution(self) -> dict[str, Any]:
return self._last_result["solution"]

def get_token_response(self): # Funcaptcha
def get_token_response(self) -> str: # Funcaptcha
return self._last_result["solution"]["token"]

def get_answers(self):
def get_answers(self) -> dict[str, str]:
return self._last_result["solution"]["answers"]

def get_captcha_text(self): # Image
def get_captcha_text(self) -> str: # Image
return self._last_result["solution"]["text"]

def get_cells_numbers(self):
def get_cells_numbers(self) -> list[int]:
return self._last_result["solution"]["cellNumbers"]

def report_incorrect(self):
def report_incorrect(self) -> bool:
warnings.warn(
"report_incorrect is deprecated, use report_incorrect_image instead",
DeprecationWarning,
)
return self.client.reportIncorrectImage()

def report_incorrect_image(self):
def report_incorrect_image(self) -> bool:
return self.client.reportIncorrectImage(self.task_id)

def report_incorrect_recaptcha(self):
def report_incorrect_recaptcha(self) -> bool:
return self.client.reportIncorrectRecaptcha(self.task_id)

def join(self, maximum_time=None):
def join(self, maximum_time: int | None = None) -> None:
elapsed_time = 0
maximum_time = maximum_time or MAXIMUM_JOIN_TIME
while not self.check_is_ready():
Expand Down Expand Up @@ -86,8 +89,8 @@ class AnticaptchaClient:
response_timeout = 5

def __init__(
self, client_key, language_pool="en", host="api.anti-captcha.com", use_ssl=True
):
self, client_key: str, language_pool: str = "en", host: str = "api.anti-captcha.com", use_ssl: bool = True,
) -> None:
self.client_key = client_key
self.language_pool = language_pool
self.base_url = "{proto}://{host}/".format(
Expand All @@ -96,14 +99,14 @@ def __init__(
self.session = requests.Session()

@property
def client_ip(self):
def client_ip(self) -> str:
if not hasattr(self, "_client_ip"):
self._client_ip = self.session.get(
"https://api.myip.com", timeout=self.response_timeout
).json()["ip"]
return self._client_ip

def _check_response(self, response):
def _check_response(self, response: dict[str, Any]) -> None:
if response.get("errorId", False) == 11:
response[
"errorDescription"
Expand All @@ -115,7 +118,7 @@ def _check_response(self, response):
response["errorId"], response["errorCode"], response["errorDescription"]
)

def createTask(self, task):
def createTask(self, task: Any) -> Job:
request = {
"clientKey": self.client_key,
"task": task.serialize(),
Expand All @@ -130,7 +133,7 @@ def createTask(self, task):
self._check_response(response)
return Job(self, response["taskId"])

def createTaskSmee(self, task, timeout=MAXIMUM_JOIN_TIME):
def createTaskSmee(self, task: Any, timeout: int = MAXIMUM_JOIN_TIME) -> Job:
"""
Beta method to stream response from smee.io
"""
Expand Down Expand Up @@ -172,15 +175,15 @@ def createTaskSmee(self, task, timeout=MAXIMUM_JOIN_TIME):
job._last_result = payload["body"]
return job

def getTaskResult(self, task_id):
def getTaskResult(self, task_id: int) -> dict[str, Any]:
request = {"clientKey": self.client_key, "taskId": task_id}
response = self.session.post(
urljoin(self.base_url, self.TASK_RESULT_URL), json=request
).json()
self._check_response(response)
return response

def getBalance(self):
def getBalance(self) -> float:
request = {
"clientKey": self.client_key,
"softId": self.SOFT_ID,
Expand All @@ -191,26 +194,35 @@ def getBalance(self):
self._check_response(response)
return response["balance"]

def getAppStats(self, soft_id, mode):
def getAppStats(self, soft_id: int, mode: str) -> dict[str, Any]:
request = {"clientKey": self.client_key, "softId": soft_id, "mode": mode}
response = self.session.post(
urljoin(self.base_url, self.APP_STAT_URL), json=request
).json()
self._check_response(response)
return response

def reportIncorrectImage(self, task_id):
def reportIncorrectImage(self, task_id: int) -> bool:
request = {"clientKey": self.client_key, "taskId": task_id}
response = self.session.post(
urljoin(self.base_url, self.REPORT_IMAGE_URL), json=request
).json()
self._check_response(response)
return response.get("status", False) != False

def reportIncorrectRecaptcha(self, task_id):
def reportIncorrectRecaptcha(self, task_id: int) -> bool:
request = {"clientKey": self.client_key, "taskId": task_id}
response = self.session.post(
urljoin(self.base_url, self.REPORT_RECAPTCHA_URL), json=request
).json()
self._check_response(response)
return response["status"] == "success"

# Snake_case aliases
create_task = createTask
create_task_smee = createTaskSmee
get_task_result = getTaskResult
get_balance = getBalance
get_app_stats = getAppStats
report_incorrect_image = reportIncorrectImage
report_incorrect_recaptcha = reportIncorrectRecaptcha
5 changes: 4 additions & 1 deletion python_anticaptcha/exceptions.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
from __future__ import annotations


class AnticaptchaException(Exception):
def __init__(self, error_id, error_code, error_description, *args):
def __init__(self, error_id: int | str | None, error_code: int | str, error_description: str, *args: object) -> None:
super().__init__(
"[{}:{}]{}".format(error_code, error_id, error_description)
)
Expand Down
35 changes: 35 additions & 0 deletions python_anticaptcha/proxy.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
from __future__ import annotations

from dataclasses import dataclass
from urllib.parse import urlparse


@dataclass(frozen=True)
class Proxy:
proxy_type: str
proxy_address: str
proxy_port: int
proxy_login: str = ""
proxy_password: str = ""

@classmethod
def parse_url(cls, url: str) -> Proxy:
parsed = urlparse(url)
if not parsed.hostname or not parsed.port:
raise ValueError(f"Invalid proxy URL: {url}")
return cls(
proxy_type=parsed.scheme,
proxy_address=parsed.hostname,
proxy_port=parsed.port,
proxy_login=parsed.username or "",
proxy_password=parsed.password or "",
)

def to_kwargs(self) -> dict[str, str | int]:
return {
"proxy_type": self.proxy_type,
"proxy_address": self.proxy_address,
"proxy_port": self.proxy_port,
"proxy_login": self.proxy_login,
"proxy_password": self.proxy_password,
}
Empty file added python_anticaptcha/py.typed
Empty file.
Loading
Loading