From 915e2ae93878939859ea2e7e6c7026a572b5f90c Mon Sep 17 00:00:00 2001 From: Andrei Date: Sat, 5 Oct 2024 21:11:52 +0300 Subject: [PATCH] Delete ImageToTextTask.py --- src/python3_anticaptcha/ImageToTextTask.py | 408 --------------------- 1 file changed, 408 deletions(-) delete mode 100644 src/python3_anticaptcha/ImageToTextTask.py diff --git a/src/python3_anticaptcha/ImageToTextTask.py b/src/python3_anticaptcha/ImageToTextTask.py deleted file mode 100644 index 601ab4e..0000000 --- a/src/python3_anticaptcha/ImageToTextTask.py +++ /dev/null @@ -1,408 +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 - -SAVE_FORMATS = ("const", "temp") - - -class ImageToTextTask: - """ - Данный метод подходит для решения капчи-изображение. - Подробней информацию смотрите в методе 'captcha_handler' и '__init__' - """ - - def __init__( - self, - anticaptcha_key: str, - sleep_time: int = 5, - save_format: str = "temp", - callbackUrl: str = None, - **kwargs, - ): - """ - Инициализация нужных переменных, создание папки для изображений и кэша - После завершения работы - удаляются временные фалйы и папки - :param anticaptcha_key: АПИ ключ капчи из кабинета пользователя - :param sleep_time: Вермя ожидания решения капчи - :param save_format: Формат в котором будет сохраняться изображение, либо как временный файл - 'temp', - либо как обычное изображение в папку созданную библиотекой - 'const'. - :param callbackUrl: URL для решения капчи с ответом через callback - :param **kwargs: За подробной информацией обратитесь к документации на сайте anticaptcha. - """ - 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 SAVE_FORMATS: - self.save_format = save_format - else: - raise ValueError( - f"\nWrong `save_format` parameter. Valid params: {SAVE_FORMATS}." f"\n\tYour param - `{save_format}`" - ) - # Пайлоад для создания задачи - self.task_payload = { - "clientKey": anticaptcha_key, - "task": {"type": "ImageToTextTask"}, - "softId": app_key, - } - - # задаём callbackUrl если передан - if callbackUrl: - self.task_payload.update({"callbackUrl": callbackUrl}) - - # отправляем запрос на результат решения капчи, если ещё капча не решена - ожидаем 5 сек - # если всё ок - идём дальше - self.result_payload = {"clientKey": anticaptcha_key} - - # создаём сессию - self.session = requests.Session() - # выставляем кол-во попыток подключения к серверу при ошибке - self.session.mount("http://", HTTPAdapter(max_retries=5)) - self.session.mount("https://", HTTPAdapter(max_retries=5)) - self.session.verify = False - - # Если переданы ещё параметры - вносим их в payload - 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 __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( - 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, - captcha_link: str = None, - captcha_file: str = None, - captcha_base64: str = None, - **kwargs, - ) -> dict: - """ - Метод получает от вас ссылку на изображение, скачивает его, отправляет изображение на сервер - RuCaptcha, дожидается решения капчи и вовзращает вам результат - :param captcha_link: Ссылка на изображение - :param captcha_file: Необязательный параметр, служит для открытия уже скачанных файлов изображений. - :param captcha_base64: Загрузка изображения в кодировке base64 - :param kwargs: Дополнительные параметры для `requests.get(....)` скачивающего изображение, если передана ссылка. - :return: Возвращает весь ответ сервера JSON-строкой. - """ - - # Список допустимых параметров Session.request() - session_attrs = [ - "headers", - "cookies", - "auth", - "proxies", - "hooks", - "params", - "verify", - "cert", - "stream", - ] - session_params = kwargs.copy() - for key in kwargs: - if key in session_attrs: - setattr(self.session, key, kwargs[key]) - else: - session_params.pop(key) - - if captcha_file: - captcha_id = self.__read_captcha_image_file(captcha_file, content_type="file") - elif captcha_base64: - captcha_id = self.__read_captcha_image_file(captcha_base64, content_type="base64") - elif captcha_link: - content = self.session.get(captcha_link, **kwargs).content - # согласно значения переданного параметра выбираем функцию для сохранения изображения - if self.save_format == SAVE_FORMATS[0]: - captcha_id = self.__image_const_saver(content) - elif self.save_format == SAVE_FORMATS[1]: - captcha_id = self.__image_temp_saver(content) - else: - raise ParamError( - additional_info="""You must set one of param - `captcha_file/captcha_base64/captcha_link`.\n - Вы должны задать один из параметров - `captcha_file/captcha_base64/captcha_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, **session_params) - - -class aioImageToTextTask: - """ - Данный метод подходит для всинхронного решения капчи-изображение. - Подробней информацию смотрите в методе 'captcha_handler' и '__init__' - """ - - def __init__( - self, - anticaptcha_key: str, - sleep_time: int = 5, - save_format: str = "temp", - callbackUrl: str = None, - **kwargs, - ): - """ - Инициализация нужных переменных, создание папки для изображений и кэша - После завершения работы - удаляются временные фалйы и папки - :param anticaptcha_key: АПИ ключ капчи из кабинета пользователя - :param sleep_time: Вермя ожидания решения капчи - :param save_format: Формат в котором будет сохраняться изображение, либо как временный файл - 'temp', - либо как обычное изображение в папку созданную библиотекой - 'const'. - :param callbackUrl: URL для решения капчи с ответом через callback - :param **kwargs: За подробной информацией обратитесь к документации на сайте anticaptcha. - """ - 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 SAVE_FORMATS: - self.save_format = save_format - else: - raise ValueError( - f"\nWrong `save_format` parameter. Valid params: {SAVE_FORMATS}." f"\n\tYour param - `{save_format}`" - ) - - # Пайлоад для создания задачи - self.task_payload = { - "clientKey": anticaptcha_key, - "task": {"type": "ImageToTextTask"}, - "softId": app_key, - } - - # задаём callbackUrl если передан - if callbackUrl: - self.task_payload.update({"callbackUrl": callbackUrl}) - - # отправляем запрос на результат решения капчи, если ещё капча не решена - ожидаем 5 сек - # если всё ок - идём дальше - self.result_payload = {"clientKey": anticaptcha_key} - - # Если переданы ещё параметры - вносим их в payload - 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 __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, captcha_link: str = None, captcha_file: str = None, captcha_base64: str = None - ) -> dict: - """ - Метод получает от вас ссылку на изображение, скачивает его, отправляет изображение на сервер - RuCaptcha, дожидается решения капчи и вовзращает вам результат - :param captcha_link: Ссылка на изображение - :return: Возвращает весь ответ сервера JSON-строкой. - """ - # если был передан линк на локальный скачаный файл - if captcha_file: - captcha_id = await self.__read_captcha_image_file(captcha_file, content_type="file") - elif captcha_base64: - captcha_id = await self.__read_captcha_image_file(captcha_base64, content_type="base64") - elif captcha_link: - # согласно значения переданного параметра выбираем функцию для сохранения изображения - if self.save_format == SAVE_FORMATS[0]: - captcha_id = await self.__image_const_saver(captcha_link) - elif self.save_format == SAVE_FORMATS[1]: - captcha_id = await self.__image_temp_saver(captcha_link) - else: - raise ParamError( - additional_info="""You must set one of param - `captcha_file/captcha_base64/captcha_link`.\n - Вы должны задать один из параметров - `captcha_file/captcha_base64/captcha_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)