robo-s3-images 0.1.0

Convert robot images to WebP, generate thumbnails, and store both in per-manufacturer S3 buckets.

robo-s3-images

Русский · English

Конвертация изображений роботов в WebP, генерация миниатюры и загрузка обеих версий в S3 — по отдельному бакету на производителя.

Небольшая async-библиотека для адаптеров производителей (adapter-viggo, adapter-gausium, …). Каждый адаптер направляет один ImageStorage в свой бакет; библиотека берёт на себя конвертацию, миниатюру 300px, схему путей, загрузку и удаление. FMS получает только готовые URL.

Что делает

• Только WebP. Оригинал перекодируется в WebP без изменения размера.
• Миниатюра. Версия _thumb_300 — 300px по большей стороне, пропорции сохраняются, апскейла нет.
• Единая схема путей, одинаковая для всех производителей (изоляция — за счёт отдельного бакета, а не префикса в пути):

  {serial}/{entity_type}/{entity_id}.webp
  {serial}/{entity_type}/{entity_id}_thumb_300.webp
  

• Источник — байты или URL. Вариант с URL сначала скачивает файл — это нужно при переезде с внешних источников.
• Удаляет обе версии одним вызовом — для хука удаления сущности.
• Асинхронность. Тяжёлая работа Pillow вынесена в отдельные потоки и не блокирует event loop.

Установка

uv add robo-s3-images
# или
pip install robo-s3-images

Python 3.11+. Тянет pillow, aioboto3, httpx.

Быстрый старт

from robo_s3_images import ImageRef, ImageStorage, ProcessingConfig, S3Config

storage = ImageStorage(
    S3Config(
        endpoint_url="https://s3.example.com",
        bucket="robo-images-viggo",
        access_key="...",
        secret_key="...",
        # public_base_url="https://cdn.example.com/viggo",  # если отдаётся не с endpoint
    ),
    ProcessingConfig(webp_quality=80, thumbnail_max_side=300),
)

ref = ImageRef(serial="ROBOT-123", entity_type="task-reports", entity_id="rep-42")

# Из готовых байтов:
result = await storage.store(ref, data=raw_bytes)

# Или скачать с внешнего URL производителя (скачать → конвертировать → загрузить):
result = await storage.store(ref, source_url="https://vendor.example/report.png")

result.original_url    # https://s3.example.com/robo-images-viggo/ROBOT-123/task-reports/rep-42.webp
result.thumbnail_url   # .../rep-42_thumb_300.webp

# При удалении сущности:
await storage.delete(ref)

Структура в бакете

{serial}/{entity_type}/{entity_id}.webp           # оригинал
{serial}/{entity_type}/{entity_id}_thumb_300.webp # миниатюра

entity_type — произвольное пространство имён ("task-reports", "maps", …), поэтому в одном бакете можно держать разные виды изображений. Префикс производителя не нужен — бакет и так отдельный на каждого.

API

S3Config

Доступы к одному бакету.

поле	по умолчанию	примечание
endpoint_url	—	S3 / S3-совместимый endpoint
bucket	—	бакет производителя
access_key, secret_key	—	доступы
region	None	передаётся в botocore
public_base_url	None	префикс возвращаемых URL; по умолчанию {endpoint}/{bucket} (path-style). Задайте для CDN или virtual-host домена.
addressing_style	"path"	"path" или "virtual"

config.public_base → итоговый префикс URL. config.is_configured → все четыре обязательных поля заданы.

ProcessingConfig

webp_quality=80, webp_method=4, thumbnail_max_side=300. thumbnail_max_side ограничивает бóльшую сторону миниатюры.

ImageRef(serial, entity_type, entity_id)

Идентифицирует одно изображение и раскладывается в путь объекта.

ImageStorage

метод	назначение
await store(ref, *, data=… | source_url=…)	конвертировать, загрузить обе версии, вернуть StoredImage. Перезаписывает существующие ключи (повторная миграция идемпотентна).
await delete(ref)	удалить оригинал + миниатюру одним batch-запросом
await exists(ref)	есть ли оригинал в бакете
await render(data) -> (original_webp, thumb_webp)	только байты, без загрузки
keys_for(ref) -> (original_key, thumbnail_key)	ключи объектов
urls_for(ref) -> (original_url, thumbnail_url)	предсказать URL без обращения к S3
public_url(key)	публичный URL одного ключа
is_hosted(url)	указывает ли url уже на этот бакет

store бросает StorageNotConfigured (неполные доступы), ImageProcessingError (не картинка) или SourceFetchError (не скачалось).

StoredImage

ref, original_url, thumbnail_url, original_key, thumbnail_key, original_bytes, thumbnail_bytes.

Отдельные функции

• to_webp(data, *, quality=80, method=4) -> bytes
• make_thumbnail(data, *, max_side=300, quality=80, method=4) -> bytes
• original_key(ref), thumbnail_key(ref, *, size), sanitize_segment(value)
• fetch_bytes(url, *, timeout=120, client=None) — скачивание для пути source_url; передайте общий httpx.AsyncClient, чтобы переиспользовать соединения при миграции.

Миграция существующих изображений

Идём по своей таблице, отдаём каждый внешний URL в store, сохраняем два полученных URL; перезапись при store делает повторный прогон безопасным. Один httpx-клиент на весь прогон:

import httpx
from robo_s3_images import ImageRef, ImageStorage
from robo_s3_images.fetch import fetch_bytes

async with httpx.AsyncClient(timeout=120, follow_redirects=True) as http:
    storage = ImageStorage(config, fetcher=lambda url: fetch_bytes(url, client=http))
    for row in rows:
        ref = ImageRef(row.serial, "task-reports", row.report_id)
        result = await storage.store(ref, source_url=row.image_url)
        save(row, result.original_url, result.thumbnail_url)

Разработка

uv sync          # venv + зависимости (с dev)
uv run pytest    # тесты
uv run ruff check .
uv build         # сборка sdist + wheel в dist/

S3-обёртка покрыта тестом против in-memory сервера moto; тест сам пропускается, если moto[server] не установлен.

Лицензия

MIT.