ahe360
commented
4 months ago
import os
from os import scandir, rename
from os.path import splitext, exists, join
from shutil import move
from time import sleep
import logging
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
# Define these - varies per user
source_dir = "/Users/<your_username>/Downloads"
dest_dir_sfx = "/Users/<your_username>/Desktop/Sound"
dest_dir_music = "/Users/<your_username>/Desktop/Sound/music"
dest_dir_video = "/Users/<your_username>/Desktop/Videos"
dest_dir_image = "/Users/<your_username>/Desktop/Images"
dest_dir_documents = "/Users/<your_username>/Desktop/Docs"
# Image Types
image_extensions = [".jpg", ".jpeg", ".jpe", ".jif", ".jfif", ".jfi", ".png", ".gif", ".webp", ".tiff",
".tif", ".psd", ".raw", ".arw", ".cr2", ".nrw", ".k25", ".bmp", ".dib", ".heif",
".heic", ".ind", ".indd", ".indt", ".jp2", ".j2k", ".jpf", ".jpf", ".jpx", ".jpm",
".mj2", ".svg", ".svgz", ".ai", ".eps", ".ico"]
# Video Types
video_extensions = [".webm", ".mpg", ".mp2", ".mpeg", ".mpe", ".mpv", ".ogg", ".mp4", ".mp4v", ".m4v",
".avi", ".wmv", ".mov", ".qt", ".flv", ".swf", ".avchd"]
# Music/Audio Types
audio_extensions = [".m4a", ".flac", "mp3", ".wav", ".wma", ".aac"]
# Document Types
document_extensions = [".doc", ".docx", ".odt", ".pdf", ".xls", ".xlsx", ".ppt", ".pptx"]
directories_to_create = [dest_dir_sfx, dest_dir_music, dest_dir_video, dest_dir_image, dest_dir_documents]
def create_directories(directories):
"""Create the necessary directories defined
Args:
'directories': List of directory paths to be created
"""
for directory in directories:
if not os.path.exists(directory):
os.makedirs(directory)
logging.info(f"Created directory: {directory}")
else:
logging.info(f"Directory already exists: {directory}")
def make_unique(dest, name):
"""Ensures a filename is unique in a given destination
Args:
'dest': Directory where the file will be saved
'name': Original filename
"""
filename, extension = splitext(name)
version = 1
# If file exists, append version by 1
while exists(f"{dest}/{name}"):
name = f"{filename}({str(version)}){extension}"
version += 1
return name
def move_file(dest, entry, name):
"""Moves a file to a destination, ensuring that file retains its original name
Args:
'dest': Destination directory
'entry': Full path of file
'name': Name of file
"""
# If the file already exists, rename it using the make_unique function
if exists(f"{dest}/{name}"):
unique_name = make_unique(dest, name)
# Full paths for old/new names
oldName = join(dest, name)
newName = join(dest, unique_name)
rename(oldName, newName)
move(entry, dest)
class MoverHandler(FileSystemEventHandler):
def on_modified(self, event):
"""A handler method that is envoked every time a file/directory is modified in the
'source_dir' directory
Args:
'event': Information of the file system event
"""
with scandir(source_dir) as entries:
for entry in entries:
name = entry.name
self.check_audio_files(entry, name)
self.check_video_files(entry, name)
self.check_image_files(entry, name)
self.check_document_files(entry, name)
def check_audio_files(self, entry, name):
"""Checks if file is an audio file based on its extension (case-insensitive), and determines its
destination directory
Args:
'entry': A DirEntry object that contains a directory entry's (i.e., file/directory) attributes
'name': Name of file
"""
dest = None
for audio_extension in audio_extensions:
if name.endswith(audio_extension) or name.endswith(audio_extension.upper()):
# Distinguish between sound effects and music based on their size
if entry.stat().st_size < 10_000_000 or "SFX" in name: # 10 MB
dest = dest_dir_sfx
else:
dest = dest_dir_music
break
if dest:
move_file(dest, entry, name)
logging.info(f"Moved audio file: {name}")
def check_video_files(self, entry, name):
"""Checks if file is a video, and determines its destination directory appropriately
Args: See check_audio_files
"""
dest = None
for video_extension in video_extensions:
if name.endswith(video_extension) or name.endswith(video_extension.upper()):
dest = dest_dir_video
break
if dest:
move_file(dest, entry, name)
logging.info(f"Moved video file: {name}")
def check_image_files(self, entry, name):
"""Checks if file is an image, and determines its destination directory appropriately
Args: See check_audio_files
"""
dest = None
for image_extension in image_extensions:
if name.endswith(image_extension) or name.endswith(image_extension.upper()):
dest = dest_dir_image
break
if dest:
move_file(dest, entry, name)
logging.info(f"Moved video file: {name}")
def check_document_files(self, entry, name):
"""Checks if file is a document, and determines its destination directory appropriately
Args: See check_audio_files
"""
dest = None
for document_extension in document_extensions:
if name.endswith(document_extension) or name.endswith(document_extension.upper()):
dest = dest_dir_documents
break
if dest:
move_file(dest, entry, name)
logging.info(f"Moved document file: {name} ")
if __name__ == "__main__": # Only run if script is executed directly
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S')
create_directories(directories_to_create)
path = source_dir
event_handler = MoverHandler()
observer = Observer()
observer.schedule(event_handler, path, recursive=True)
observer.start()
try:
while True:
sleep(10)
except KeyboardInterrupt:
observer.stop()
observer.join()
This python script supports dynamic directory creation for the user upon download. Users should update the directories in the Python file with personal directories, and run the script via terminal.