#!/usr/bin/env python import time import signal import sys import json from os import getenv from pathlib import Path from PIL import Image, ExifTags, UnidentifiedImageError # Set the directories source_directories: list[Path] = [Path(directory_string).resolve() for directory_string in getenv('WATCH_DIRS', '/sync').split(':')] photos_directory: Path = Path(getenv('PHOTOS_DIRECTORY', '/photos')).resolve() recordings_directory: Path = Path(getenv('RECORDINGS_DIRECTORY', '/recordings')).resolve() photo_extensions: list[str] = ['.jpg', '.jpeg'] video_extensions: list[str] = ['.mp4', '.mov', '.avi', '.mkv'] sleep_duration: int = 30 def main(): # Check the source directories exist and create the ignore files if they don't exist for directory in source_directories: if not directory.exists() or not directory.is_dir(): print(f"Source directory {directory} does not exist or is not a directory. Exiting.") sys.exit(1) ignore_file: Path = directory.joinpath("ignored_files.json") if not ignore_file.exists(): print(f"Ignored files list {ignore_file} does not exist. Creating.") with open(ignore_file, 'w') as f: json.dump([], f) # Check the destination directories exist and are writable for directory in [photos_directory, recordings_directory]: if not directory.exists() or not directory.is_dir() : print(f"Destination directory {directory} does not exist or is not a directory. Exiting.") sys.exit(1) write_test_file: Path = directory.joinpath("write_test") try: write_test_file.touch() write_test_file.unlink() except PermissionError: print(f"Destination directory {directory} is not writable") sys.exit(1) # Start the main loop while True: for directory in source_directories: print(f"Starting sort of directory {directory}") start_time = time.time() sort_directory(directory = directory) end_time = time.time() print(f"Finished sorting directory {directory}") print(f"Time taken: {end_time - start_time} seconds") print(f"Sleeping for {sleep_duration} seconds") time.sleep(sleep_duration) def sort_directory(directory: Path): # Load the ignored files list ignore_file = directory.joinpath("ignored_files.json") with open(file = ignore_file) as f: ignore_list: list[str] = json.load(f) # Get all files in the directory file_list: list[Path] = [file for file in directory.iterdir() if file.is_file()] # Filter out the ignored files sort_list = [file for file in file_list if file.name not in ignore_list] # Create a list of photo files photo_list = [file for file in sort_list if file.suffix in photo_extensions] # Create a list of video files video_list = [file for file in sort_list if file.suffix in video_extensions] print(f''' Found {str(len(file_list))} total files in directory {directory.name}. {str(len(ignore_list))} to ignore. {str(len(photo_list) + len(video_list))} to sort: {str(len(photo_list))} photos. {str(len(video_list))} videos. ''') # Sort the photos print(f"Sorting {str(len(photo_list))} photos from directory {directory.name}") ignore_list = sort_photos(photo_list, ignore_list) # Sort the videos print(f"Sorting {str(len(video_list))} videos from directory {directory.name}") ignore_list = sort_videos(video_list, ignore_list) # Update the ignore_file with open(file = ignore_file, mode='w') as f: json.dump(ignore_list, f) def sort_photos(photo_list: list[Path], ignore_list: list[str]) -> list[str]: # Init the previous_name for collision avoidance previous_name: str = "" for photo in photo_list: # Read the exif data from the photo try: with Image.open(photo.resolve()) as image: exif = image.getexif() except OSError: print(f"File {photo.name} could not be opened (permissions?), ignoring.") continue except UnidentifiedImageError: print(f"File {photo.name} is not a valid image file. Ignoring.") ignore_list.append(photo.name) continue if ExifTags.Base.DateTimeOriginal in exif.keys(): image_date_time_string = exif[ExifTags.Base.DateTimeOriginal] elif ExifTags.Base.DateTime in exif.keys(): image_date_time_string = exif[ExifTags.Base.DateTime] else: image_date_time_string = None if image_date_time_string is None or image_date_time_string == '': print(f"File {photo.name} does not have a valid exif timestamp, ignoring.") ignore_list.append(photo.name) continue image_timestamp = time.strptime(image_date_time_string, "%Y:%m:%d %H:%M:%S") image_year: str = time.strftime("%Y", image_timestamp) image_date_string: str = time.strftime("%Y-%m-%d", image_timestamp) image_time_string: str = time.strftime("%H-%M-%S", image_timestamp) # Set index_count for this iteration index_count: int = 0 # Format the new name new_name: str = f"{image_date_string}_{image_time_string}_{str(index_count).zfill(3)}.jpg" while new_name == previous_name: index_count += 1 new_name = f"{image_date_string}_{image_time_string}_{str(index_count).zfill(3)}.jpg" new_path: Path = photos_directory.joinpath(f"{image_year}-Photos", new_name).resolve() # Ensure the parent year folder exists new_path.parent.mkdir(parents=True, exist_ok=True) # Move the file if not new_path.exists(): print(f"Moving {photo.as_posix()} to {new_path.as_posix()}") photo.rename(new_path) else: print(f"File {new_name} already exists, ignoring.") ignore_list.append(photo.name) # Update the previous_name for next iteration previous_name = new_name return ignore_list def sort_videos(video_list: list[Path], ignore_list: list[str]) -> list[str]: print("Video sorting not implemened yet.") return ignore_list def signal_handler(sig, frame): print('Shutdown signal received. Exiting gracefully.') sys.exit(0) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) if __name__ == "__main__": main()