Skip to content

Instantly share code, notes, and snippets.

@chocolatkey
Created April 2, 2019 05:16
Show Gist options
  • Save chocolatkey/c00b973f2eda9575707ae6a2dd5e76ce to your computer and use it in GitHub Desktop.
Save chocolatkey/c00b973f2eda9575707ae6a2dd5e76ce to your computer and use it in GitHub Desktop.
PNG Image optimizer to run as a cronjob using oxipng
# -*- coding: utf-8 -*-
# Cron PNG Image optimizer
# chocolatkey (Henry) 2019-03-12
version = "1.1.0"
BASE = "/your/path"
LOCK_FILE = BASE + "/optim.lock"
import os
import sys
import csv
import datetime
from queue import Queue
import threading
import multiprocessing
THREADS = multiprocessing.cpu_count()
def worker():
global already_optimized, q
while True:
obj = q.get()
if obj is None:
break
fpath = os.path.join(obj[0], obj[1])
print("Optimizing " + fpath)
# Run oxipng
os.system("nice -5 oxipng -o 4 -i 0 --strip safe " + fpath)
# Add to optimized list
already_optimized.append(obj[1])
q.task_done()
def main():
global already_optimized, q
print("CronOptimPNG " + version)
if os.path.exists(LOCK_FILE):
print("Lock file exists, exiting")
sys.exit(0)
else:
with open(LOCK_FILE, 'w', encoding='utf-8') as lockfile:
lockfile.write(str(datetime.datetime.now()))
for root, dirs, files in os.walk(BASE):
if root is BASE:
continue
# Each part cache directory
already_optimized = []
print("---------" + root + "---------")
stat_path = root + "_optim.csv"
if os.path.exists(stat_path):
print("Stat file exists for " + root)
with open(stat_path, newline='', encoding='utf-8') as csvfile:
stat_reader = csv.reader(csvfile)
for row in stat_reader:
already_optimized.append(row[0]) # add filename to skip list
print("Directory " + root + " has " + str(len(already_optimized)) + ", " + str((len(files) - len(already_optimized))) + " left")
else:
print("Directory " + root + " has not yet been optimized at all")
q = Queue(maxsize=0)
threads = []
for i in range(THREADS):
t = threading.Thread(target=worker)
t.start()
threads.append(t)
# Optimize files
for name in files:
if not name.endswith(".png"):
continue
if name in already_optimized:
print(name + " already optimized")
else:
q.put((root, name))
for i in range(THREADS):
q.put(None)
for t in threads:
t.join()
csv_data = []
for item in already_optimized:
csv_data.append((item,))
with open(stat_path, 'w', newline='', encoding='utf-8') as csvfile:
writer = csv.writer(csvfile)
writer.writerows(csv_data)
print("Finished, removing lock")
os.remove(LOCK_FILE)
if __name__ == '__main__':
sys.exit(main())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment