import os import threading from watchdog.events import FileSystemEventHandler from watchdog.observers import Observer from db_py import reset_by_file_id_and_device from scanner import full_scan, parse_deleted_result_dir_name, process_result_dir from sse_py import broadcast _target_observer = None _results_observer = None _scan_timer = None _scan_lock = threading.Lock() def _schedule_full_scan(target_dir, results_dir, delay_seconds=1.0): global _scan_timer with _scan_lock: if _scan_timer: _scan_timer.cancel() def run_scan(): try: full_scan(target_dir, results_dir) broadcast({"type": "update"}) except Exception as exc: print(f"[watcher] fullScan error: {exc}") _scan_timer = threading.Timer(delay_seconds, run_scan) _scan_timer.daemon = True _scan_timer.start() class _TargetHandler(FileSystemEventHandler): def __init__(self, target_dir, results_dir): self.target_dir = os.path.normcase(os.path.abspath(target_dir)) self.results_dir = results_dir def _is_target_test_file(self, path): normalized = os.path.normcase(os.path.abspath(path)) relative = os.path.relpath(normalized, self.target_dir) parts = relative.split(os.sep) if len(parts) != 2: return False filename = parts[-1] return filename.endswith(".ini") and not filename.startswith("GLOBAL") def on_created(self, event): if not event.is_directory and self._is_target_test_file(event.src_path): _schedule_full_scan(self.target_dir, self.results_dir) def on_deleted(self, event): if not event.is_directory and self._is_target_test_file(event.src_path): _schedule_full_scan(self.target_dir, self.results_dir) def on_moved(self, event): if event.is_directory: return # On Windows, create/delete in Explorer can show up as move/rename events. if self._is_target_test_file(event.src_path) or self._is_target_test_file(event.dest_path): _schedule_full_scan(self.target_dir, self.results_dir) class _ResultsHandler(FileSystemEventHandler): def __init__(self, results_dir): self.results_dir = os.path.normcase(os.path.abspath(results_dir)) def _is_direct_child_dir(self, path): parent = os.path.normcase(os.path.abspath(os.path.dirname(path))) return parent == self.results_dir def _is_file_under_result_child(self, path): parent_dir = os.path.normcase(os.path.abspath(os.path.dirname(path))) grandparent = os.path.normcase(os.path.abspath(os.path.dirname(parent_dir))) return grandparent == self.results_dir def _result_child_name_for_file(self, path): return os.path.basename(os.path.dirname(path)) def _result_child_name_for_dir(self, path): return os.path.basename(path) def on_created(self, event): # On Windows, is_directory may be False even for directories (timing issue), # so check _is_direct_child_dir regardless of the flag. if self._is_direct_child_dir(event.src_path): dir_name = self._result_child_name_for_dir(event.src_path) process_result_dir(self.results_dir, dir_name) broadcast({"type": "update"}) return if self._is_file_under_result_child(event.src_path): dir_name = self._result_child_name_for_file(event.src_path) process_result_dir(self.results_dir, dir_name) broadcast({"type": "update"}) def on_deleted(self, event): # On Windows, when a directory is deleted watchdog may report is_directory=False # because os.path.isdir() returns False by the time the event is processed. # Check _is_direct_child_dir first regardless of the is_directory flag. if self._is_direct_child_dir(event.src_path): dir_name = self._result_child_name_for_dir(event.src_path) test_id, device = parse_deleted_result_dir_name(dir_name) if test_id: reset_by_file_id_and_device(test_id, device) broadcast({"type": "update"}) return if self._is_file_under_result_child(event.src_path): dir_name = self._result_child_name_for_file(event.src_path) process_result_dir(self.results_dir, dir_name) broadcast({"type": "update"}) def on_moved(self, event): src_in_root = self._is_direct_child_dir(event.src_path) dst_in_root = self._is_direct_child_dir(event.dest_path) # Handle directory-level moves regardless of is_directory flag (Windows timing issue). if src_in_root or dst_in_root: # Result directory moved out (includes Recycle Bin delete on Windows). if src_in_root and not dst_in_root: old_name = self._result_child_name_for_dir(event.src_path) test_id, device = parse_deleted_result_dir_name(old_name) if test_id: reset_by_file_id_and_device(test_id, device) broadcast({"type": "update"}) return # Result directory moved in. if dst_in_root and not src_in_root: new_name = self._result_child_name_for_dir(event.dest_path) process_result_dir(self.results_dir, new_name) broadcast({"type": "update"}) return # Result directory renamed within root. if src_in_root and dst_in_root: old_name = self._result_child_name_for_dir(event.src_path) new_name = self._result_child_name_for_dir(event.dest_path) test_id, device = parse_deleted_result_dir_name(old_name) if test_id: reset_by_file_id_and_device(test_id, device) process_result_dir(self.results_dir, new_name) broadcast({"type": "update"}) return src_file_in_result = self._is_file_under_result_child(event.src_path) dst_file_in_result = self._is_file_under_result_child(event.dest_path) if src_file_in_result: src_name = self._result_child_name_for_file(event.src_path) process_result_dir(self.results_dir, src_name) if dst_file_in_result: dst_name = self._result_child_name_for_file(event.dest_path) if not src_file_in_result or src_name != dst_name: process_result_dir(self.results_dir, dst_name) if src_file_in_result or dst_file_in_result: broadcast({"type": "update"}) def stop_watching(): global _target_observer, _results_observer, _scan_timer if _scan_timer: _scan_timer.cancel() _scan_timer = None if _target_observer: _target_observer.stop() _target_observer.join(timeout=2) _target_observer = None if _results_observer: _results_observer.stop() _results_observer.join(timeout=2) _results_observer = None def start_watching(target_dir, results_dir): global _target_observer, _results_observer stop_watching() if not target_dir or not results_dir: print("[watcher] target_dir/results_dir not configured; watcher disabled.") return if not os.path.isdir(target_dir): print(f"[watcher] target_dir does not exist inside runtime: {target_dir}") print("[watcher] watcher disabled until valid paths are configured.") return if not os.path.isdir(results_dir): print(f"[watcher] results_dir does not exist inside runtime: {results_dir}") print("[watcher] watcher disabled until valid paths are configured.") return try: _target_observer = Observer() _target_observer.schedule(_TargetHandler(target_dir, results_dir), target_dir, recursive=True) _target_observer.daemon = True _target_observer.start() _results_observer = Observer() _results_observer.schedule(_ResultsHandler(results_dir), results_dir, recursive=True) _results_observer.daemon = True _results_observer.start() except Exception as exc: print(f"[watcher] failed to start watchers: {exc}") stop_watching()