Add saved pixel rotation for JPG, PNG and TIFF v0.9.2

app/main.py

@@ -6,6 +6,7 @@ import json
 import logging
 import os
 import re
+import shutil
 import subprocess
 import tempfile
 import threading
@@ -24,6 +25,7 @@ from pydantic import BaseModel, Field
 from fastapi.responses import FileResponse, Response, StreamingResponse
 from fastapi.staticfiles import StaticFiles
 from PIL import Image, ImageOps, UnidentifiedImageError
+from PIL import ImageSequence, JpegImagePlugin
 
 FORMAT_GROUPS: dict[str, tuple[str, ...]] = {
     "jpg": (".jpg", ".jpeg"),
@@ -89,7 +91,7 @@ APP_SUBTITLE = (
     or f"Local folder · in-place {ENABLED_FORMAT_LABEL} metadata"
 )
 APP_URL = os.getenv("APP_URL", "http://localhost:8080")
-APP_VERSION = "0.9.1"
+APP_VERSION = "0.9.2"
 MAX_UPLOAD_MB = int(os.getenv("MAX_UPLOAD_MB", "150"))
 MAX_UPLOAD_BYTES = MAX_UPLOAD_MB * 1024 * 1024
 PREVIEW_MAX_EDGE = int(os.getenv("PREVIEW_MAX_EDGE", "2400"))
@@ -98,6 +100,7 @@ PREVIEW_MAX_PIXELS = int(os.getenv("PREVIEW_MAX_PIXELS", "300000000"))
 Image.MAX_IMAGE_PIXELS = PREVIEW_MAX_PIXELS
 
 TIFF_EXTENSIONS = {".tif", ".tiff"}
+PIXEL_ROTATION_EXTENSIONS = {".jpg", ".jpeg", ".png", ".tif", ".tiff"}
 HEIF_EXTENSIONS = {".heic", ".heif", ".hif"}
 DNG_EXTENSIONS = {".dng"}
 CANON_EXTENSIONS = {".cr2", ".cr3"}
@@ -788,6 +791,146 @@ def _render_raw(path: Path, format_name: str) -> Image.Image:
     return Image.fromarray(rgb)
 
 
+def _transpose_for_clockwise_rotation(image: Image.Image, rotation: int) -> Image.Image:
+    operation = {
+        90: Image.Transpose.ROTATE_270,
+        180: Image.Transpose.ROTATE_180,
+        270: Image.Transpose.ROTATE_90,
+    }[rotation]
+    return image.transpose(operation)
+
+
+def _copy_metadata(source: Path, destination: Path) -> None:
+    completed = subprocess.run(
+        [
+            "exiftool",
+            "-overwrite_original",
+            "-m",
+            "-TagsFromFile",
+            str(source),
+            "-all:all",
+            str(destination),
+        ],
+        capture_output=True,
+        text=True,
+        timeout=90,
+        check=False,
+    )
+    if completed.returncode != 0:
+        logger.error("ExifTool metadata restore after rotation failed: %s", completed.stderr.strip())
+        raise ValueError("Metadata could not be preserved after rotating the image.")
+
+
+def _rotate_jpeg(source: Path, destination: Path, rotation: int) -> str:
+    with Image.open(source) as image:
+        orientation = int(image.getexif().get(274, 1) or 1)
+
+    if orientation == 1:
+        completed = subprocess.run(
+            [
+                "jpegtran",
+                "-copy",
+                "all",
+                "-perfect",
+                "-rotate",
+                str(rotation),
+                "-outfile",
+                str(destination),
+                str(source),
+            ],
+            capture_output=True,
+            timeout=90,
+            check=False,
+        )
+        if completed.returncode == 0 and destination.exists():
+            return "jpeg-lossless"
+
+    with Image.open(source) as image:
+        image.load()
+        corrected = ImageOps.exif_transpose(image)
+        rotated = _transpose_for_clockwise_rotation(corrected, rotation)
+        try:
+            sampling = JpegImagePlugin.get_sampling(image)
+            save_options: dict[str, Any] = {
+                "format": "JPEG",
+                "quality": 95,
+                "optimize": True,
+                "progressive": bool(image.info.get("progressive") or image.info.get("progression")),
+            }
+            if sampling >= 0:
+                save_options["subsampling"] = sampling
+            if image.info.get("icc_profile"):
+                save_options["icc_profile"] = image.info["icc_profile"]
+            rotated.save(destination, **save_options)
+        finally:
+            if corrected is not image:
+                corrected.close()
+            rotated.close()
+    _copy_metadata(source, destination)
+    return "jpeg-reencoded"
+
+
+def _rotate_png_or_tiff(source: Path, destination: Path, suffix: str, rotation: int) -> str:
+    with Image.open(source) as image:
+        image.seek(0)
+        if suffix == ".png" and int(getattr(image, "n_frames", 1)) > 1:
+            raise ValueError("Animated PNG rotation is not supported.")
+        frames: list[Image.Image] = []
+        for frame in ImageSequence.Iterator(image):
+            base = frame.copy()
+            corrected = ImageOps.exif_transpose(base)
+            if corrected is not base:
+                base.close()
+            rotated = _transpose_for_clockwise_rotation(corrected, rotation)
+            corrected.close()
+            frames.append(rotated)
+        if not frames:
+            raise ValueError("The image contains no rotatable frames.")
+        try:
+            if suffix == ".png":
+                options: dict[str, Any] = {"format": "PNG", "optimize": True}
+                if image.info.get("icc_profile"):
+                    options["icc_profile"] = image.info["icc_profile"]
+                if image.info.get("dpi"):
+                    options["dpi"] = image.info["dpi"]
+                frames[0].save(destination, **options)
+                mode = "png-lossless"
+            else:
+                options = {
+                    "format": "TIFF",
+                    "save_all": len(frames) > 1,
+                    "append_images": frames[1:],
+                    "compression": image.info.get("compression", "tiff_deflate"),
+                }
+                if image.info.get("icc_profile"):
+                    options["icc_profile"] = image.info["icc_profile"]
+                if image.info.get("dpi"):
+                    options["dpi"] = image.info["dpi"]
+                frames[0].save(destination, **options)
+                mode = "tiff-lossless-reencode"
+        finally:
+            for frame in frames:
+                frame.close()
+    _copy_metadata(source, destination)
+    return mode
+
+
+def _rotate_pixels(path: Path, suffix: str, rotation: int) -> str:
+    source = path.with_name(f"rotation-source{suffix}")
+    destination = path.with_name(f"rotation-output{suffix}")
+    shutil.copy2(path, source)
+    try:
+        if suffix in {".jpg", ".jpeg"}:
+            mode = _rotate_jpeg(source, destination, rotation)
+        else:
+            mode = _rotate_png_or_tiff(source, destination, suffix, rotation)
+        destination.replace(path)
+        return mode
+    except (Image.DecompressionBombError, UnidentifiedImageError, OSError, ValueError) as exc:
+        logger.warning("Pixel rotation failed for %s: %s", path.name, exc)
+        raise HTTPException(status_code=422, detail=f"The image could not be rotated safely: {exc}") from exc
+
+
 def _preview_bytes(temp_path: Path, suffix: str, filename: str) -> bytes:
     try:
         if suffix in RAW_EXTENSIONS:
@@ -939,6 +1082,7 @@ async def process_photo(
     latitude: Annotated[float | None, Form()] = None,
     longitude: Annotated[float | None, Form()] = None,
     location_name: Annotated[str, Form()] = "",
+    rotation: Annotated[int, Form()] = 0,
 ) -> Response:
     filename = file.filename or "photo.jpg"
     suffix = Path(filename).suffix.lower()
@@ -947,6 +1091,13 @@ async def process_photo(
             status_code=415,
             detail=f"This format is disabled. Enabled format groups: {ENABLED_FORMAT_LABEL}.",
         )
+    if rotation not in {0, 90, 180, 270}:
+        raise HTTPException(status_code=422, detail="Rotation must be 0, 90, 180, or 270 degrees.")
+    if rotation and suffix not in PIXEL_ROTATION_EXTENSIONS:
+        raise HTTPException(
+            status_code=422,
+            detail=f"Pixel rotation is not supported for {_format_name(suffix)} files.",
+        )
 
     exif_datetime, xmp_date, precision_label = _normalise_datetime(
         precision=precision,
@@ -978,6 +1129,12 @@ async def process_photo(
         upload_started = time.perf_counter()
         await _save_upload(file, temp_path, suffix)
         upload_seconds = time.perf_counter() - upload_started
+        rotation_mode = "none"
+        rotation_seconds = 0.0
+        if rotation:
+            rotation_started = time.perf_counter()
+            rotation_mode = _rotate_pixels(temp_path, suffix, rotation)
+            rotation_seconds = time.perf_counter() - rotation_started
 
         supports_iptc = suffix in IPTC_EXTENSIONS
         command = [
@@ -999,6 +1156,8 @@ async def process_photo(
         ]
         if supports_iptc:
             command.extend(["-IPTC:Caption-Abstract=", "-IPTC:Keywords="])
+        if rotation:
+            command.extend(["-EXIF:Orientation=1", "-XMP-tiff:Orientation=1"])
 
         clean_description = description.strip()
         if clean_description:
@@ -1066,9 +1225,11 @@ async def process_photo(
         updated_bytes = temp_path.read_bytes()
         response_read_seconds = time.perf_counter() - response_started
         logger.info(
-            "Processed %s: upload %.2fs, ExifTool %.2fs, response read %.2fs",
+            "Processed %s: upload %.2fs, rotation %.2fs (%s), ExifTool %.2fs, response read %.2fs",
             filename,
             upload_seconds,
+            rotation_seconds,
+            rotation_mode,
             exiftool_seconds,
             response_read_seconds,
         )
@@ -1086,8 +1247,10 @@ async def process_photo(
             "Content-Length": str(len(updated_bytes)),
             "Cache-Control": "no-store",
             "X-Date-Precision": precision_label,
+            "X-Pixel-Rotation": rotation_mode,
             "Server-Timing": (
                 f'upload;dur={upload_seconds * 1000:.1f}, '
+                f'rotation;dur={rotation_seconds * 1000:.1f}, '
                 f'exiftool;dur={exiftool_seconds * 1000:.1f}, '
                 f'read;dur={response_read_seconds * 1000:.1f}'
             ),