Реализация конвейера обработки изображений (resize, crop, watermark, format conversion)

Реализация конвейера обработки изображений (resize, crop, watermark, format conversion)

Image Processing Pipeline — автоматизированная цепочка операций над изображениями: изменение размера, обрезка, наложение водяного знака, конвертация форматов. Выполняется при загрузке или по запросу для оптимизации хранения и скорости загрузки.

Архитектура pipeline

Upload → Validation → Processing Queue → Transform → Storage → CDN
                                              ↓
                               resize → crop → watermark → convert → compress

Синхронный pipeline (при загрузке)

// Node.js + Sharp
const sharp = require('sharp')
const path = require('path')

class ImagePipeline {
  constructor(config) {
    this.config = config
  }

  async process(inputBuffer, filename) {
    const ext = path.extname(filename).toLowerCase()
    const baseName = path.basename(filename, ext)
    const results = {}

    // Базовая обработка
    const image = sharp(inputBuffer)
    const metadata = await image.metadata()

    // Конвертация в WebP для всех форматов
    const webp = await this._toWebP(image.clone(), metadata)
    results.webp = webp

    // Генерация превью разных размеров
    for (const [size, dims] of Object.entries(this.config.sizes)) {
      results[size] = await this._resize(image.clone(), dims)
      results[`${size}_webp`] = await this._resize(image.clone(), dims, 'webp')
    }

    // Водяной знак для публичных изображений
    if (this.config.watermark) {
      results.watermarked = await this._addWatermark(image.clone())
    }

    return results
  }

  async _resize(image, { width, height, fit = 'inside' }, format = null) {
    let processed = image.resize(width, height, {
      fit,
      withoutEnlargement: true,
      background: { r: 255, g: 255, b: 255, alpha: 1 }
    })

    if (format === 'webp') {
      processed = processed.webp({ quality: 85, effort: 4 })
    } else {
      processed = processed.jpeg({ quality: 85, mozjpeg: true })
    }

    return processed.toBuffer()
  }

  async _toWebP(image, metadata) {
    return image
      .webp({ quality: 85, effort: 4, smartSubsample: true })
      .toBuffer()
  }

  async _addWatermark(image) {
    const watermark = await sharp('./assets/watermark.png')
      .resize(200)
      .toBuffer()

    return image
      .composite([{
        input: watermark,
        gravity: 'southeast',
        blend: 'over'
      }])
      .toBuffer()
  }
}

const pipeline = new ImagePipeline({
  sizes: {
    thumbnail: { width: 150, height: 150, fit: 'cover' },
    medium: { width: 800, height: 600 },
    large: { width: 1920, height: 1080 },
  },
  watermark: true
})

Асинхронный pipeline через очередь

Для высокой нагрузки — обработка в фоне:

# tasks.py (Celery)
from celery import Celery
from PIL import Image
import io, boto3

app = Celery('image_tasks', broker='redis://redis:6379')

@app.task(bind=True, max_retries=3)
def process_image(self, image_id: int):
    try:
        # Загрузить исходное изображение
        record = db.get_image(image_id)
        raw = s3.get_object(Bucket='uploads', Key=record.original_key)['Body'].read()

        img = Image.open(io.BytesIO(raw))

        # Нормализовать EXIF ориентацию
        img = ImageOps.exif_transpose(img)

        # Конвертировать RGBA → RGB для JPEG
        if img.mode == 'RGBA':
            background = Image.new('RGB', img.size, (255, 255, 255))
            background.paste(img, mask=img.split()[3])
            img = background

        variants = {}
        for name, (w, h) in SIZES.items():
            resized = img.copy()
            resized.thumbnail((w, h), Image.LANCZOS)

            # Сохранить как WebP
            buf = io.BytesIO()
            resized.save(buf, 'WEBP', quality=85, method=6)
            buf.seek(0)

            key = f"processed/{image_id}/{name}.webp"
            s3.put_object(
                Bucket='media',
                Key=key,
                Body=buf,
                ContentType='image/webp',
                CacheControl='public, max-age=31536000'
            )
            variants[name] = key

        db.update_image_variants(image_id, variants)

    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)

On-the-fly трансформация (imgproxy)

Альтернатива предгенерации — imgproxy: сервис, который трансформирует изображения по URL:

docker run -p 8080:8080 \
  -e IMGPROXY_KEY=YOUR_KEY \
  -e IMGPROXY_SALT=YOUR_SALT \
  -e IMGPROXY_MAX_SRC_RESOLUTION=50 \
  darthsim/imgproxy

import hmac, hashlib, base64

def sign_imgproxy_url(path: str, key: str, salt: str) -> str:
    encoded = base64.urlsafe_b64encode(
        hmac.new(
            bytes.fromhex(key),
            bytes.fromhex(salt) + path.encode(),
            hashlib.sha256
        ).digest()
    ).rstrip(b'=').decode()
    return encoded

# URL формат: /SIGNATURE/resize:fit:WIDTH:HEIGHT:ENLARGE/format:webp/base64(URL)
source_url = base64.urlsafe_b64encode(b'https://storage.com/original/photo.jpg').decode()
path = f"/resize:fill:800:600:1/format:webp/{source_url}"
signature = sign_imgproxy_url(path, IMGPROXY_KEY, IMGPROXY_SALT)

image_url = f"https://img.company.com/{signature}{path}"

Срок выполнения

Реализация image processing pipeline с Sharp или imgproxy + интеграция с S3 — 2–3 рабочих дня.