331 lines
12 KiB
Python
331 lines
12 KiB
Python
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import io
|
||
import logging
|
||
import math
|
||
import shutil
|
||
import subprocess
|
||
import tempfile
|
||
import warnings
|
||
from pathlib import Path
|
||
from typing import TYPE_CHECKING, Callable
|
||
|
||
from celery.exceptions import SoftTimeLimitExceeded
|
||
from dateutil import parser as date_parser
|
||
from django.core.files.base import ContentFile
|
||
from django.db import transaction
|
||
from django.utils import timezone
|
||
from PIL import Image, ImageOps
|
||
|
||
from commerce.models import IntegrationConnection, ProductImage
|
||
|
||
if TYPE_CHECKING:
|
||
from commerce.integrations.moysklad import MoySkladClient
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
PRODUCT_THUMBNAIL_SIZE = (480, 480)
|
||
MAX_PRODUCT_IMAGE_PIXELS = 40_000_000
|
||
MAX_PRODUCT_IMAGE_DIMENSION = 4096
|
||
OVERSIZED_IMAGE_TIMEOUT_SECONDS = 90
|
||
IMAGE_SUFFIXES = {".jpg", ".jpeg", ".png", ".webp", ".gif", ".bmp"}
|
||
|
||
|
||
class OversizedProductImage(ValueError):
|
||
pass
|
||
|
||
|
||
def _reraise_task_control_exception(error: Exception) -> None:
|
||
if isinstance(error, SoftTimeLimitExceeded):
|
||
# Celery uses this exception to stop a long-running task cleanly.
|
||
# Treating it as a recoverable image failure makes the worker run
|
||
# until the hard limit kills the process without updating the run.
|
||
raise error
|
||
|
||
|
||
def _source_updated_at(value: str):
|
||
if not value:
|
||
return None
|
||
try:
|
||
parsed = date_parser.parse(value)
|
||
except (TypeError, ValueError, OverflowError):
|
||
return None
|
||
if timezone.is_naive(parsed):
|
||
parsed = timezone.make_aware(parsed)
|
||
return parsed
|
||
|
||
|
||
def _inspect_product_image(content: bytes) -> str:
|
||
with warnings.catch_warnings():
|
||
warnings.simplefilter("error", Image.DecompressionBombWarning)
|
||
try:
|
||
with Image.open(io.BytesIO(content)) as probe:
|
||
if probe.width * probe.height > MAX_PRODUCT_IMAGE_PIXELS:
|
||
raise OversizedProductImage(
|
||
"Изображение превышает безопасный лимит разрешения."
|
||
)
|
||
image_format = (probe.format or "").upper()
|
||
probe.verify()
|
||
except (Image.DecompressionBombError, Image.DecompressionBombWarning) as exc:
|
||
raise OversizedProductImage(
|
||
"Изображение превышает безопасный лимит разрешения."
|
||
) from exc
|
||
return image_format
|
||
|
||
|
||
def _resize_oversized_image(content: bytes) -> bytes:
|
||
magick = shutil.which("magick") or shutil.which("convert")
|
||
if magick is None:
|
||
raise RuntimeError(
|
||
"Для безопасного уменьшения большого изображения нужен ImageMagick."
|
||
)
|
||
max_edge = min(
|
||
MAX_PRODUCT_IMAGE_DIMENSION,
|
||
math.isqrt(MAX_PRODUCT_IMAGE_PIXELS),
|
||
)
|
||
with tempfile.TemporaryDirectory(prefix="commerce-image-") as temporary_dir:
|
||
input_path = Path(temporary_dir) / "source-image"
|
||
output_path = Path(temporary_dir) / "normalized.webp"
|
||
input_path.write_bytes(content)
|
||
try:
|
||
result = subprocess.run(
|
||
[
|
||
magick,
|
||
"-limit",
|
||
"thread",
|
||
"1",
|
||
"-limit",
|
||
"memory",
|
||
"128MiB",
|
||
"-limit",
|
||
"map",
|
||
"256MiB",
|
||
"-limit",
|
||
"disk",
|
||
"2GiB",
|
||
"-limit",
|
||
"time",
|
||
str(OVERSIZED_IMAGE_TIMEOUT_SECONDS),
|
||
"-define",
|
||
f"jpeg:size={max_edge}x{max_edge}",
|
||
f"{input_path}[0]",
|
||
"-auto-orient",
|
||
"-thumbnail",
|
||
f"{max_edge}x{max_edge}>",
|
||
"-strip",
|
||
"-quality",
|
||
"84",
|
||
str(output_path),
|
||
],
|
||
capture_output=True,
|
||
check=False,
|
||
timeout=OVERSIZED_IMAGE_TIMEOUT_SECONDS + 5,
|
||
)
|
||
except subprocess.TimeoutExpired as exc:
|
||
raise ValueError(
|
||
"Не удалось уменьшить большое изображение за безопасное время."
|
||
) from exc
|
||
if result.returncode != 0 or not output_path.exists():
|
||
detail = result.stderr.decode("utf-8", errors="replace").strip()[:500]
|
||
raise ValueError(
|
||
"Не удалось безопасно уменьшить большое изображение."
|
||
+ (f" ImageMagick: {detail}" if detail else "")
|
||
)
|
||
resized = output_path.read_bytes()
|
||
_inspect_product_image(resized)
|
||
return resized
|
||
|
||
|
||
def _prepare_product_image(content: bytes) -> tuple[bytes, bytes, str, bool]:
|
||
try:
|
||
image_format = _inspect_product_image(content)
|
||
stored_content = content
|
||
was_resized = False
|
||
except OversizedProductImage:
|
||
stored_content = _resize_oversized_image(content)
|
||
image_format = "WEBP"
|
||
was_resized = True
|
||
|
||
with Image.open(io.BytesIO(stored_content)) as source:
|
||
source.seek(0)
|
||
image = ImageOps.exif_transpose(source)
|
||
if image.mode not in {"RGB", "RGBA"}:
|
||
image = image.convert("RGBA" if "transparency" in image.info else "RGB")
|
||
image.thumbnail(PRODUCT_THUMBNAIL_SIZE, Image.Resampling.LANCZOS)
|
||
output = io.BytesIO()
|
||
image.save(output, format="WEBP", quality=84, method=5)
|
||
suffix = {
|
||
"JPEG": ".jpg",
|
||
"PNG": ".png",
|
||
"WEBP": ".webp",
|
||
"GIF": ".gif",
|
||
"BMP": ".bmp",
|
||
}.get(image_format, ".img")
|
||
return stored_content, output.getvalue(), suffix, was_resized
|
||
|
||
|
||
def _storage_filename(
|
||
external_id: str,
|
||
source_name: str,
|
||
content: bytes,
|
||
detected_suffix: str,
|
||
*,
|
||
force_detected_suffix: bool = False,
|
||
) -> str:
|
||
suffix = Path(source_name).suffix.lower()
|
||
if force_detected_suffix or suffix not in IMAGE_SUFFIXES:
|
||
suffix = detected_suffix
|
||
digest = hashlib.sha256(content).hexdigest()[:16]
|
||
return f"{external_id}-{digest}{suffix}"
|
||
|
||
|
||
def _delete_replaced_files(image, old_names):
|
||
for field_name, old_name in old_names.items():
|
||
field = getattr(image, field_name)
|
||
if old_name and old_name != field.name:
|
||
field.storage.delete(old_name)
|
||
|
||
|
||
def sync_moysklad_product_images(
|
||
*,
|
||
connection: IntegrationConnection,
|
||
row: dict,
|
||
client: MoySkladClient,
|
||
error_callback: Callable[[object, Exception], None] | None = None,
|
||
) -> int:
|
||
external_product_id = str(row.get("id") or "").strip()
|
||
if not external_product_id:
|
||
return 0
|
||
binding = connection.source.product_bindings.select_related("product").get(
|
||
external_id=external_product_id
|
||
)
|
||
product = binding.product
|
||
images_meta = row.get("images") if isinstance(row.get("images"), dict) else {}
|
||
collection_meta = (
|
||
images_meta.get("meta") if isinstance(images_meta.get("meta"), dict) else {}
|
||
)
|
||
image_count = max(0, int(collection_meta.get("size") or 0))
|
||
if image_count == 0:
|
||
product.images.filter(source=connection.source).delete()
|
||
return 0
|
||
|
||
entity_meta = row.get("meta") if isinstance(row.get("meta"), dict) else {}
|
||
entity_type = str(entity_meta.get("type") or "")
|
||
embedded_rows = images_meta.get("rows")
|
||
remote_images = (
|
||
client.parse_product_images(embedded_rows)
|
||
if isinstance(embedded_rows, list)
|
||
else client.product_images(entity_type, external_product_id)
|
||
)
|
||
retained_ids = []
|
||
synchronized = 0
|
||
|
||
for position, remote in enumerate(remote_images):
|
||
image = None
|
||
try:
|
||
source_updated_at = _source_updated_at(remote.updated)
|
||
image = ProductImage.objects.filter(
|
||
product=product,
|
||
source=connection.source,
|
||
external_id=remote.external_id,
|
||
).first()
|
||
unchanged = bool(
|
||
image
|
||
and image.source_size == remote.size
|
||
and image.source_updated_at == source_updated_at
|
||
and image.file
|
||
and image.file.storage.exists(image.file.name)
|
||
and image.thumbnail
|
||
and image.thumbnail.storage.exists(image.thumbnail.name)
|
||
)
|
||
if unchanged:
|
||
image.product = product
|
||
image.filename = remote.filename
|
||
image.title = remote.title
|
||
image.position = position
|
||
image.is_primary = position == 0
|
||
image.save(
|
||
update_fields=[
|
||
"product",
|
||
"filename",
|
||
"title",
|
||
"position",
|
||
"is_primary",
|
||
"updated_at",
|
||
]
|
||
)
|
||
retained_ids.append(remote.external_id)
|
||
synchronized += 1
|
||
continue
|
||
|
||
source_content, _content_type = client.download_image(remote.download_href)
|
||
content, thumbnail, detected_suffix, was_resized = _prepare_product_image(
|
||
source_content
|
||
)
|
||
if was_resized:
|
||
logger.info(
|
||
"Oversized commerce product image resized",
|
||
extra={
|
||
"event": "commerce.product_image.resized",
|
||
"connection_id": str(connection.pk),
|
||
"product_id": str(product.pk),
|
||
"external_product_id": external_product_id,
|
||
"external_image_id": remote.external_id,
|
||
"source_bytes": len(source_content),
|
||
"stored_bytes": len(content),
|
||
},
|
||
)
|
||
storage_name = _storage_filename(
|
||
remote.external_id,
|
||
remote.filename,
|
||
content,
|
||
detected_suffix,
|
||
force_detected_suffix=was_resized,
|
||
)
|
||
if image is None:
|
||
image = ProductImage(
|
||
source=connection.source,
|
||
external_id=remote.external_id,
|
||
)
|
||
old_names = {
|
||
"file": image.file.name if image.file else "",
|
||
"thumbnail": image.thumbnail.name if image.thumbnail else "",
|
||
}
|
||
image.product = product
|
||
image.filename = remote.filename
|
||
image.title = remote.title
|
||
image.source_size = remote.size
|
||
image.source_updated_at = source_updated_at
|
||
image.position = position
|
||
image.is_primary = position == 0
|
||
image.file.save(storage_name, ContentFile(content), save=False)
|
||
image.thumbnail.save(
|
||
f"{remote.external_id}-{hashlib.sha256(content).hexdigest()[:16]}.webp",
|
||
ContentFile(thumbnail),
|
||
save=False,
|
||
)
|
||
with transaction.atomic():
|
||
image.save()
|
||
transaction.on_commit(
|
||
lambda current=image, previous=old_names: _delete_replaced_files(
|
||
current,
|
||
previous,
|
||
)
|
||
)
|
||
retained_ids.append(remote.external_id)
|
||
synchronized += 1
|
||
except Exception as error:
|
||
_reraise_task_control_exception(error)
|
||
if image is not None and image.pk:
|
||
retained_ids.append(remote.external_id)
|
||
if error_callback is None:
|
||
raise
|
||
error_callback(remote, error)
|
||
|
||
product.images.filter(source=connection.source).exclude(
|
||
external_id__in=retained_ids
|
||
).delete()
|
||
return synchronized
|