Разметка данных — тренируемся на кошках

Погружаясь все глубже в процессы автоматизации в какой то момент ты сталкиваешься с необходимостью разметки данных, хотя буквально пару недель назад, словосочетания - разметка данных и ты, стояли на вечеренике под названием "Заработок в интернетах" в разных комнатах, вернее ты стоял около бассейна, а разметка данных была на третьем этаже, курила на балконе со специалистами в области машинного обучения. Как мы встретились? Вероятно кто-то столкнул ее с балкона в бассейн, а я помог ей выбраться, попутно замочив и свою одежду.

И вот, вы сидите на кухне, курите одну сигарету на двоих и пытаетесь разобраться, чем каждый из вас занимается, и как вы можете быть друг другу полезными?

В общем не так важно, для чего мне это понадобилось, но тот факт, что у меня это получилось намного интереснее. И теперь, когда вам уже достаточно душно (или нет), переходим к сути.

Задача

В заголовке приведена такая фраза - тренируемся на кошках, и нет, это не метафора. Это прямое указание на то, что необходимо сделать. Нужно определить, что изображено на фотографии (какое животное), а что будет делать с этой информацией заказчик, дело третье.

Задача, в принципе, максимально понятна. И самое главное, максимально проста. Теперь осталось ее реализовать. А будем мы это делать через сервис по разметке данных. Действительно, не вручную же мне писать всякие программы для разметки, правда? Да я и не умею)))

В общем, идем по пути упрощения.

Итак, задача есть, способ решения есть, теперь к подробностям:

На входе мы подаем набор фотографий и картинок, на которых изображены различные животные. Задача - в ответе получить текстовое описание животного изображенного на изображении. Объем большой, поэтому руками загружать этот объем не вариант. Будем отправлять все это по АПИ, для чего пишем простой скрипт.

Скрипт

Для начала нам нужно импортировать необходимые библиотеки. В нашем скрипте мы будем использовать requests для выполнения HTTP-запросов, base64 для кодирования изображений, os для работы с файловой системой и json для обработки JSON данных.

import requests import base64 import os import json

Функция для создания задачи

Теперь создадим функцию, которая будет создавать задачу на сервер сервиса по разметке данных. Это будет функция create_task. Она будет принимать URL API, ID проекта, путь к изображению и ключ API.

Шаги выполнения:

Открываем изображение и кодируем его в base64.
Формируем спецификацию задачи (task_spec), содержащую закодированное изображение.
Создаем payload для запроса.
Устанавливаем заголовки запроса, включая ключ API.
Отправляем POST-запрос на сервер 2Captcha.
Обрабатываем ответ сервера и возвращаем результат.

def create_task(api_url, project_id, image_path, api_key): try: with open(image_path, "rb") as image_file: encoded_image = base64.b64encode(image_file.read()).decode('utf-8') image_data = f"data:image/jpeg;base64,{encoded_image}" task_spec = [ { "image_with_animal": image_data } ] payload = { "project_id": project_id, "task_spec": task_spec # Массив с одним объектом внутри } headers = { "Content-Type": "application/json", "Authorization": f"{api_key}" } print("Отправляемые данные:", json.dumps(payload, indent=4)) # Логирование данных перед отправкой response = requests.post(api_url + "/tasks", json=payload, headers=headers) if response.status_code == 201: print(f"Задача успешно создана для файла {image_path}") return True else: print(f"Ошибка при создании задачи для файла {image_path}: {response.status_code}, {response.text}") return False except Exception as e: print(f"Произошла ошибка при создании задачи для файла {image_path}: {str(e)}") return False

В ходе тестирования скрипта, был вынужден создать защиту от дурака. Функция validate_project_id проверяет, корректен ли указанный ID проекта. Она отправляет GET-запрос на сервер сервиса и возвращает результат проверки.

def validate_project_id(api_url, project_id, api_key): headers = { "Authorization": f"{api_key}" } response = requests.get(f"{api_url}/projects/{project_id}", headers=headers) return response.status_code == 200

Так как изображения не сразу будут загружены в проект, а будут подаваться туда некоторыми порциями, потребовалась функцияя для обработки и проверки изображений на повторение. Функция process_images обрабатывает все изображения в указанной директории. Она проверяет валидность ID проекта, считывает изображения, проверяет, были ли они уже отправлены, и создает задачи для новых изображений.

Шаги выполнения:

Проверка валидности ID проекта.
Загрузка истории отправленных изображений.
Перебор всех файлов в указанной директории.
Проверка, является ли файл изображением и не был ли он отправлен ранее.
Создание задачи для каждого нового изображения.
Обновление истории отправленных изображений.

def process_images(api_url, project_id, images_dir, api_key): try: if not validate_project_id(api_url, project_id, api_key): print(f"Неправильный `project_id`: {project_id}") return sent_images = set() # Файл для хранения отправленных изображений history_file = "sent_images.json" # Загрузка истории отправленных изображений, если файл существует if os.path.exists(history_file): with open(history_file, "r") as file: sent_images = set(json.load(file)) # Проход по всем файлам в директории for filename in os.listdir(images_dir): image_path = os.path.join(images_dir, filename) # Проверяем, является ли файл изображением и не был ли он отправлен ранее if os.path.isfile(image_path) and filename.lower().endswith(('.png', '.jpg', '.jpeg')) and filename not in sent_images: print(f"Обработка файла: {image_path}") if create_task(api_url, project_id, image_path, api_key): sent_images.add(filename) # Обновление истории отправленных изображений with open(history_file, "w") as file: json.dump(list(sent_images), file) print("Процесс завершен успешно.") except Exception as e: print(f"Произошла ошибка при обработке изображений: {str(e)}")

В основном блоке программы мы задаем параметры: URL API, ID проекта, директорию с изображениями и ключ API. Затем вызываем функцию process_images.

# Пример использования функции if __name__ == "__main__": api_url = "http://dataapi.2captcha.com" # Обновленный URL API для создания задач project_id = 64 # Замените на ваш ID проекта images_dir = "C:/images" # Укажите директорию с изображениями api_key = "Ваш ключ АПИ" # Замените на ваш API ключ # Проверка наличия изображений в директории if not os.path.isdir(images_dir): print(f"Директория {images_dir} не существует") else: image_files = [f for f in os.listdir(images_dir) if f.lower().endswith(('.png', '.jpg', '.jpeg'))] if not image_files: print(f"В директории {images_dir} нет изображений для обработки") else: print(f"Найдено {len(image_files)} изображений для обработки") process_images(api_url, project_id, images_dir, api_key)

Таким образом у меня получился создали скрипт на Python, который автоматизирует процесс создания задач на платформе по разметке данных. Скрипт считывает изображения из директории, кодирует их в base64, отправляет их на сервер и сохраняет историю отправленных изображений. Собственно, вот весь скрипт ниже.

import requests import base64 import os import json def create_task(api_url, project_id, image_path, api_key): try: with open(image_path, "rb") as image_file: encoded_image = base64.b64encode(image_file.read()).decode('utf-8') image_data = f"data:image/jpeg;base64,{encoded_image}" task_spec = [ { "image_with_animal": image_data } ] payload = { "project_id": project_id, "task_spec": task_spec # Массив с одним объектом внутри } headers = { "Content-Type": "application/json", "Authorization": f"{api_key}" } print("Отправляемые данные:", json.dumps(payload, indent=4)) # Логирование данных перед отправкой response = requests.post(api_url + "/tasks", json=payload, headers=headers) if response.status_code == 201: print(f"Задача успешно создана для файла {image_path}") return True else: print(f"Ошибка при создании задачи для файла {image_path}: {response.status_code}, {response.text}") return False except Exception as e: print(f"Произошла ошибка при создании задачи для файла {image_path}: {str(e)}") return False def validate_project_id(api_url, project_id, api_key): headers = { "Authorization": f"{api_key}" } response = requests.get(f"{api_url}/projects/{project_id}", headers=headers) return response.status_code == 200 def process_images(api_url, project_id, images_dir, api_key): try: if not validate_project_id(api_url, project_id, api_key): print(f"Неправильный `project_id`: {project_id}") return sent_images = set() # Файл для хранения отправленных изображений history_file = "sent_images.json" # Загрузка истории отправленных изображений, если файл существует if os.path.exists(history_file): with open(history_file, "r") as file: sent_images = set(json.load(file)) # Проход по всем файлам в директории for filename in os.listdir(images_dir): image_path = os.path.join(images_dir, filename) # Проверяем, является ли файл изображением и не был ли он отправлен ранее if os.path.isfile(image_path) and filename.lower().endswith(('.png', '.jpg', '.jpeg')) and filename not in sent_images: print(f"Обработка файла: {image_path}") if create_task(api_url, project_id, image_path, api_key): sent_images.add(filename) # Обновление истории отправленных изображений with open(history_file, "w") as file: json.dump(list(sent_images), file) print("Процесс завершен успешно.") except Exception as e: print(f"Произошла ошибка при обработке изображений: {str(e)}") # Пример использования функции if __name__ == "__main__": api_url = "http://dataapi.2captcha.com" # Обновленный URL API для создания задач project_id = 64 # Замените на ваш ID проекта images_dir = "C:/images" # Укажите директорию с изображениями api_key = "Ваш Ключ АПИ" # Замените на ваш API ключ # Проверка наличия изображений в директории if not os.path.isdir(images_dir): print(f"Директория {images_dir} не существует") else: image_files = [f for f in os.listdir(images_dir) if f.lower().endswith(('.png', '.jpg', '.jpeg'))] if not image_files: print(f"В директории {images_dir} нет изображений для обработки") else: print(f"Найдено {len(image_files)} изображений для обработки") process_images(api_url, project_id, images_dir, api_key)

Теперь, чтобы скрипт заработал необходимо все подготовить:

Создаем папку, где создаем файл с раширением .py и называем его, как нравится, у меня с фантазией не очень, поэтому это будет script.py

Также, создаем в папке подпапку, где будут храниться наши изображения. Путь к этой папке прописываем в скрипте - 84 строка.

Теперь нам нужен ключ АПИ, URL АПИ и номер проекта - 85, 82 и 83 строки в скрипте, соответсвенно.

Все это забираем в сервисе разметки данных.

Ключ АПИ и номер проекта в вашем дашборде.

А УРЛ АПИ забираем в документации АПИ сервиса, но я вам его и так уже прописал в скрипте. Правда если нужно что-то более сложное, чем разметка животных, можете изучить и ее, в качестве академического интереса.

Ну и собственно, необходимо создать сам проект в сервисе, чтобы было куда отправлять изображения. В теории, можно было все отправить через АПИ, но чет я психанул и сделал все это руками, если есть желание, можете разобраться самостоятельно, как вообще все слать через АПИ.

Итак, нажимаем на кнопку "Создать проект"

Заполняем поля "Заголовок", "Описание" и "Публичное описание". Разница между описанием и публичным описанием в том, что первое - это короткое описание, а публичное - это описание задачи. Не спрашивайте почему так, это вне моей компетенции.

Выбираем язык (1) и создаем две спецификации (2 и 3). (2) - это поля для отправки изображений. Там всего два варианта - изображение или текст, в нашем случае шлем картинку, значит выбираем изображение.

(3) - Это поля для работника, то есть по факту - это поля, где он будет писать нам свой ответ (название животного). Так как мне нужно, чтобы он написал в ответе, что за животное изображено на картинке - использую Input. Кроме input там есть еще селект, радио, чекбокс. В общем есть из чего выбрать.

Ниже на скриншоте можно увидеть нажатый чекбокс "Required" - это своего рода защита для самого себя (двойной контроль) - чтобы не отправить пустое задание. То есть, если он нажат, задание не создастся, пока не будет выполнено условие (в нашем случае наличие изображения).

Есть еще возможность получения результата ответов напрямую на сервер, но мне это не понадобилось. Вероятно скоро уже понадобится, но не в этот раз.

Собственно все, сохраняем поект, копируем его номер и вставляем в скрипт (строка 83) и можно запускать! Запускаем его командой python script.py в консоли разработчика.

Далее задача улетает работнику и после того, как она решена в личном кабинете пояляются ответы в таком вот формате

Собственно все, задача решена.

Разметка данных — тренируемся на кошках

Функция для проверки валидности ID проекта

Функция для обработки изображений

Основной блок программы

Настройка скрипта