58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
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") |