Created
July 30, 2021 09:30
-
-
Save iiey/fdcec3a979a3771f2806646233cfb4ab to your computer and use it in GitHub Desktop.
python watchdog
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| """Script is used to observe filesystem changes | |
| of given directory using 'watchdog' api | |
| Usage: | |
| watchdir <PATH> | |
| version 1.0 | |
| """ | |
| import logging | |
| import os | |
| import sys | |
| import time | |
| from watchdog.events import FileSystemEventHandler | |
| from watchdog.observers import Observer | |
| class OnObserver: | |
| """Observer class | |
| which is used to start and stop monitoring | |
| """ | |
| def __init__(self, target): | |
| """constructor of Observer takes one input argument | |
| instance attributes: | |
| observer (Observer): object of watchdog | |
| watch_dir (str): target directory to monitor | |
| Args: | |
| target(str): directory path as input argument | |
| """ | |
| self.observer = Observer() | |
| self.watch_dir = os.path.expanduser(target) | |
| def run(self): | |
| """function to start monitoring | |
| observer sends event to event_handler when file changed in watchdir | |
| event_handler obj will then execute its coressponding method | |
| """ | |
| event_handler = EventHandler() | |
| self.observer.schedule(event_handler, self.watch_dir, recursive=True) | |
| self.observer.start() | |
| try: | |
| while True: | |
| time.sleep(5) | |
| except KeyboardInterrupt: | |
| self.observer.stop() | |
| print("Observer Stopped") | |
| self.observer.join() | |
| class EventHandler(FileSystemEventHandler): | |
| """EventHandler class | |
| override methods of parent class to do disired action on change | |
| """ | |
| def on_any_event(self, event): | |
| """override method to log event with datetime | |
| """ | |
| if event.is_directory: | |
| return None | |
| logging.info("{}: {}".format(event.event_type, event.src_path)) | |
| # work only inside observed path | |
| def on_moved(self, event): | |
| """override method to notify movement | |
| work only inside observed path | |
| """ | |
| print("Moved: {} to {}".format(event.src_path, event.dest_path)) | |
| def main(): | |
| # open doc | |
| if len(sys.argv) == 2 and (sys.argv[1] == '-h' or sys.argv[1] == '--help'): | |
| print(__doc__) | |
| sys.exit(1) | |
| # set the format for logging info | |
| logging.basicConfig(level=logging.INFO, | |
| format='%(asctime)s - %(message)s', | |
| datefmt='%Y-%m-%d %H:%M:%S') | |
| # start monitoring at given path | |
| path = sys.argv[1] if len(sys.argv) > 1 else '.' | |
| watch = OnObserver(target=path) | |
| watch.run() | |
| if __name__ == '__main__': | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment