Added watchdog support
This commit is contained in:
Ignacio Serantes
203
filesystemwatcher.py
Normal file
203
filesystemwatcher.py
Normal file
@@ -0,0 +1,203 @@
|
||||
import os
|
||||
try:
|
||||
from watchdog.observers import Observer
|
||||
from watchdog.events import FileSystemEventHandler
|
||||
HAVE_WATCHDOG = True
|
||||
except ImportError:
|
||||
HAVE_WATCHDOG = False
|
||||
from PySide6.QtCore import QObject, Signal, QTimer
|
||||
from constants import IMAGE_EXTENSIONS
|
||||
|
||||
|
||||
class FileSystemWatcher(QObject):
|
||||
"""
|
||||
Monitors file system events (created, deleted, modified) for specified directories.
|
||||
Emits signals to notify the main application thread of changes.
|
||||
"""
|
||||
file_created = Signal(str)
|
||||
file_deleted = Signal(str)
|
||||
file_modified = Signal(str)
|
||||
_file_modified_from_handler = Signal(str) # Internal signal from handler thread
|
||||
file_moved = Signal(str, str)
|
||||
monitoring_status_changed = Signal(bool) # Nuevo: Señal para el estado de monitoreo
|
||||
directory_moved = Signal(str, str)
|
||||
directory_modified = Signal(str) # For changes that might not be specific files
|
||||
|
||||
_modified_events_queue = {} # {path: QTimer}
|
||||
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
self._watched_directories = set()
|
||||
|
||||
if HAVE_WATCHDOG:
|
||||
self._observer = Observer()
|
||||
self._event_handler = self._Handler(self)
|
||||
self._observer.start()
|
||||
else:
|
||||
self._observer = None # Keep observer as None if watchdog is not available
|
||||
|
||||
# Debounce timer for modified events to avoid multiple signals for a single save
|
||||
self._debounce_interval = 500 # milliseconds
|
||||
|
||||
# Connect the internal signal to the debouncing slot
|
||||
if HAVE_WATCHDOG:
|
||||
self._file_modified_from_handler.connect(self._on_file_modified_debounced)
|
||||
|
||||
def _on_file_modified_debounced(self, path):
|
||||
"""Slot to handle modified events from the watchdog thread, debounced in the
|
||||
main thread."""
|
||||
# Debounce timer for modified events to avoid multiple signals for a single save
|
||||
if path in self._modified_events_queue:
|
||||
self._modified_events_queue[path].stop()
|
||||
else:
|
||||
# Ensure timer lives in the main thread (parent is self)
|
||||
timer = QTimer(self)
|
||||
timer.setSingleShot(True)
|
||||
timer.setInterval(self._debounce_interval)
|
||||
timer.timeout.connect(lambda p=path: self._emit_modified_after_debounce(p))
|
||||
self._modified_events_queue[path] = timer
|
||||
self._modified_events_queue[path].start()
|
||||
|
||||
def _emit_modified_after_debounce(self, path):
|
||||
"""Emits the file_modified signal after the debounce period."""
|
||||
self.file_modified.emit(path)
|
||||
if path in self._modified_events_queue:
|
||||
# Safely delete the QTimer object when done
|
||||
self._modified_events_queue[path].deleteLater()
|
||||
del self._modified_events_queue[path]
|
||||
|
||||
def add_path(self, path):
|
||||
"""Adds a directory to be monitored."""
|
||||
if not HAVE_WATCHDOG or self._observer is None:
|
||||
return
|
||||
|
||||
# Normalize and expand path to ensure consistent comparison
|
||||
abs_path = os.path.normpath(os.path.abspath(os.path.expanduser(path)))
|
||||
|
||||
# 1. Check if path is already covered by an existing watch (exact or parent)
|
||||
for watched in self._watched_directories:
|
||||
if abs_path == watched:
|
||||
return
|
||||
parent_prefix = watched if watched.endswith(os.sep) else watched + os.sep
|
||||
if abs_path.startswith(parent_prefix):
|
||||
return # Path is a subdirectory of an already watched directory
|
||||
|
||||
old_monitoring_state = bool(self._watched_directories)
|
||||
|
||||
# 2. Check if this new path covers existing watches (is a parent of them)
|
||||
# If so, consolidate them into this single parent watch
|
||||
child_prefix = abs_path if abs_path.endswith(os.sep) else abs_path + os.sep
|
||||
covered_children = [w for w in self._watched_directories
|
||||
if w.startswith(child_prefix)]
|
||||
|
||||
try:
|
||||
if covered_children:
|
||||
self._observer.unschedule_all()
|
||||
for child in covered_children:
|
||||
self._watched_directories.remove(child)
|
||||
self._watched_directories.add(abs_path)
|
||||
for p in self._watched_directories:
|
||||
self._observer.schedule(self._event_handler, p, recursive=True)
|
||||
print(f"Consolidated monitoring at parent: {abs_path}")
|
||||
else:
|
||||
self._observer.schedule(self._event_handler, abs_path, recursive=True)
|
||||
self._watched_directories.add(abs_path)
|
||||
print(f"Monitoring: {abs_path}")
|
||||
except Exception as e:
|
||||
print(f"Error scheduling watchdog for {abs_path}: {e}")
|
||||
return
|
||||
|
||||
if not old_monitoring_state and self._watched_directories:
|
||||
self.monitoring_status_changed.emit(True)
|
||||
|
||||
def remove_path(self, path):
|
||||
"""Removes a directory from monitoring."""
|
||||
if not HAVE_WATCHDOG or self._observer is None:
|
||||
return
|
||||
abs_path = os.path.normpath(os.path.abspath(os.path.expanduser(path)))
|
||||
if abs_path in self._watched_directories:
|
||||
old_monitoring_state = bool(self._watched_directories)
|
||||
self._observer.unschedule_all() # Simpler to unschedule all and re-add
|
||||
self._watched_directories.remove(abs_path)
|
||||
for p in list(self._watched_directories): # Iterate over a copy
|
||||
self._observer.schedule(self._event_handler, p, recursive=True)
|
||||
print(f"Stopped monitoring: {abs_path}")
|
||||
if HAVE_WATCHDOG and old_monitoring_state and not self._watched_directories:
|
||||
self.monitoring_status_changed.emit(False)
|
||||
|
||||
def clear_paths(self):
|
||||
"""Clears all monitored paths."""
|
||||
if not HAVE_WATCHDOG or not self._observer:
|
||||
return
|
||||
|
||||
old_monitoring_state = bool(self._watched_directories)
|
||||
self._observer.unschedule_all()
|
||||
self._watched_directories.clear()
|
||||
print("Cleared all monitored paths.")
|
||||
if old_monitoring_state:
|
||||
self.monitoring_status_changed.emit(False)
|
||||
|
||||
def stop(self):
|
||||
"""Stops the file system observer."""
|
||||
if HAVE_WATCHDOG and self._observer:
|
||||
self._observer.stop()
|
||||
self._observer.join()
|
||||
|
||||
for timer in self._modified_events_queue.values():
|
||||
timer.stop()
|
||||
|
||||
if HAVE_WATCHDOG:
|
||||
print("FileSystemWatcher stopped.")
|
||||
|
||||
if HAVE_WATCHDOG:
|
||||
class _Handler(FileSystemEventHandler):
|
||||
# Signal to communicate to main thread
|
||||
file_modified_from_thread = Signal(str)
|
||||
"""Custom event handler for watchdog events."""
|
||||
def __init__(self, watcher):
|
||||
super().__init__()
|
||||
self.watcher = watcher
|
||||
|
||||
def on_created(self, event):
|
||||
if event.is_directory:
|
||||
self.watcher.directory_modified.emit(event.src_path)
|
||||
return
|
||||
if self._is_image_file(event.src_path):
|
||||
self.watcher.file_created.emit(event.src_path)
|
||||
|
||||
def on_deleted(self, event):
|
||||
if event.is_directory:
|
||||
self.watcher.directory_modified.emit(event.src_path)
|
||||
return
|
||||
if self._is_image_file(event.src_path):
|
||||
self.watcher.file_deleted.emit(event.src_path)
|
||||
|
||||
def on_moved(self, event):
|
||||
if event.is_directory:
|
||||
self.watcher.directory_moved.emit(event.src_path, event.dest_path)
|
||||
self.watcher.directory_modified.emit(event.src_path)
|
||||
self.watcher.directory_modified.emit(event.dest_path)
|
||||
return
|
||||
self.watcher.file_moved.emit(event.src_path, event.dest_path)
|
||||
|
||||
def on_closed(self, event):
|
||||
if event.is_directory:
|
||||
self.watcher.directory_modified.emit(event.src_path)
|
||||
return
|
||||
if self._is_image_file(event.src_path):
|
||||
self.watcher.file_modified.emit(event.src_path)
|
||||
|
||||
def on_modified(self, event):
|
||||
if event.is_directory:
|
||||
self.watcher.directory_modified.emit(event.src_path)
|
||||
return
|
||||
if self._is_image_file(event.src_path):
|
||||
self.watcher._file_modified_from_handler.emit(event.src_path)
|
||||
|
||||
def _emit_modified(self, path):
|
||||
self.watcher.file_modified.emit(path)
|
||||
if path in self.watcher._modified_events_queue:
|
||||
del self.watcher._modified_events_queue[path]
|
||||
|
||||
def _is_image_file(self, path):
|
||||
return os.path.splitext(path)[1].lower() in IMAGE_EXTENSIONS
|
||||
Reference in New Issue
Block a user