# pylint: disable=too-few-public-methods from time import time from watchdog.observers import Observer from watchdog.events import FileSystemMovedEvent, PatternMatchingEventHandler from watchdog.events import ( FileCreatedEvent, FileModifiedEvent, FileMovedEvent, FileDeletedEvent, DirCreatedEvent, DirModifiedEvent, DirMovedEvent, DirDeletedEvent ) class InotifyEvent: def __init__(self, src_path): self.date = time() self.src_path = src_path def __str__(self): return self.__repr__() def __repr__(self): return f'{self.date}\t{self.__class__.__name__}:\t{self.src_path}' class InotifyMovedEvent(InotifyEvent): def __init__(self, src_path, dest_path): self.dest_path = dest_path super().__init__(src_path) def __repr__(self): return super().__repr__()+f'\t-> {self.dest_path}' class InotifyFileCreatedEvent(InotifyEvent): pass class InotifyFileModifiedEvent(InotifyEvent): pass class InotifyFileMovedEvent(InotifyMovedEvent): pass class InotifyFileDeletedEvent(InotifyEvent): pass class InotifyDirCreatedEvent(InotifyEvent): pass class InotifyDirModifiedEvent(InotifyEvent): pass class InotifyDirMovedEvent(InotifyMovedEvent): pass class InotifyDirDeletedEvent(InotifyEvent): pass class InotifyObserver: def __init__(self, path, patterns=None, exclude=None, recursive=False): self._path = path self._patterns = patterns self._exclude = exclude self._recursive = recursive self._observer = Observer() self._reset(path, patterns, exclude) def _reset(self, path, patterns, exclude): dispatch_event = self.dispatch_event class TransformerEventHandler(PatternMatchingEventHandler): def on_any_event(self, event): dispatch_event(event) self.handler = TransformerEventHandler(patterns, exclude) self._observer.unschedule_all() self._observer.schedule(self.handler, path, self._recursive) @property def path(self): return self._path @path.setter def path(self, path): self._path = path self._reset(path, self._patterns, self._exclude) @property def patterns(self): return self._patterns @patterns.setter def patterns(self, patterns): self._patterns = patterns self._reset(self._path, patterns, self._exclude) @property def exclude(self): return self._exclude @exclude.setter def exclude(self, exclude): self._exclude = exclude self._reset(self._path, self._patterns, exclude) def start(self): self._observer.start() def stop(self): self._observer.stop() self._observer.join() def dispatch_event(self, event): event_to_action = { InotifyFileCreatedEvent : self.on_created, InotifyFileModifiedEvent : self.on_modified, InotifyFileMovedEvent : self.on_moved, InotifyFileDeletedEvent : self.on_deleted, InotifyDirCreatedEvent : self.on_created, InotifyDirModifiedEvent : self.on_modified, InotifyDirMovedEvent : self.on_moved, InotifyDirDeletedEvent : self.on_deleted } event = self.transform_event(event) self.on_any_event(event) event_to_action[type(event)](event) @staticmethod def transform_event(event): watchdog_to_inotify = { FileCreatedEvent : InotifyFileCreatedEvent, FileModifiedEvent : InotifyFileModifiedEvent, FileMovedEvent : InotifyFileMovedEvent, FileDeletedEvent : InotifyFileDeletedEvent, DirCreatedEvent : InotifyDirCreatedEvent, DirModifiedEvent : InotifyDirModifiedEvent, DirMovedEvent : InotifyDirMovedEvent, DirDeletedEvent : InotifyDirDeletedEvent } try: cls = watchdog_to_inotify[type(event)] except KeyError: raise NameError('Watchdog API returned an unknown event.') if isinstance(event, FileSystemMovedEvent): return cls(event.src_path, event.dest_path) return cls(event.src_path) def on_any_event(self, event): pass def on_created(self, event): pass def on_modified(self, event): pass def on_moved(self, event): pass def on_deleted(self, event): pass