You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
118 lines
4.1 KiB
118 lines
4.1 KiB
import io
|
|
import logging
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from PIL import Image
|
|
|
|
logger = logging.getLogger(__name__)
|
|
DEBUG_PREFIX = "[image-compression]"
|
|
|
|
|
|
def _load_imgcompress():
|
|
try:
|
|
from imgcompress import ( # type: ignore
|
|
CompressionMode,
|
|
CompressionSettings,
|
|
ImageCompressor,
|
|
)
|
|
print(f"{DEBUG_PREFIX} imported imgcompress from installed environment")
|
|
return ImageCompressor, CompressionSettings, CompressionMode
|
|
except ImportError as exc:
|
|
print(f"{DEBUG_PREFIX} installed import failed: {exc}")
|
|
repo_root = Path(__file__).resolve().parents[2]
|
|
src_path = repo_root / "image-compressor" / "src"
|
|
if src_path.exists():
|
|
sys.path.insert(0, str(src_path))
|
|
print(f"{DEBUG_PREFIX} trying local source path: {src_path}")
|
|
from imgcompress import ( # type: ignore
|
|
CompressionMode,
|
|
CompressionSettings,
|
|
ImageCompressor,
|
|
)
|
|
print(f"{DEBUG_PREFIX} imported imgcompress from local source")
|
|
return ImageCompressor, CompressionSettings, CompressionMode
|
|
raise
|
|
|
|
|
|
def is_compressible_image(filename: str, content_type: str | None = None) -> bool:
|
|
name = (filename or "").lower()
|
|
ctype = (content_type or "").lower()
|
|
|
|
if name.endswith(".gif") or name.endswith(".svg"):
|
|
return False
|
|
if ctype in {"image/gif", "image/svg+xml"}:
|
|
return False
|
|
|
|
compressible_exts = (".jpg", ".jpeg", ".png", ".webp", ".bmp", ".tif", ".tiff", ".avif")
|
|
return name.endswith(compressible_exts) or ctype.startswith("image/")
|
|
|
|
|
|
def compress_image_bytes(data: bytes, filename: str) -> bytes:
|
|
if not data:
|
|
return data
|
|
|
|
print(f"{DEBUG_PREFIX} compress_image_bytes start file={filename} size={len(data)}")
|
|
ImageCompressor, CompressionSettings, CompressionMode = _load_imgcompress()
|
|
settings = CompressionSettings(
|
|
mode=CompressionMode.VISUALLY_LOSSLESS,
|
|
allow_format_switch=False,
|
|
strip_metadata=False,
|
|
png_use_zopfli=False,
|
|
)
|
|
compressor = ImageCompressor(settings=settings)
|
|
result = compressor.compress_bytes(data, Path(filename))
|
|
final_bytes = result.payload or data
|
|
print(
|
|
f"{DEBUG_PREFIX} compress_image_bytes done file={filename} "
|
|
f"original={len(data)} final={len(final_bytes)} skipped={getattr(result, 'skipped', False)} "
|
|
f"note={getattr(result, 'note', '')}"
|
|
)
|
|
return final_bytes
|
|
|
|
|
|
def maybe_compress_uploaded_file(file_obj):
|
|
filename = getattr(file_obj, "name", "upload")
|
|
content_type = getattr(file_obj, "content_type", None)
|
|
print(
|
|
f"{DEBUG_PREFIX} received upload file={filename} "
|
|
f"content_type={content_type} size={getattr(file_obj, 'size', 'unknown')}"
|
|
)
|
|
|
|
if not is_compressible_image(filename, content_type):
|
|
print(f"{DEBUG_PREFIX} skip non-compressible file={filename}")
|
|
if hasattr(file_obj, "seek"):
|
|
file_obj.seek(0)
|
|
return None
|
|
|
|
try:
|
|
raw_bytes = file_obj.read()
|
|
if hasattr(file_obj, "seek"):
|
|
file_obj.seek(0)
|
|
|
|
if not raw_bytes:
|
|
print(f"{DEBUG_PREFIX} skip empty payload file={filename}")
|
|
return None
|
|
|
|
with Image.open(io.BytesIO(raw_bytes)) as img:
|
|
img.verify()
|
|
|
|
compressed = compress_image_bytes(raw_bytes, filename)
|
|
if compressed and len(compressed) <= len(raw_bytes):
|
|
print(
|
|
f"{DEBUG_PREFIX} accepted file={filename} "
|
|
f"original={len(raw_bytes)} compressed={len(compressed)}"
|
|
)
|
|
return compressed
|
|
|
|
print(
|
|
f"{DEBUG_PREFIX} keeping original file={filename} "
|
|
f"original={len(raw_bytes)} candidate={len(compressed) if compressed else 'none'}"
|
|
)
|
|
return raw_bytes
|
|
except Exception as exc:
|
|
print(f"{DEBUG_PREFIX} failed file={filename} error={exc}")
|
|
logger.warning("Image compression skipped for %s: %s", filename, exc)
|
|
if hasattr(file_obj, "seek"):
|
|
file_obj.seek(0)
|
|
return None
|