I have made a script that syncs one folder to another. So if, for example, you saved some changes in a file in directory A, then that file would have the same contents om directory B if you set it up.
It runs on Python 3 and the watchdog module. If you are wishing to try it out (and please do), make sure to install watchdog with pip install watchdog.
Here is the code:
import logging
import pathlib
from watchdog.observers import Observer # pip install watchdog
from watchdog.events import FileSystemEventHandler # PyPI: https://pypi.org/project/watchdog/
import time
# Thanks @Tarik and @joelhed on Stack Overflow
# https://stackoverflow.com/questions/62501333/
# Set logging level
logging.basicConfig(level=logging.INFO,
format="%(asctime)s - %(levelname)s:\t%(message)s",
datefmt="%Y-%m-%d %H:%M:%S")
# List of files to sync to
# Example: [["move", [path (string), path]],
# ["create", path],
# ["delete", path],
# ["modify", path]]
changes = []
# Define the main application code
class FolderSyncer(object):
def __init__(self, src, dst):
# Define paths
self.source_path = src
self.destination_path = dst
logging.debug(f"\nSource path:\t\t{self.source_path}\n"
f"Destination path:\t{self.destination_path}")
# Make a file system observer
self.observer = Observer()
# Schedule it with our EventHandler(), the path, and recursive
self.observer.schedule(EventHandler(), str(self.source_path), recursive=True)
# If we are completely synced
self.synced = False
def __enter__(self):
logging.debug("Entered the Folder Syncer")
# Start the observer
self.observer.start()
# Must return self for context managers
return self
def run(self):
while True:
if len(changes) > 0:
# Remove first change from queue
change = changes.pop(0)
# We are still handling changes, so we are not synced
self.synced = False
logging.debug(f"Handling {change[0]} from {change[1]}")
# Handle change here, pretend to do something
if change[0] == "move":
(self.destination_path / change[1][0].replace(str(self.source_path), "")).replace(
self.destination_path / change[1][1].replace(str(self.source_path), "")
)
elif change[0] == "create":
# If it's a file
if pathlib.Path(change[1]).is_file():
# Write the file's contents
(self.destination_path / change[1].replace(str(self.source_path), "")).write_bytes(
pathlib.Path(change[1]).read_bytes()
)
# Else, it's a directory
else:
(self.destination_path / change[1].replace(str(self.source_path), "")).mkdir(exist_ok=True)
elif change[0] == "delete":
try:
# Try to remove as file
(self.destination_path / change[1].replace(str(self.source_path), "")).unlink()
except PermissionError:
# It's a directory, so remove it as a directory
(self.destination_path / change[1].replace(str(self.source_path), "")).rmdir()
elif change[0] == "modify":
try:
(self.destination_path / change[1].replace(str(self.source_path), "")).write_bytes(
pathlib.Path(change[1]).read_bytes()
)
except PermissionError:
pass
logging.info(f"Finished handling {change[0]} from {change[1]}, {len(changes)} changes left!")
else:
if not self.synced:
self.synced = True
logging.info("You are all completely synced!")
time.sleep(1)
def __exit__(self, exc_type, exc_value, traceback):
logging.warning("Exited the Folder Syncer")
# Stop the observer
self.observer.stop()
# Join the observer to the current thread
self.observer.join()
# Define an event handler
class EventHandler(FileSystemEventHandler):
def on_moved(self, event):
super(EventHandler, self).on_moved(event)
what = "directory" if event.is_directory else "file"
logging.debug(f"Moved {what}: from {event.src_path} to {event.dest_path}")
changes.append(["move", [event.src_path, event.dest_path]])
def on_created(self, event):
super(EventHandler, self).on_created(event)
what = "directory" if event.is_directory else "file"
logging.debug(f"Created {what}: {event.src_path}")
changes.append(["create", event.src_path])
def on_deleted(self, event):
super(EventHandler, self).on_deleted(event)
what = "directory" if event.is_directory else "file"
logging.debug(f"Deleted {what}: {event.src_path}")
changes.append(["delete", event.src_path])
def on_modified(self, event):
super(EventHandler, self).on_modified(event)
what = "directory" if event.is_directory else "file"
logging.debug(f"Modified {what}: {event.src_path}")
changes.append(["modify", event.src_path])
with FolderSyncer(pathlib.Path(r"U:"), pathlib.Path(r"F:\USB 64GB sync")) as folder_syncer:
folder_syncer.run()
You will most likely want to change the directory you want to sync and the location. To do that, scroll down to the bottom and change the parameters in the pathlib.Path() objects. For example, if you want to change the directory you want to sync D: to E:, than you would change:
with FolderSyncer(pathlib.Path(r"U:"), pathlib.Path(r"F:\USB 64GB sync")) as folder_syncer:
folder_syncer.run()
to
with FolderSyncer(pathlib.Path(r"D:"), pathlib.Path(r"E:")) as folder_syncer:
folder_syncer.run()
I would appreciate any code optimizations, bug-squashing, clean-ups (It's quite messy, even with classes), and some security fixes are good too. Performance is a plus, but I would like to keep it readable too.
Thanks in advance!
(̶O̶h̶,̶ ̶a̶n̶d̶ ̶B̶T̶W̶,̶ ̶w̶h̶y̶ ̶i̶s̶ ̶̶f̶o̶l̶d̶e̶r̶̶ ̶n̶o̶t̶ ̶a̶ ̶t̶a̶g̶?̶)̶ Totally didn't forget that folder means directory*. Thanks Reinderien
*sarcasm
file-systemcaptures it well enough. \$\endgroup\$