Convert all images and videos to a lower quality/resolution to save storage space on Ente.

ente-convert.py

# Note - This is a messy v1 of the script and needs a lot of cleaning up
# It works fine but it's not nice

"""Convert all images and videos to a lower quality/resolution to save storage space.
EXIF data is kept intact.

Warning: Do not delete your original photos! The output from this is low quality.

Setup:
    1. Copy `ente-convert.py` to somewhere on the system path.
    2. Create `ente-convert.bat` with the following text:
        @echo off
        py "%~dp0%~n0.py" %*

Usage:
    1. Load `cmd`
    2. `cd path/to/images`
    3. `ente-convert path/to/output`
"""

import os
import argparse
from PIL import Image, ImageOps
import pillow_heif
import subprocess
from concurrent.futures import ProcessPoolExecutor
import piexif
import struct
import shutil
import json


PXD = piexif.ExifIFD.PixelXDimension
PYD = piexif.ExifIFD.PixelYDimension
IW = piexif.ImageIFD.ImageWidth
IH = piexif.ImageIFD.ImageLength


def compress_video(input_folder, output_folder, max_dimension=1280):
    if not os.path.exists(output_folder):
        os.makedirs(output_folder)

    supported_formats = ('.mp4', '.mov', '.avi', '.mkv', '.flv')

    try:
        for filename in os.listdir(input_folder):
            if not filename.lower().endswith(supported_formats):
                continue

            input_path = os.path.join(input_folder, filename)
            output_path = os.path.join(output_folder, filename)


            # Command to get the video resolution
            probe_command = [
                "ffprobe",
                "-i", input_path,
                "-v", "quiet",
                "-print_format", "json",
                "-show_streams"
            ]
            result = subprocess.run(probe_command, capture_output=True, text=True)
            video_info = result.stdout

            # Parse the output to find the width and height

            data = json.loads(video_info)
            width = height = 0
            for stream in data['streams']:
                try:
                    width = stream['width']
                    height = stream['height']
                    tags = stream['tags']
                except KeyError:
                    continue
            if 0 in (width, height):
                raise RuntimeError('failed to read width and height from stream')

            if tags.get('rotate') in ('90', '-90', '270', '-270'):
                width, height = height, width

            # Calculate new resolution preserving the aspect ratio
            if width > height:
                new_width = min(width, max_dimension)
                new_height = int(new_width * height / width)
            else:
                new_height = min(height, max_dimension)
                new_width = int(new_height * width / height)

            # Command to compress and resize video
            command = [
                "ffmpeg",
                "-i", input_path,
                '-n', # don't overwrite
                "-vf", f"scale={new_width}:{new_height}",
                "-c:v", "libx264",
                "-crf", "28",  # Compression quality (lower is better)
                "-preset", "fast",
                output_path
            ]

            subprocess.run(command)
            print(f"Compressed: {filename} -> {output_path}")
            shutil.copystat(input_path, output_path)

    except Exception as e:
        print('ERROR', input_path, e)
        raise

def convert_image(image_src, image_dst):
    # Dummy implementation for demonstration
    print(f"Processing {image_src} -> {image_dst}")

    if os.path.exists(image_dst):
        try:
            with open(image_dst, "rb") as f:
                exif = piexif.load(f.read())
        except struct.error:
            pass
        else:
            if max((exif["Exif"].get(PXD, 0), exif["Exif"].get(PYD, 0), exif['0th'].get(IW, 0), exif['0th'].get(IH, 0))) < 2048:
                shutil.copystat(image_src, image_dst)
                return

    try:
        # Open the image using Pillow
        if pillow_heif.is_supported(image_src):
            # Use pillow-heif to handle HEIC
            heif_image = pillow_heif.open_heif(image_src)
            info = heif_image.info
            image = Image.frombytes(heif_image.mode, heif_image.size, heif_image.data)
        else:
            # Other formats are handled directly by Pillow
            image = Image.open(image_src)
            info = image.info

        # Resize the image, keeping aspect ratio
        image.thumbnail((2048, 2048))

        # Set the correct rotation
        image = ImageOps.exif_transpose(image)
        try:
            exif = piexif.load(info['exif'])
        except (KeyError, struct.error):
            dump = None
        else:
            exif['thumbnail'] = None
            exif['0th'][piexif.ImageIFD.Orientation] = 1
            exif["Exif"][PXD], exif["Exif"][PYD] = exif['0th'][IW], exif['0th'][IH] = image.size

            while True:
                try:
                    dump = piexif.dump(exif)

                # "dump" got wrong type of exif value.\n41729 in Exif IFD. Got as <class 'int'>.
                except ValueError as e:
                    if 'got wrong type of exif value' in str(e):
                        value, _, key = str(e).split('\n')[1].split(' ', 3)[:3]
                        del exif[key][int(value)]
                    else:
                        raise
                else:
                    break

        # Prepare output file path
        kwargs = {'quality': 85}
        if dump is not None:
            kwargs['exif'] = dump
        image.convert('RGB').save(image_dst, "JPEG", **kwargs)
        shutil.copystat(image_src, image_dst)

    except Exception as e:
        print('ERROR', image_src, e)
        raise


def convert_image_to_jpg(input_folder, output_folder):
    max_size = (2048, 2048)  # Maximum dimensions (width, height)

    # Supported image formats
    supported_formats = ('.heic', '.jpg', '.jpeg', '.png', '.tiff', '.bmp', '.gif')

    # Ensure the output directory exists
    if not os.path.exists(output_folder):
        os.makedirs(output_folder)

    # Iterate through all files in the input folder
    image_files = []
    for filename in os.listdir(input_folder):
        if not filename.lower().endswith(supported_formats):
            continue
        image_files.append(filename)

    tasks = []
    with ProcessPoolExecutor(max_workers=8) as executor:
        for i, filename in enumerate(image_files):
            image_src = os.path.join(input_folder, filename)
            image_dst = os.path.join(output_folder, os.path.splitext(filename)[0] + '.jpg')
            tasks.append(executor.submit(convert_image, image_src, image_dst))

            if not i % 32:
                while tasks:
                    tasks.pop().result()

        # Optional: wait for all tasks to complete and check results
        for task in tasks:
           task.result()  # This will re-raise any exceptions if they occurred

        # input_path = os.path.join(input_folder, filename)

        # # Open the image using Pillow
        # if filename.lower().endswith('.heic'):
        #     # Use pillow-heif to handle HEIC
        #     heif_image = pillow_heif.open_heif(input_path)
        #     image = Image.frombytes(heif_image.mode, heif_image.size, heif_image.data)
        # else:
        #     # Other formats are handled directly by Pillow
        #     image = ImageOps.exif_transpose(Image.open(input_path))

        # # Resize the image, keeping aspect ratio
        # image.thumbnail(max_size)

        # # Prepare output file path
        # base_filename = os.path.splitext(filename)[0]
        # output_path = os.path.join(output_folder, f"{base_filename}.jpg")

        # image.save(output_path, "JPEG", quality=86, exif=image.info['exif'])

        # print(f"Processed: {filename} -> {output_path}")


def convert_folders(input_folder, output_folder):
    convert_image_to_jpg(input_folder, output_folder)
    compress_video(input_folder, output_folder)
    for root, dirs, files in os.walk(input_folder):
        for d in dirs:
            copy_from = os.path.join(root, d)
            copy_to = os.path.join(root.replace(input_folder, output_folder), d)

            convert_image_to_jpg(copy_from, copy_to)
            compress_video(copy_from, copy_to)


if __name__ == "__main__":
    # Set up argument parser
    parser = argparse.ArgumentParser(description="Optimise and downscale images/videos")
    parser.add_argument("output_folder", help="Path to the output folder where the files will be saved")

    # Parse the arguments
    args = parser.parse_args()

    # Use the current working directory as the input folder
    input_folder = os.getcwd()

    # Run the conversion
    #convert_image_to_jpg(input_folder, args.output_folder)
    #compress_video(input_folder, args.output_folder)
    convert_folders(input_folder, os.path.normpath(args.output_folder))

A bit of mess happens in case the output folder is a subdirectory of the input folder as this seems to create recursive subdirectories.