Rooba
/
python3-anticaptcha
mirror of https://github.com/AndreiDrang/python3-anticaptcha.git
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

refactoring

This commit is contained in:
Andrei 2024-12-16 23:39:20 +03:00
parent 9cb8a1e954
commit dd8ce08b0a
5 changed files with 26 additions and 614 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
docs/modules/image/example.rst
View File

@ -1,12 +1,12 @@
ImageToTextCaptcha
ImageToText
==================
To import this module:
.. code-block:: python
from python3_anticaptcha.image_captcha import ImageToTextCaptcha
from python3_anticaptcha.image_captcha import ImageToText
.. autoclass:: python3_anticaptcha.image_captcha.ImageToTextCaptcha
.. autoclass:: python3_anticaptcha.image_captcha.ImageToText
:members:

166
src/python3_anticaptcha/NoCaptchaTask.py
View File

@ -1,166 +0,0 @@
import time
import asyncio
import aiohttp
import requests
from python3_anticaptcha import app_key, create_task_url, get_sync_result, get_async_result
class NoCaptchaTask:
def __init__(self, anticaptcha_key: str, sleep_time: int = 10, callbackUrl: str = None, **kwargs):
"""
Модуль отвечает за решение NoCaptcha.
:param anticaptcha_key: ключ от АнтиКапчи
:param sleep_time: Время ожидания решения
:param callbackUrl: URL для решения капчи с ответом через callback
:param kwargs: Параметры для подключения к прокси. Подробнее в официальной документации или примерe - anticaptcha_examples/anticaptcha_nocaptcha_example.py
"""
if sleep_time < 10:
raise ValueError(f"Param `sleep_time` must be greater than 10. U set - {sleep_time}")
self.sleep_time = sleep_time
# Пайлоад для создания задачи
self.task_payload = {
"clientKey": anticaptcha_key,
"task": {"type": "NoCaptchaTask"},
"softId": app_key,
}
# задаём callbackUrl если передан
if callbackUrl:
self.task_payload.update({"callbackUrl": callbackUrl})
# пайлоад для получения ответа сервиса
self.result_payload = {"clientKey": anticaptcha_key}
# заполнить пайлоад остальными аргументами
if kwargs:
for key in kwargs:
self.task_payload["task"].update({key: kwargs[key]})
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
if exc_type:
return False
return True
def captcha_handler(self, websiteURL: str, websiteKey: str, recaptchaDataSValue: str = "", **kwargs) -> dict:
"""
Метод получает ссылку на страницу, где расположена капча, и ключ капчи
:param: websiteURL: Ссылка на страницу с капчёй
:param: websiteKey: Ключ капчи(как его получить - описано в документаии на сайте антикапчи)
:param recaptchaDataSValue: Некоторые реализации виджета рекапчи могут содержать
дополнительный параметр "data-s" в div'е рекапчи,
который является одноразовым токеном и
должен собираться каждый раз при решении рекапчи.
:param kwargs: Дополнительные параметры для `requests.post(....)`.
return: Возвращает ответ сервера в виде JSON(ответ так же можно глянуть в документации антикапчи)
"""
self.task_payload["task"].update(
{
"websiteURL": websiteURL,
"websiteKey": websiteKey,
"recaptchaDataSValue": recaptchaDataSValue,
}
)
# отправляем реквест, в ответ получаем JSON содержащий номер решаемой капчи
captcha_id = requests.post(create_task_url, json=self.task_payload, verify=False, **kwargs).json()
# Проверка статуса создания задачи, если создано без ошибок - извлекаем ID задачи, иначе возвращаем ответ сервера
if captcha_id["errorId"] == 0:
captcha_id = captcha_id["taskId"]
self.result_payload.update({"taskId": captcha_id})
else:
return captcha_id
# если передан параметр `callbackUrl` - не ждём решения капчи а возвращаем незаполненный ответ
if self.task_payload.get("callbackUrl"):
return self.result_payload
else:
# Ждем решения капчи
time.sleep(self.sleep_time)
return get_sync_result(result_payload=self.result_payload, sleep_time=self.sleep_time)
class aioNoCaptchaTask:
def __init__(self, anticaptcha_key: str, sleep_time: str = 10, callbackUrl: str = None, **kwargs):
"""
Модуль отвечает за решение NoCaptcha.
:param anticaptcha_key: ключ от АнтиКапчи
:param sleep_time: Время ожидания решения
:param callbackUrl: URL для решения капчи с ответом через callback
:param kwargs: Параметры для подключения к прокси. Подробнее в официальной документации или примерe - anticaptcha_examples/anticaptcha_nocaptcha_example.py
"""
if sleep_time < 10:
raise ValueError(f"Param `sleep_time` must be greater than 10. U set - {sleep_time}")
self.sleep_time = sleep_time
# Пайлоад для создания задачи
self.task_payload = {
"clientKey": anticaptcha_key,
"task": {"type": "NoCaptchaTask"},
"softId": app_key,
}
# задаём callbackUrl если передан
if callbackUrl:
self.task_payload.update({"callbackUrl": callbackUrl})
# пайлоад для получения ответа сервиса
self.result_payload = {"clientKey": anticaptcha_key}
# заполнить пайлоад остальными аргументами
if kwargs:
for key in kwargs:
self.task_payload["task"].update({key: kwargs[key]})
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
if exc_type:
return False
return True
# Работа с капчей
async def captcha_handler(self, websiteURL: str, websiteKey: str, recaptchaDataSValue: str = "") -> dict:
"""
Метод получает ссылку на страницу, где расположена капча, и ключ капчи
:param: websiteURL: Ссылка на страницу с капчёй.
:param: websiteKey: Ключ капчи(как его получить - описано в документаии на сайте антикапчи).
:param recaptchaDataSValue: Некоторые реализации виджета рекапчи могут содержать
дополнительный параметр "data-s" в div'е рекапчи,
который является одноразовым токеном и
должен собираться каждый раз при решении рекапчи.
return: Возвращает ответ сервера в виде JSON(ответ так же можно глянуть в документации антикапчи)
"""
self.task_payload["task"].update(
{
"websiteURL": websiteURL,
"websiteKey": websiteKey,
"recaptchaDataSValue": recaptchaDataSValue,
}
)
# отправляем реквест, в ответ получаем JSON содержащий номер решаемой капчи
async with aiohttp.ClientSession() as session:
async with session.post(create_task_url, json=self.task_payload) as resp:
captcha_id = await resp.json()
# Проверка статуса создания задачи, если создано без ошибок - извлекаем ID задачи, иначе возвращаем ответ сервера
if captcha_id["errorId"] == 0:
captcha_id = captcha_id["taskId"]
self.result_payload.update({"taskId": captcha_id})
else:
return captcha_id
# если передан параметр `callbackUrl` - не ждём решения капчи а возвращаем незаполненный ответ
if self.task_payload.get("callbackUrl"):
return self.result_payload
else:
# Ждем решения капчи
await asyncio.sleep(self.sleep_time)
return await get_async_result(result_payload=self.result_payload, sleep_time=self.sleep_time)

422
src/python3_anticaptcha/SquareNetTextTask.py
View File

@ -1,422 +0,0 @@
import os
import time
import base64
import asyncio
import hashlib
import aiohttp
import requests
from requests.adapters import HTTPAdapter
from python3_anticaptcha import ReadError, ParamError, app_key, create_task_url, get_sync_result, get_async_result
class SquareNetTextTask:
"""
Данный метод подходит для решения капчи-изображение.
Подробней информацию смотрите в методе 'captcha_handler' и '__init__'
"""
def __init__(
self,
anticaptcha_key: str,
sleep_time: int = 5,
save_format: str = "temp",
callbackUrl: str = None,
):
"""
Инициализация нужных переменных, создание папки для изображений и кэша
После завершения работы - удаляются временные фалйы и папки
:param anticaptcha_key: АПИ ключ капчи из кабинета пользователя
:param sleep_time: Вермя ожидания решения капчи
:param save_format: Формат в котором будет сохраняться изображение, либо как временный файл - 'temp',
либо как обычное изображение в папку созданную библиотекой - 'const'.
:param callbackUrl: URL для решения капчи с ответом через callback
"""
if sleep_time < 5:
raise ValueError(f"Param `sleep_time` must be greater than 5. U set - {sleep_time}")
self.sleep_time = sleep_time
# проверяем переданный параметр способа сохранения капчи
if save_format in ["const", "temp"]:
self.save_format = save_format
else:
raise ValueError(
"\nПередан неверный формат сохранения файла изображения. "
f"\n\tВозможные варинты: `temp` и `const`. Вы передали - `{save_format}`"
"\nWrong `save_format` parameter. Valid params: `const` or `temp`."
f"\n\tYour param - `{save_format}`"
)
# Пайлоад для создания задачи
self.task_payload = {
"clientKey": anticaptcha_key,
"task": {"type": "SquareNetTask"},
"softId": app_key,
}
# задаём callbackUrl если передан
if callbackUrl:
self.task_payload.update({"callbackUrl": callbackUrl})
# создаём сессию
self.session = requests.Session()
# выставляем кол-во попыток подключения к серверу при ошибке
self.session.mount("http://", HTTPAdapter(max_retries=5))
self.session.mount("https://", HTTPAdapter(max_retries=5))
self.session.verify = False
# отправляем запрос на результат решения капчи, если ещё капча не решена - ожидаем 5 сек
# если всё ок - идём дальше
self.result_payload = {"clientKey": anticaptcha_key}
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
if exc_type:
return False
return True
def __image_temp_saver(self, content: bytes):
"""
Метод сохраняет файл изображения как временный и отправляет его сразу на сервер для расшифровки.
:return: Возвращает ID капчи
"""
# Создаём пайлоад, вводим ключ от сайта, выбираем метод ПОСТ и ждём ответа в JSON-формате
self.task_payload["task"].update({"body": base64.b64encode(content).decode("utf-8")})
# Отправляем на рукапча изображение капчи и другие парметры,
# в результате получаем JSON ответ с номером решаемой капчи и получая ответ - извлекаем номер
captcha_id = self.session.post(create_task_url, json=self.task_payload).json()
return captcha_id
def __image_const_saver(self, content: bytes):
"""
Метод создаёт папку и сохраняет в неё изображение, затем передаёт его на расшифровку и удалет файл.
:return: Возвращает ID капчи
"""
img_path = "PythonAntiCaptchaImages"
if not os.path.exists(img_path):
os.mkdir(img_path)
# Высчитываем хэш изображения, для того что бы сохранить его под уникальным именем
image_hash = hashlib.sha224(content).hexdigest()
with open(os.path.join(img_path, "im-{0}.png".format(image_hash)), "wb") as out_image:
out_image.write(content)
with open(os.path.join(img_path, "im-{0}.png".format(image_hash)), "rb") as captcha_image:
# Добавляем в пайлоад картинку и отправляем
self.task_payload["task"].update({"body": base64.b64encode(captcha_image.read()).decode("utf-8")})
# Отправляем на антикапча изображение капчи и другие парметры,
# в результате получаем JSON ответ содержащий номер решаемой капчи
captcha_id = self.session.post(create_task_url, json=self.task_payload).json()
# удаляем файл капчи
os.remove(os.path.join(img_path, "im-{0}.png".format(image_hash)))
return captcha_id
def __read_captcha_image_file(self, content: bytes, content_type: str = "file"):
"""
Функция отвечает за чтение уже сохранённого файла или файла в уодировке base64
:param content: Параметр строка-путь указывающий на изображение капчи для отправки её на сервер
:param content_type: Тип переданной переменной. Возможны варианты: `file` и `base64`
:return: Возвращает ID капчи
"""
try:
if content_type == "file":
with open(content, "rb") as captcha_image:
# Добавляем в пайлоад картинку и отправляем
self.task_payload["task"].update({"body": base64.b64encode(captcha_image.read()).decode("utf-8")})
elif content_type == "base64":
self.task_payload["task"].update({"body": content})
else:
raise ValueError(
"\nПередан неверный формат файла."
f"\n\tВозможные варинты: `file` и `base64`. Вы передали - `{content_type}`"
f"\nWrong `content_type` parameter. Valid params: `file` or `base64`."
f"\n\tYour param - `{content_type}`"
)
# Отправляем на антикапча изображение капчи и другие парметры,
# в результате получаем JSON ответ содержащий номер решаемой капчи
captcha_id = self.session.post(create_task_url, json=self.task_payload).json()
except (IOError, FileNotFoundError) as err:
raise ReadError(err)
return captcha_id
# Работа с капчёй
def captcha_handler(
self,
objectName: str,
rowsCount: int,
columnsCount: int,
image_link: str = None,
image_file: str = None,
image_base64: str = None,
**kwargs,
) -> dict:
"""
Этот тип задачи берет вашу картинку, добавляет на нее сетку нужного размера и отдает работнику с требованием выбрать объекты нужного типа.
:param objectName: Имя объекта.
:param rowsCount: Кол-во строк. Min 2, max 5
:param columnsCount: Кол-во колонок. Min 2, max 5
:param image_link: Ссылка на изображение
:param image_file: Необязательный параметр, служит для открытия уже скачанных файлов изображений.
:param image_base64: Загрузка изображения в кодировке base64
:param kwargs: Дополнительные параметры для `requests.get(....)` скачивающего изображение, если передана ссылка.
:return: Возвращает весь ответ сервера JSON-строкой.
"""
# проверка параметров сетки
if not 2 <= rowsCount <= 5 or not 2 <= columnsCount <= 5:
raise ValueError(
"Параметр `rowsCount` и `columnsCount` должен быть больше 2 и меньше 5."
f"Вы передали - `rowsCount`:{rowsCount}, `columnsCount`:{columnsCount}"
)
# обновляем task_payload новыми данными для сетки разметки и названия предмета
self.task_payload["task"].update({"objectName": objectName})
self.task_payload["task"].update({"rowsCount": rowsCount})
self.task_payload["task"].update({"columnsCount": columnsCount})
# проводим действия над файлом(декодируем и передаём на сервер)
if image_file:
captcha_id = self.__read_captcha_image_file(image_file, content_type="file")
# проводим действия над файлом уже закодированном в base64(передаём на сервер)
elif image_base64:
captcha_id = self.__read_captcha_image_file(image_base64, content_type="base64")
# проводим действия над ссылкой на файл(скачиваем, сохраняем и передаём на сервер)
elif image_link:
content = self.session.get(image_link, **kwargs).content
# согласно значения переданного параметра выбираем функцию для сохранения изображения
if self.save_format == "const":
captcha_id = self.__image_const_saver(content)
elif self.save_format == "temp":
captcha_id = self.__image_temp_saver(content)
else:
raise ParamError(
additional_info="""You must set one of param - `image_file/image_base64/image_link`.\n
Вы должны задать один из параметров - `image_file/image_base64/image_link`."""
)
# Проверка статуса создания задачи, если создано без ошибок - извлекаем ID задачи, иначе возвращаем ответ сервера
if captcha_id["errorId"] == 0:
captcha_id = captcha_id["taskId"]
self.result_payload.update({"taskId": captcha_id})
else:
return captcha_id
# если передан параметр `callbackUrl` - не ждём решения капчи а возвращаем незаполненный ответ
if self.task_payload.get("callbackUrl"):
return self.result_payload
else:
# Ожидаем решения капчи
time.sleep(self.sleep_time)
return get_sync_result(result_payload=self.result_payload, sleep_time=self.sleep_time)
class aioSquareNetTextTask:
"""
Данный метод подходит для всинхронного решения капчи-изображение.
Подробней информацию смотрите в методе 'captcha_handler' и '__init__'
"""
def __init__(
self,
anticaptcha_key: str,
sleep_time: int = 5,
save_format: str = "temp",
callbackUrl: str = None,
):
"""
Инициализация нужных переменных, создание папки для изображений и кэша
После завершения работы - удаляются временные фалйы и папки
:param anticaptcha_key: АПИ ключ капчи из кабинета пользователя
:param sleep_time: Вермя ожидания решения капчи
:param save_format: Формат в котором будет сохраняться изображение, либо как временный файл - 'temp',
либо как обычное изображение в папку созданную библиотекой - 'const'.
:param callbackUrl: URL для решения капчи с ответом через callback
"""
if sleep_time < 5:
raise ValueError(f"Param `sleep_time` must be greater than 5. U set - {sleep_time}")
self.sleep_time = sleep_time
# проверяем переданный параметр способа сохранения капчи
if save_format in ["const", "temp"]:
self.save_format = save_format
else:
raise ValueError(
"\nWrong `save_format` parameter. Valid params: `const` or `temp`." f"\n\tYour param - `{save_format}`"
)
# Пайлоад для создания задачи
self.task_payload = {
"clientKey": anticaptcha_key,
"task": {"type": "SquareNetTask"},
"softId": app_key,
}
# задаём callbackUrl если передан
if callbackUrl:
self.task_payload.update({"callbackUrl": callbackUrl})
# отправляем запрос на результат решения капчи, если ещё капча не решена - ожидаем 5 сек
# если всё ок - идём дальше
self.result_payload = {"clientKey": anticaptcha_key}
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
if exc_type:
return False
return True
async def __image_temp_saver(self, captcha_link: str):
"""
Метод сохраняет файл изображения как временный и отправляет его сразу на сервер для расшифровки.
:return: Возвращает ID капчи
"""
# Скачиваем капчу
async with aiohttp.ClientSession() as session:
async with session.get(captcha_link) as resp:
content = await resp.content.read()
# Создаём пайлоад, вводим ключ от сайта, выбираем метод ПОСТ и ждём ответа в JSON-формате
self.task_payload["task"].update({"body": base64.b64encode(content).decode("utf-8")})
# Отправляем на рукапча изображение капчи и другие парметры,
# в результате получаем JSON ответ с номером решаемой капчи
async with aiohttp.ClientSession() as session:
async with session.post(create_task_url, json=self.task_payload) as resp:
return await resp.json()
async def __image_const_saver(self, captcha_link: str):
"""
Метод создаёт папку и сохраняет в неё изображение, затем передаёт его на расшифровку и удалет файл.
:return: Возвращает ID капчи
"""
img_path = "PythonAntiCaptchaImages"
if not os.path.exists(img_path):
os.mkdir(img_path)
# Скачиваем капчу
async with aiohttp.ClientSession() as session:
async with session.get(captcha_link) as resp:
content = await resp.content.read()
# Высчитываем хэш изображения, для того что бы сохранить его под уникальным именем
image_hash = hashlib.sha224(content).hexdigest()
with open(os.path.join(img_path, "im-{0}.png".format(image_hash)), "wb") as out_image:
out_image.write(content)
with open(os.path.join(img_path, "im-{0}.png".format(image_hash)), "rb") as captcha_image:
# Добавляем в пайлоад картинку и отправляем
self.task_payload["task"].update({"body": base64.b64encode(captcha_image.read()).decode("utf-8")})
# Отправляем на антикапча изображение капчи и другие парметры,
# в результате получаем JSON ответ содержащий номер решаемой капчи
async with aiohttp.ClientSession() as session:
async with session.post(create_task_url, json=self.task_payload) as resp:
captcha_id = await resp.json()
# удаляем файл капчи
os.remove(os.path.join(img_path, "im-{0}.png".format(image_hash)))
return captcha_id
async def __read_captcha_image_file(self, content: bytes, content_type: str = "file"):
"""
Функция отвечает за чтение уже сохранённого файла или файла в уодировке base64
:param content: Параметр строка-путь указывающий на изображение капчи для отправки её на сервер
:param content_type: Тип переданной переменной. Возможны варианты: `file` и `base64`
:return: Возвращает ID капчи
"""
try:
if content_type == "file":
with open(content, "rb") as captcha_image:
# Добавляем в пайлоад картинку и отправляем
self.task_payload["task"].update({"body": base64.b64encode(captcha_image.read()).decode("utf-8")})
elif content_type == "base64":
self.task_payload["task"].update({"body": content})
else:
raise ValueError(
f"\nWrong `content_type` parameter. Valid params: `file` or `base64`."
f"\n\tYour param - `{content_type}`"
)
# Отправляем на антикапча изображение капчи и другие парметры,
# в результате получаем JSON ответ содержащий номер решаемой капчи
async with aiohttp.ClientSession() as session:
async with session.post(create_task_url, json=self.task_payload) as resp:
captcha_id = await resp.json()
except (IOError, FileNotFoundError) as err:
raise ReadError(err)
return captcha_id
# Работа с капчёй
async def captcha_handler(
self,
objectName: str,
rowsCount: int,
columnsCount: int,
image_link: str = None,
image_file: str = None,
image_base64: str = None,
) -> dict:
"""
Этот тип задачи берет вашу картинку, добавляет на нее сетку нужного размера и отдает работнику с требованием выбрать объекты нужного типа.
:param objectName: Имя объекта.
:param rowsCount: Кол-во строк. Min 2, max 5
:param columnsCount: Кол-во колонок. Min 2, max 5
:param image_link: Ссылка на изображение
:param image_file: Необязательный параметр, служит для открытия уже скачанных файлов изображений.
:param image_base64: Загрузка изображения в кодировке base64
:return: Возвращает весь ответ сервера JSON-строкой.
"""
# проверка параметров сетки
if (2 < rowsCount < 5) and (2 < columnsCount < 5):
raise ValueError(
"Параметр `rowsCount` и `columnsCount` должен быть больше 2 и меньше 5."
f"Вы передали - `rowsCount`:{rowsCount}, `columnsCount`:{columnsCount}"
)
# обновляем task_payload новыми данными для сетки разметки и названия предмета
self.task_payload["task"].update({"objectName": objectName})
self.task_payload["task"].update({"rowsCount": rowsCount})
self.task_payload["task"].update({"columnsCount": columnsCount})
# если был передан линк на локальный скачаный файл
if image_file:
captcha_id = await self.__read_captcha_image_file(image_file, content_type="file")
elif image_base64:
captcha_id = await self.__read_captcha_image_file(image_base64, content_type="base64")
elif image_link:
# согласно значения переданного параметра выбираем функцию для сохранения изображения
if self.save_format == "const":
captcha_id = await self.__image_const_saver(image_link)
elif self.save_format == "temp":
captcha_id = await self.__image_temp_saver(image_link)
else:
raise ParamError(
additional_info="""You must set one of param - `image_file/image_base64/image_link`.\n
Вы должны задать один из параметров - `image_file/image_base64/image_link`."""
)
# Проверка статуса создания задачи, если создано без ошибок - извлекаем ID задачи, иначе возвращаем ответ сервера
if captcha_id["errorId"] == 0:
captcha_id = captcha_id["taskId"]
self.result_payload.update({"taskId": captcha_id})
else:
return captcha_id
# если передан параметр `callbackUrl` - не ждём решения капчи а возвращаем незаполненный ответ
if self.task_payload.get("callbackUrl"):
return self.result_payload
else:
# Ждем решения капчи
await asyncio.sleep(self.sleep_time)
return await get_async_result(result_payload=self.result_payload, sleep_time=self.sleep_time)

10
src/python3_anticaptcha/image_captcha.py
View File

@ -6,10 +6,10 @@ from .core.enum import CaptchaTypeEnm, SaveFormatsEnm
from .core.aio_captcha_handler import AIOCaptchaHandler
from .core.sio_captcha_handler import SIOCaptchaHandler
__all__ = ("ImageToTextCaptcha",)
__all__ = ("ImageToText",)
class ImageToTextCaptcha(CaptchaParams):
class ImageToText(CaptchaParams):
def __init__(
self,
@ -32,7 +32,7 @@ class ImageToTextCaptcha(CaptchaParams):
img_path: Folder to save captcha images
Examples:
>>> ImageToTextCaptcha(api_key="99d7d111a0111dc11184111c8bb111da",
>>> ImageToText(api_key="99d7d111a0111dc11184111c8bb111da",
... save_format=SaveFormatsEnm.CONST
... ).captcha_handler(captcha_file='files/captcha-image.jpg')
{
@ -52,7 +52,7 @@ class ImageToTextCaptcha(CaptchaParams):
"taskId": 396687629
}
>>> ImageToTextCaptcha(api_key="99d7d111a0111dc11184111c8bb111da",
>>> ImageToText(api_key="99d7d111a0111dc11184111c8bb111da",
... save_format=SaveFormatsEnm.CONST
... ).captcha_handler(captcha_link='https://........../captcha-image.jpg')
{
@ -72,7 +72,7 @@ class ImageToTextCaptcha(CaptchaParams):
"taskId": 396687629
}
>>> await ImageToTextCaptcha(api_key="99d7d111a0111dc11184111c8bb111da",
>>> await ImageToText(api_key="99d7d111a0111dc11184111c8bb111da",
... save_format=SaveFormatsEnm.CONST
... ).aio_captcha_handler(captcha_link='https://........../captcha-image.jpg')
{

36
tests/test_image_captcha.py
View File

@ -5,7 +5,7 @@ import pytest
from tests.conftest import BaseTest
from python3_anticaptcha.core.enum import SaveFormatsEnm, ResponseStatusEnm
from python3_anticaptcha.image_captcha import ImageToTextCaptcha
from python3_anticaptcha.image_captcha import ImageToText
from python3_anticaptcha.core.serializer import GetTaskResultResponseSer
from python3_anticaptcha.core.aio_captcha_handler import AIOCaptchaHandler
from python3_anticaptcha.core.sio_captcha_handler import SIOCaptchaHandler
@ -28,7 +28,7 @@ class TestImageCaptcha(BaseTest):
}
def test_sio_success_file(self):
instance = ImageToTextCaptcha(api_key=self.API_KEY)
instance = ImageToText(api_key=self.API_KEY)
result = instance.captcha_handler(captcha_file=self.captcha_file)
assert isinstance(result, dict)
@ -38,7 +38,7 @@ class TestImageCaptcha(BaseTest):
assert ser_result.cost != 0.0
async def test_aio_success_file(self):
instance = ImageToTextCaptcha(api_key=self.API_KEY)
instance = ImageToText(api_key=self.API_KEY)
result = await instance.aio_captcha_handler(captcha_file=self.captcha_file)
assert isinstance(result, dict)
@ -61,7 +61,7 @@ class TestImageCaptcha(BaseTest):
)
mocked_method.return_value = "tested"
instance = ImageToTextCaptcha(api_key=self.API_KEY, save_format=save_format, img_clearing=img_clearing)
instance = ImageToText(api_key=self.API_KEY, save_format=save_format, img_clearing=img_clearing)
result = instance.captcha_handler(captcha_link=self.captcha_url)
assert mocked_method.call_count == 1
@ -84,7 +84,7 @@ class TestImageCaptcha(BaseTest):
)
mocked_method.return_value = "tested"
instance = ImageToTextCaptcha(api_key=self.API_KEY, save_format=save_format, img_clearing=img_clearing)
instance = ImageToText(api_key=self.API_KEY, save_format=save_format, img_clearing=img_clearing)
result = await instance.aio_captcha_handler(captcha_link=self.captcha_url)
assert mocked_method.call_count == 1
@ -99,7 +99,7 @@ class TestImageCaptcha(BaseTest):
)
mocked_method.side_effect = ValueError("Test error")
instance = ImageToTextCaptcha(api_key=self.API_KEY)
instance = ImageToText(api_key=self.API_KEY)
result = instance.captcha_handler(captcha_link=self.captcha_url)
assert isinstance(result, dict)
@ -114,7 +114,7 @@ class TestImageCaptcha(BaseTest):
)
mocked_method.side_effect = ValueError("Test error")
instance = ImageToTextCaptcha(api_key=self.API_KEY)
instance = ImageToText(api_key=self.API_KEY)
result = await instance.aio_captcha_handler(captcha_link=self.captcha_url)
assert isinstance(result, dict)
@ -132,7 +132,7 @@ class TestImageCaptcha(BaseTest):
file_data = self.read_file(file_path=self.captcha_file)
instance = ImageToTextCaptcha(api_key=self.API_KEY)
instance = ImageToText(api_key=self.API_KEY)
result = instance.captcha_handler(captcha_base64=file_data)
assert captcha_params_spy.call_args.kwargs["captcha_base64"] == file_data
@ -150,7 +150,7 @@ class TestImageCaptcha(BaseTest):
file_data = self.read_file(file_path=self.captcha_file)
instance = ImageToTextCaptcha(api_key=self.API_KEY)
instance = ImageToText(api_key=self.API_KEY)
result = await instance.aio_captcha_handler(captcha_base64=file_data)
assert captcha_params_spy.call_args.kwargs["captcha_base64"] == file_data
@ -160,18 +160,18 @@ class TestImageCaptcha(BaseTest):
assert result == mocked_method.return_value
def test_methods_exists(self):
assert "captcha_handler" in ImageToTextCaptcha.__dict__.keys()
assert "aio_captcha_handler" in ImageToTextCaptcha.__dict__.keys()
instance = ImageToTextCaptcha(api_key=self.API_KEY)
assert "captcha_handler" in ImageToText.__dict__.keys()
assert "aio_captcha_handler" in ImageToText.__dict__.keys()
instance = ImageToText(api_key=self.API_KEY)
assert instance.create_task_payload.clientKey == self.API_KEY
def test_args(self):
instance = ImageToTextCaptcha(api_key=self.API_KEY)
instance = ImageToText(api_key=self.API_KEY)
assert instance.create_task_payload.clientKey == self.API_KEY
def test_del(self, mocker):
mocked_method: MagicMock = mocker.patch("shutil.rmtree")
ImageToTextCaptcha(api_key=self.API_KEY, save_format=SaveFormatsEnm.CONST, img_clearing=True)
ImageToText(api_key=self.API_KEY, save_format=SaveFormatsEnm.CONST, img_clearing=True)
assert mocked_method.call_count == 1
def test_kwargs(self, mocker):
@ -179,7 +179,7 @@ class TestImageCaptcha(BaseTest):
"python3_anticaptcha.core.sio_captcha_handler.SIOCaptchaHandler.body_file_processing"
)
instance = ImageToTextCaptcha(api_key=self.API_KEY)
instance = ImageToText(api_key=self.API_KEY)
instance.captcha_handler(**self.kwargs_params)
assert mocked_method.call_count == 1
@ -192,7 +192,7 @@ class TestImageCaptcha(BaseTest):
"python3_anticaptcha.core.aio_captcha_handler.AIOCaptchaHandler.body_file_processing"
)
instance = ImageToTextCaptcha(api_key=self.API_KEY)
instance = ImageToText(api_key=self.API_KEY)
await instance.aio_captcha_handler(**self.kwargs_params)
assert mocked_method.call_count == 1
@ -201,7 +201,7 @@ class TestImageCaptcha(BaseTest):
assert set(self.kwargs_params.values()).issubset(set(instance.task_params.values()))
def test_err_body_file_processing(self, mocker):
instance = ImageToTextCaptcha(api_key=self.API_KEY)
instance = ImageToText(api_key=self.API_KEY)
result = instance.captcha_handler(**self.kwargs_params)
assert isinstance(result, dict)
@ -212,7 +212,7 @@ class TestImageCaptcha(BaseTest):
assert ser_result.cost == 0.0
async def test_aio_err_body_file_processing(self, mocker):
instance = ImageToTextCaptcha(api_key=self.API_KEY)
instance = ImageToText(api_key=self.API_KEY)
result = await instance.aio_captcha_handler(**self.kwargs_params)
assert isinstance(result, dict)