File size: 1,931 Bytes
3dabe4a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 |
import hashlib
import os
from typing import List
from scripts.iib.tool import get_formatted_date, get_cache_dir, is_image_file
from concurrent.futures import ThreadPoolExecutor
import time
from PIL import Image
def generate_image_cache(dirs, size:str, verbose=False):
start_time = time.time()
cache_base_dir = get_cache_dir()
def process_image(item):
if item.is_dir():
verbose and print(f"Processing directory: {item.path}")
for sub_item in os.scandir(item.path):
process_image(sub_item)
return
if not os.path.exists(item.path) or not is_image_file(item.path):
return
try:
path = os.path.normpath(item.path)
stat = item.stat()
t = get_formatted_date(stat.st_mtime)
hash_dir = hashlib.md5((path + t).encode("utf-8")).hexdigest()
cache_dir = os.path.join(cache_base_dir, "iib_cache", hash_dir)
cache_path = os.path.join(cache_dir, f"{size}.webp")
if os.path.exists(cache_path):
verbose and print(f"Image cache already exists: {path}")
return
if os.path.getsize(path) < 64 * 1024:
verbose and print(f"Image size less than 64KB: {path}", "skip")
return
with Image.open(path) as img:
w, h = size.split("x")
img.thumbnail((int(w), int(h)))
os.makedirs(cache_dir, exist_ok=True)
img.save(cache_path, "webp")
verbose and print(f"Image cache generated: {path}")
except Exception as e:
print(f"Error generating image cache: {path}")
print(e)
with ThreadPoolExecutor() as executor:
for dir_path in dirs:
folder_listing: List[os.DirEntry] = os.scandir(dir_path)
for item in folder_listing:
executor.submit(process_image, item)
print("Image cache generation completed. ✨")
end_time = time.time()
execution_time = end_time - start_time
print(f"Execution time: {execution_time} seconds") |