JPEG Compression

main.py

import json, sys, os
from PIL import Image

def isNumber(i):
    if type(i) == int or type(i) == float:
        return True
    else:
        return False

def isJPEG(filename):
    chunks = filename.split(".")
    ext = chunks[-1].upper()
    if ext == "JPG" or ext == "JPEG":
        return True
    else:
        return False

if len(sys.argv) != 2:
    print("Incorrect path for config file")
    exit()

config_path = sys.argv[1]
print("Reading configuration:" + config_path)

config_file = open(config_path)
configs = json.load(config_file)

source_dir = configs['source']
if not os.path.exists(source_dir):
    print("source path not found: " + source_dir)
    exit()

dest_dir = configs['destination']
if dest_dir == "":
    dest_dir = os.path.join(source_dir, "austin_exported")

ratio = configs['ratio']
if ratio == "" or not isNumber(ratio):
    print("Illegal compression ratio: ", ratio)
    exit()

# for safety, minimum compression ratio is 50%
if ratio < 0.5:
    ratio = 0.5

if ratio > 1:
    ratio = 1

print("* Source directory:\t\t\t", source_dir)
print("* Destination directory: \t\t", dest_dir)
print("* JPEG compression ratio(0.5~1):\t", ratio)

input("\n\nPress Enter to continue...")

# create export directory
os.makedirs(dest_dir, exist_ok=True)

source_images = []

for entry in os.listdir(source_dir):
    if os.path.isfile(os.path.join(source_dir, entry)):
        if (isJPEG(entry)):
            source_images.append(entry)

verbose_prefix = "[//]"

print(verbose_prefix, "Total images: ", len(source_images))

for idx, img in enumerate(source_images):
    src_image_path = os.path.join(source_dir, img)
    dest_image_path = os.path.join(dest_dir, img)

    src_image = Image.open(src_image_path)

    exif = src_image.info['exif']

    src_image.save(dest_image_path,
                 "JPEG",
                 optimize = True,
                 quality = int(ratio*100),
                 subsampling = 0,
                 exif = exif)

    source_size = os.stat(src_image_path).st_size
    dest_size = os.stat(dest_image_path).st_size

    if dest_size > source_size:
        # remove the bigger dest img
        os.remove(dest_image_path)

    print(f'[{idx+1}/{len(source_images)}] {img}')

recursive.py

import json, sys, os, subprocess, math
from datetime import datetime
from PIL import Image

def isNumber(i):
    if type(i) == int or type(i) == float:
        return True
    else:
        return False

def isImage(filename):
    chunks = filename.split(".")
    ext = chunks[-1].upper()
    if ext == "JPG" or ext == "JPEG" or ext == "PNG":
        return True
    else:
        return False

def appendToFile(filename, s):
    with open(filename, "a", encoding='utf-8') as f:
        f.write(f"{str(s)}\n")

DRY_RUN = False

if len(sys.argv) != 2:
    print("Incorrect path for config file")
    exit()

config_path = sys.argv[1]
print("Reading configuration:" + config_path)

config_file = open(config_path)
configs = json.load(config_file)

root_dir = configs['recursive']['root_dir']
if not os.path.exists(root_dir):
    print("root path not found: " + root_dir)
    exit()

# Create log file
datetime_str = datetime.now().strftime("%Y%m%d_%H%M%S")
log_path = os.path.join(root_dir, f"image_convert_{datetime_str}.log")

quality = configs['quality']
if quality == "" or not isNumber(quality):
    print(f"Illegal compression quality: {quality}")
    appendToFile(log_path, f"Illegal compression quality: {quality}")
    exit()

# for safety, minimum compression quality is 50%
if quality < 0.5:
    quality = 0.5

if quality > 1:
    quality = 1

ratio = 1
if configs['resize']['enable']:
    ratio = configs['resize']['ratio']

print(f"* Root directory:\t\t\t:{root_dir}")
appendToFile(log_path, f"* Root directory:\t\t\t:{root_dir}")
print(f"* Conversion log:\t\t\t:{log_path}")
print(f"* JPEG compression quality(0.5~1):\t:{quality}")
appendToFile(log_path, f"* JPEG compression quality(0.5~1):\t:{quality}")
print(f"* Resize ratio:\t\t\t\t:{ratio}")
appendToFile(log_path, f"* Resize ratio:\t\t\t\t: {ratio}")

input("\n\nPress Enter to continue...")

reduced_size = 0 # in bytes

for current_dir, under_dirs, under_files in os.walk(root_dir):
    appendToFile(log_path, f"Processing directory: {current_dir}")

    source_images = []

    for entry in under_files:
        if (isImage(entry)):
            source_images.append(entry)

    if len(source_images) == 0:
        continue

    # create export directory
    export_dir = os.path.join(current_dir, f"austin_export_image")
    if not DRY_RUN:
        os.makedirs(export_dir, exist_ok=True)

    appendToFile(log_path, f"\tTotal images: {len(source_images)}")

    for idx, image in enumerate(source_images):
        src_image_path = os.path.join(current_dir, image)
        dest_image_path = os.path.join(export_dir, image).replace(".png", ".JPEG").replace(".PNG", ".JPEG")

        appendToFile(log_path, f"\t[{idx+1}/{len(source_images)}] {image}, converting to {dest_image_path}")

        if not DRY_RUN:
            try:
                src_image = Image.open(src_image_path)
                src_image = src_image.convert("RGB")

                exif = None
                if "exif" in src_image.info:
                    exif = src_image.info['exif']
                else:
                    appendToFile(log_path, f"\t[{idx+1}/{len(source_images)}] {image}: no EXIF found")

                # image resizing
                new_width = int(src_image.size[0]*ratio)
                new_height = int(src_image.size[1]*ratio)
                src_image = src_image.resize((new_width, new_height), Image.Resampling.LANCZOS)

                if exif:
                    src_image.save(dest_image_path,
                                "JPEG",
                                optimize = True,
                                quality = int(quality*100),
                                subsampling = 0,
                                exif = exif)
                else:
                    src_image.save(dest_image_path,
                                "JPEG",
                                optimize = True,
                                quality = int(quality*100),
                                subsampling = 0)

                source_size = os.stat(src_image_path).st_size # in bytes
                dest_size = os.stat(dest_image_path).st_size

                if dest_size > source_size:
                    # remove the bigger dest image
                    os.remove(dest_image_path)
                else:
                    reduced_size += (source_size - dest_size)
                appendToFile(log_path, f"\t[{idx+1}/{len(source_images)}] {image}, from {source_size/(1024*1024):.2f}MB to {dest_size/(1024*1024):.2f}MB")
            except Exception:
                appendToFile(log_path, f"\t[{idx+1}/{len(source_images)}] {image}: exception happen! skipped")
                pass

appendToFile(log_path, f"\n====\nReduced sizes: {reduced_size/(1024*1024):.2f}MB")

sample_config.json

{
    "source": "E:\\Unedited_photos\\2023 TOKYO",
    "destination": "",
    "quality": 0.7,
    "recursive": {
        "root_dir": "E:\\eventphoto"
    },
    "resize": {
        "enable": false,
        "ratio": 0.5
    }
}