Ключевание стоковых изображений с помощью Batch API от OpenAI

Ключевание стоковых изображений с помощью Batch API от OpenAI

Атрибутирование изображений – обязательный этап их подготовки для продажи на фотостоках. У каждой работы (фотографии или иллюстрации) должно быть название, описание, ключевые слова, и все это на английском.

С ключеванием неплохо справляется ChatGPT. Но пересылать ему картинки по отдельности, а потом копировать атрибуты вручную – слишком долго. Давайте автоматизируем этот процесс.

Инструментарий

Нам понадобится:

1. PyCharm или другая IDE для Python.

2. API-ключ от OpenAI и пара долларов на балансе.

3. Пара гигабайт на каком-нибудь хостинге.

Опционально:

4. Доступ к веб-интерфейсу платформы разработчика OpenAI.

5. VPS за пределами РФ.

Постановка задачи

Я буду атрибутировать свой фотоархив, отснятый в отпусках разных лет. На языке стокеров – тревел-фото. Файлы распределены по странам и городам, а также разделены на коммерческие и так называемые editorial (с людьми или узнаваемыми достопримечательностями). Структура папок вот такая:

photo

├── Russia

│   └── Ruskeala

│       ├── editorial

│       └── commercial

└── Turkey

    └── Istanbul

        ├── editorial

        └── commercial

Для каждого файла в конечных папках editoral/commercial, являющегося изображением, нужно подобрать и записать в EXIF его название на английском языке длиной не более 200 символов, а также 50 ключевых слов. При этом для файлов из папок editorial название имеет формат

City, Country - Month Day Year: Description

Для коммерческих фото таких требований нет, но желательно использовать название города и страны в списке ключевых слов.

Вот план решения этой задачи:

1. Генерация пакета заданий для платформы OpenAI.

2. Отправка пакета и получение результатов.

3. Обработка результатов и заполнение метаданных для изображений.

Генерация пакета заданий

Мы не будем пересылать нейросети каждое фото по отдельности. Вместо этого используем Batch API. Запросы будут выполнены не сразу, а в течение суток, зато мы получим 50%-ную скидку на диалог с ChatGPT.

Обратимся к версии gpt-4o-mini, так как она имеет довольно демократичные лимиты сравнительно с gpt-4o. Вот, например, значения для аккаунтов первого уровня (менее 50 долларов израсходовано за время жизни аккаунта).

Как видим, с версией gpt-4o-mini мы можем отправлять пакеты размером до 200 тыс. токенов.

Подготовим параметры нашего скрипта в виде глобальных переменных:

valid_extensions = ('.jpg', '.jpeg')

api_key = 'ВАШ_КЛЮЧ'

your_site = 'http://ВАШ_САЙТ/'
photo_dir = 'C:\ПУТЬ\К\ФОТО'
tasks_path = "batch_tasks.jsonl"
results_path = "batch_tasks_output.jsonl"
job_id_path = "batch_job_id.txt"

description_length = 200
keywords_count = 50
prompt = f"Please create a description no more than {description_length} characters long for this image in stock style " 
         f"and a list of {keywords_count} popular single-word keywords, separated with commas. " 
         f"Tailor the description to a specific niche and target audience. "
         f"Your keywords are to enhance searchability within that niche." 
         f"If there are architectural decoration elements in the image, be sure to include them. " 
         f"If there are inscriptions in a language other than English in the photo, include their translation in the description. " 
         f"Be sure to separate the description from the list of keywords with a newline character. " 
         f"Don't write anything except a description and a list of keywords. " 
         f"If there are any plants in the picture, identify their names and weave them into the description and the keywords list. " 
         f"Ensure no word is repeated. Be sure to include in both the description and in the keywords list the next words: "

batch_output_map = {}

Назначение каждого параметра понятно из его названия. Я остановлюсь подробнее только на промпте – сердце нашей системы. Он предлагает фокусироваться на конкретной нише, то есть стоковой тематике, что позволит повысить SEO-характеристики стокового портфеля. Результат мы ожидаем в виде двух абзацев текста – описание и список ключевых слов. Нам придется добавлять к промпту ключевые слова, поэтому он заканчивается соответствующей фразой и двоеточием.

Теперь сгенерируем пакет заданий:

from urllib.parse import quote

def generate_tasks():
    with open(tasks_path, 'w') as file:
        task_index = 0
        for root, dirs, files in os.walk(photo_dir):
            path_parts = root.split(os.sep)
            # Если у нас есть три уровня директорий и мы находимся в последней из них
            if len(path_parts) >= 3 and path_parts[-1] in {'editorial', 'commercial'}:
                for filename in files:
                    if not filename.lower().endswith(valid_extensions):
                        continue
                    file_path = os.path.relpath(os.path.join(root, filename), photo_dir).replace('\', '/')
                    image_url = f"{your_site}photo/{file_path}"
                    image_url = quote(image_url, safe=':/')

                    path_parts = re.split(r'[\/]', file_path)
                    country, city = path_parts[0], path_parts[1] if len(path_parts) > 1 else (None, None)

                    new_prompt = f"{prompt} {country}, {city}" if country and city else prompt

                    task = {
                        "custom_id": f"task-{task_index}",
                        "method": "POST",
                        "url": "/v1/chat/completions",
                        "body": {
                            "model": "gpt-4o-mini",
                            "messages": [
                                {
                                    "role": "system",
                                    "content": new_prompt
                                },
                                {
                                    "role": "user",
                                    "content": [
                                        {
                                            "type": "image_url",
                                            "image_url": {
                                                "url": image_url
                                            }
                                        },
                                    ],
                                }
                            ]
                        }
                    }

                    file.write(json.dumps(task) + 'n')
                    print(f"Добавили задание: {image_url}")
                    task_index += 1

Здесь мы проходим по дереву папок, начиная с photo_dir, и для каждого файла …/editorial/*.jp(e)g или …/commercial/*.jp(e)g выделяем из пути к нему название страны и города, добавляем эти параметры к промпту и генерируем задание в виде объекта JSON. У каждого задания есть уникальный в рамках пакета custom_id, по которому мы потом определим имя файла. Тело задания состоит из двух сообщений – промпта и URL картинки на хостинге.

Предполагается, что файлы лежат на вашем хостинге в папке photo. Пропускаем URL через функцию quote() на случай, если имя файла содержит зарезервированные символы. Процентная кодировка заменит их на знак процента и номер. Например, пробел превратится в %20.

Каждое задание записывается в файл tasks_path отдельной строкой. Поэтому файл и имеет расширение .jsonl (JSON Lines).

Нужно следить за объемом полученного файла, так как есть риск превысить количество токенов в одном пакете. Как уже говорилось, для аккаунтов первого уровня и версии gpt-4o-mini пакет не должен превышать 200 тыс. токенов. Чтобы следить за лимитом, можно использовать токенизатор. По опыту, при промпте такой длины в лимит укладывается до 800 строк=JSON-объектов.

Отправка пакета и загрузка результатов

Отправим получившийся файл на обработку.

from openai import OpenAI

def send_batch():
    client = OpenAI(
        api_key=api_key
    )

    batch_file = client.files.create(
        file=open(tasks_path, "rb"),
        purpose="batch"
    )
    batch_job = client.batches.create(
        input_file_id=batch_file.id,
        endpoint="/v1/chat/completions",
        completion_window="24h"
    )
    print(f"Создали пакетное задание с ID {batch_job.id}")

    with open(job_id_path, 'w') as f:
        f.write(batch_job.id)

    return batch_job.id

Здесь мы загружаем файл с заданиям в свое хранилище на сервере, указав параметр purpose="batch". Файлу присваивается уникальный ID, который мы затем получаем и передаем в функцию создания пакетного задания. Задание тоже получит свой ID, который мы на всякий случай сохраним в отдельный файл, чтобы не потерять, а также вернем в качестве выходного параметра функции.

Попробуем получить результат:

def try_get_results():
    client = OpenAI(
        api_key=api_key
    )

    with open(job_id_path, 'r') as f:
        batch_job_id = f.read().strip()

    batch_job = client.batches.retrieve(batch_job_id)
    print(f"Статус пакетного задания: {batch_job.status}")
    if batch_job.status == 'completed':
        result = client.files.content(batch_job.output_file_id).content
        with open(results_path, 'wb') as file:
            file.write(result)
        print(f"Результаты сохранены в файл {results_path}")
    else:
        print(batch_job)

Восстановив ID задания, мы получаем его статус с помощью функции retrieve(). В случае успеха сохраняем результат в файл results_path.

То же самое можно выполнить с помощью веб-интерфейса, где удобно отслеживать проблемы с пакетами.

Обработка полученных результатов

Результат мы получили в том же формате JSON Lines.

Ответ на каждое задание записан в виде JSON-объекта:

{
  "id": "batch_req_6771…",
  "custom_id": "task-0",
  "response": {
    "status_code": 200,
    "request_id": "b247…",
    "body": {
      "id": "chatcmpl-Ajt…",
      "object": "chat.completion",
      "created": 1735500227,
      "model": "gpt-4o-mini-2024-07-18",
      "choices": [
        {
          "index": 0,
          "message": {
            "role": "assistant",
            "content": "Elegant interior showcasing intricate metalwork and lush green drapery. Ideal for modern design enthusiasts. Explore the charm of Ruskeala, Russia through this stylish setting.nnRussia, Ruskeala, interior, design, elegance, green, drapery, metalwork, decoration, style, modern, seating, artistic, ambiance, wood, vintage, upholstery, tranquility, cozy, decor, booth, pattern, texture, illumination, hospitality, creativity, aesthetic, luxurious, charming, refinement, sophistication, boutique, comfort, travel, cultural, picturesque, artisan, heritage, classic, unique, rustic, classy, inviting, traditional, fashionable, ornate.",
            "refusal": null
          },
          "logprobs": null,
          "finish_reason": "stop"
        }
      ],
      "usage": {
        "prompt_tokens": 1331,
        "completion_tokens": 131,
        "total_tokens": 1462,
        "prompt_tokens_details": {
          "cached_tokens": 0,
          "audio_tokens": 0
        },
        "completion_tokens_details": {
          "reasoning_tokens": 0,
          "audio_tokens": 0,
          "accepted_prediction_tokens": 0,
          "rejected_prediction_tokens": 0
        }
      },
      "system_fingerprint": "fp_d…"
    }
  },
  "error": null
}

Как мы видим, для задания task-0 и заданных ключевых слов «Russia, Ruskeala» нейросеть сгенерировала описание длиной 177 символов и ровно 50 ключевых слов, включая и заданные. Теперь все это надо внести в метаданные.

Создадим функцию для считывания файла формата JSON Lines в список:

def load_jsonl(filepath):
    with open(filepath, 'r', encoding='utf-8') as file:
        return [json.loads(line) for line in file]

Считаем сразу два файла – задания и результат:

def load_batch_output(tasks_file, outputs_file):
    global batch_output_map

    tasks_data = load_jsonl(tasks_file)
    outputs_data = load_jsonl(outputs_file)

    batch_output_map = {}

    tasks_index = {task['custom_id']: task for task in tasks_data}

    for output in outputs_data:
        custom_id = output.get('custom_id')
        if custom_id in tasks_index:
            task = tasks_index[custom_id]
            try:
                image_url = task['body']['messages'][1]['content'][0]['image_url']['url']
                image_url = image_url.replace(your_site, '', 1)
                content = output['response']['body']['choices'][0]['message']['content']
                batch_output_map[image_url] = content
            except (IndexError, KeyError, TypeError) as e:
                print(f'Ошибка обработки задачи {custom_id}: {e}')

Так как в ответе на пакетное задание не указаны URL фотографий, к которым эти ответы относятся, нам пришлось считать сразу два файла – исходный и полученный от API. Связывая их по custom_id, мы создали хэш-таблицу batch_output_map, в которой для каждого URL хранится ответ нейросети.

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

Функция для записи результатов в метаданные похожа на функцию генерации заданий. Точно так же проходим по папке photos с помощью os.walk().

def process_directory(root_path):
    global batch_output_map

    if not batch_output_map:
        print("Сначала загрузите результаты пакетной обработки!")
        return

    for root, dirs, files in os.walk(root_path):
        path_parts = root.split(os.sep)
        # Если у нас есть три уровня директорий и мы находимся в последней из них
        if len(path_parts) >= 3 and (path_parts[-1] == 'editorial' or path_parts[-1] == 'commercial'):
            country = path_parts[-3]  # третий с конца элемент в списке
            city = path_parts[-2]  # предпоследний элемент в списке
            category = path_parts[-1]  # последний элемент в списке

            print(f"Анализ папки {root}")
            for file in files:
                if file.lower().endswith(valid_extensions):
                    relative_file_path = os.path.join(os.path.relpath(root, root_path), file)
                    relative_file_path = relative_file_path.replace(os.sep, '/')
                    relative_file_path = f"photo/{relative_file_path}"
                    relative_file_path = quote(relative_file_path)

                    if not relative_file_path in batch_output_map:
                        print(f"Для {relative_file_path} не нашлось метаданных!!!")
                        continue

                    file_path = os.path.join(root, file)
                    year, month, day = extract_date_taken(file_path)

                    response = batch_output_map[relative_file_path].split("nn")
                    if len(response) < 2:
                        print("В ответе меньше 2 разделов!")
                    else:
                        default_title, tags = response[:2] # иногда нейросеть ставит лишние переносы строк после списка ключевых слов
                        add_metadata(file_path, default_title, category, make_tag_list(tags), month, day, year, country,
                                 city)
        else:
            print(f"{root} не commercial и не editorial, пропускаю его :-(")

Для каждого файла с изображением мы ищем соответствующую запись в глобальной переменной batch_output_map, извлекаем оттуда два абзаца текста, разделяем их с помощью функции split() и передаем в add_metadata() как название и список тегов.

Теги записаны через запятую, как мы и просили. Обычно их список заканчивается знаком точки, который нам не нужен, равно как и пробелы:

def make_tag_list(tags):
    if tags.endswith('.'):
        tags = tags[:-1]
    tags_list = tags.split(',')
    tags_list = [tag.strip() for tag in tags_list]
    return tags_list

Работа с метаданными изображений

В цифровой фотографии используется три основных формата метаданных:

• EXIF (Exchangeable Image File) – информация, встроенная в изображение камерой: время и место съемки, модель камеры и объектива, параметры съемки (фокусное расстояние, экспозиция, ISO), а также пробег камеры, незаменимое поле метаданных для любителей покупать фототехнику с рук;

• IPTC (International Press Telecommunications Council) – информация, добавленная в файл специализированным ПО. Этот формат создан для использования СМИ и включает в себя информацию, необходимую для публикации фотографии - название, описание, местоположение, информация о фотографе и авторских правах, список ключевых слов;

• XMP (Extensible Metadata Platform) – информация об изменениях, внесенных в изображение при постобработке, например, в Lightroom.

Для editorial-фото нам понадобятся параметры съемки – место и время. Следовало бы считать GPS-координаты из EXIF, но я пока обойдусь считыванием названия страны/города из папки, в которой лежит файл. Что касается времени, то его легко получить из EXIF. Используем для этого библиотеку piexif:

import piexif

def extract_date_taken(image_path):
    exif_dict = piexif.load(image_path)
    date_taken_str = exif_dict['Exif'].get(piexif.ExifIFD.DateTimeOriginal)

    if date_taken_str:
        try:
            # string parse time
            date_taken = datetime.strptime(date_taken_str.decode('utf-8'), '%Y:%m:%d %H:%M:%S')
            return date_taken.year, date_taken.month, date_taken.day
        except ValueError:
            return None, None, None
    return None, None, None

Для добавления метаданных я сначала попробовала использовать эту же библиотеку piexif.

exif_dict["0th"][piexif.ImageIFD.ImageDescription] = title
exif_dict["0th"][piexif.ImageIFD.XPKeywords] = (','.join(combined_tags)).encode('utf-16')
exif_bytes = piexif.dump(exif_dict)
piexif.insert(exif_bytes, image_path)

Однако оказалось, что Shutterstock не видит текст, записанный в exif_dict["0th"][piexif.ImageIFD.ImageDescription]. Замена на exif_dict["0th"][piexif.ImageIFD.XPSubject] не помогла. При этом теги, записанные в piexif.ImageIFD.XPKeywords, отлично отображались.

Тогда в ход пошла тяжелая артиллерия – стандарт IPTC, для работы с которым можно использовать ExifTool.

Установим ExifTool средствами ОС и запустим его из нашего скрипта с помощью библиотеки subprocess:

def add_metadata(image_path: str, title, category, tags, month, day, year, country, city):
    if category == "editorial" and country and city and day and month and year:
        title = f"{city}, {country} - {month}.{day}.{year}: " + title

    try:
        commands = [
            'C:\Program Files\exiftool\exiftool.exe',
            '-overwrite_original',
            f'-Headline={title}',
            f'-Keywords={",".join(tags)}'
            image_path
        ]

        subprocess.run(commands, check=True)
        
    except subprocess.CalledProcessError as e:
        print(f"Ошибка при выполнении ExifTool для файла {image_path}: {e}")
    except Exception as e:
        print(f"Непредвиденная ошибка при обработке файла {image_path}: {e}")

Если мы имеем дело с editorial файлом, то сразу приводим его название в соответствие с форматом, принятым на Shutterstock. Затем готовим команду для перезаписи названия изображения. Опытным путем я выяснила, что для стоков подходит параметр -Headline. Имеющиеся в ExifTool параметры -Title и -ImageDescription стоки не видят.

Но, как часто бывает, когда чинишь одно – ломается другое. Теги, которые отлично записывались с помощью piexif, не выдержали испытания ExifTool:

Пришлось добавлять теги поштучно:

       commands = [
            'C:\Program Files\exiftool\exiftool.exe',
            '-overwrite_original'
            f'-Headline={title}'
        ]
        if tags:
            for tag in tags:
                commands.append(f'-Keywords={tag}')

        commands.append(image_path)

Теперь остается запустить скрипт и любоваться процессом прибавления метаданных.

Добавление аргументов командной строки

В нашем скрипте есть несколько функций, но запускать всегда нужно только одну из них. Можно, конечно, перед каждым вызовом комментировать вызов лишних, но это слишком долго. Добавим аргументы командной строки, что особенно удобно, если скрипт работает на VPS.

import argparse

if __name__ == '__main__':
    choices = {
        'generate_tasks': generate_tasks,
        'send_batch': send_batch,
        'try_get_results': try_get_results,
        'process_output': process_output
    }
    parser = argparse.ArgumentParser(description="Обработка шагов.")
    parser.add_argument('-s', '--step', choices=choices.keys(),
                        required=True, help="Выберите шаг для обработки")

    args = parser.parse_args()

    step_function = choices.get(args.step)
    if step_function:
        step_function()
    else:
        print(f"Неизвестный шаг. Введите одно из значений {', '.join(choices.keys())}")

Здесь мы создали параметр командной строки -s (--step) со значениями, одноименными нашим функциям. Правда, за обработку вывода у нас отвечают сразу две функции, поэтому создадим для них обертку:

def process_output():
    load_batch_output(tasks_path, results_path)
    process_directory(photo_dir)

Теперь можно запускать скрипт, используя интуитивно понятные тому, кто прочел статью внимательно, значения аргумента -s:

main.py -s generate_tasks
main.py -s process_output
…

А что если мы работаем локально, из IDE? В PyCharm нам поможет настройка конфигураций запуска.

Заходим в меню Run -> Edit Configurations…, нажимаем на знак «+», указываем путь к скрипту main.py и параметры: -s generate_tasks. Повторяем для остальных значений из списка choices.

Теперь можно запускать любую конфигурацию щелчком мыши.

Дальнейшая работа

Этого скрипта мне вполне хватило для проработки своего фотоархива в несколько тысяч фото, которые понемногу начали переезжать на стоки. За время работы над статьей уже даже капнула первая продажа изображения за традиционные для Шаттерстока 10 центов 😊

Полный код скрипта лежит на Гитхаб.

Что следовало бы доработать в скрипте?

1. Токенизатор и разбиение больших пакетов заданий на более мелкие, чтобы укладываться в лимит.

2. Чтение GPS из EXIF и автоматическое определение населенного пункта, где снято фото.

3. Классификация фото на commercial и editorial средствами опять же нейросети.

4. Автоматические ресайз фото и загрузка уменьшенных копий на хостинг.