First version

src/DebouncedHandler.py

@@ -0,0 +1,39 @@
+import threading
+import logging
+
+from watchdog.events import PatternMatchingEventHandler
+
+from config import *
+from helpers import export_single_zone
+
+# Internal state for debounce
+debounce_timer = None
+debounce_lock = threading.Lock()
+
+def run_export(trigger_path):
+    global debounce_timer
+    with debounce_lock:
+        debounce_timer = None
+    try:
+        export_single_zone(trigger_path)
+    except Exception:
+        logging.exception("Export run failed.")
+
+def schedule_export(trigger_path):
+    global debounce_timer
+    with debounce_lock:
+        if debounce_timer is not None:
+            debounce_timer.cancel()
+        debounce_timer = threading.Timer(DEBOUNCE_SECONDS, run_export, args=(trigger_path,))
+        debounce_timer.daemon = True
+        debounce_timer.start()
+        logging.debug("Debounce timer started/reset (%.1fs)", DEBOUNCE_SECONDS)
+
+class DebouncedHandler(PatternMatchingEventHandler):
+    def __init__(self, patterns=None, ignore_patterns=None, ignore_directories=False, case_sensitive=True):
+        super().__init__(patterns=patterns or ["*"], ignore_patterns=ignore_patterns or [], ignore_directories=ignore_directories, case_sensitive=case_sensitive)
+
+    def on_any_event(self, event):
+        # When any matching event occurs, start/reset debounce timer
+        logging.debug(f"Filesystem event: {event.event_type} on {event.src_path}")
+        schedule_export(event.src_path)