148 lines
6.4 KiB
Python
Executable File
148 lines
6.4 KiB
Python
Executable File
#!/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(image_year, 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() |