import os import threading import smbclient from watchdog.events import FileSystemEventHandler from watchdog.observers.polling import PollingObserver from pathlib import Path from db_py import ( mark_tests_completed, reset_by_file_id_and_device, update_all_p2p_coe_pairs_sql, update_all_p3p_pairs_sql, ) from scanner import process_result_dir, resolve_runtime_path from parser import parse_deleted_result_dir_name _WATCHERS: list = [] _WATCHERS_LOCK = threading.RLock() class ResultsDirEventHandler(FileSystemEventHandler): def __init__(self, results_root): super().__init__() self.results_root = Path(results_root) def _is_direct_child(self, path_value): return Path(path_value).parent == self.results_root def _mark_completed(self, dir_path): if not self._is_direct_child(dir_path): return dir_name = Path(dir_path).name completion = process_result_dir(self.results_root, dir_name) if not completion: return mark_tests_completed([completion]) coe_pairs = update_all_p2p_coe_pairs_sql() p3p_pairs = update_all_p3p_pairs_sql() print( "[watcher] Marked completed from new result dir " f"{dir_name} (coe_pair={coe_pairs}, p3p_pair={p3p_pairs})" ) def _mark_completed_from_file(self, file_path): # New measurement/log files arriving after directory creation should update that single test. p = Path(file_path) filename = p.name.lower() is_duration_log = filename.endswith(".log") or filename.endswith(".txt") or "log" in filename if filename != "measurement.db" and not is_duration_log: return self._mark_completed(p.parent) def _clear_completed(self, dir_path): if not self._is_direct_child(dir_path): return dir_name = Path(dir_path).name test_id, device = parse_deleted_result_dir_name(dir_name) if not test_id: return reset_by_file_id_and_device(test_id, device) coe_pairs = update_all_p2p_coe_pairs_sql() p3p_pairs = update_all_p3p_pairs_sql() print( "[watcher] Cleared completed from removed result dir " f"{dir_name} (coe_pair={coe_pairs}, p3p_pair={p3p_pairs})" ) def on_created(self, event): if event.is_directory: self._mark_completed(event.src_path) return self._mark_completed_from_file(event.src_path) def on_deleted(self, event): if event.is_directory: self._clear_completed(event.src_path) def on_moved(self, event): if event.is_directory: self._clear_completed(event.src_path) self._mark_completed(event.dest_path) return self._mark_completed_from_file(event.dest_path) class _SmbPollerThread(threading.Thread): """Polls a UNC results directory via smbclient and updates the DB on directory changes.""" _INTERVAL = 5.0 def __init__(self, unc_path): super().__init__(daemon=True, name=f"smb-poller-{unc_path}") self._path = unc_path.rstrip("\\") self._stop_evt = threading.Event() def stop(self): self._stop_evt.set() def run(self): prev = self._snapshot() while not self._stop_evt.wait(self._INTERVAL): try: curr = self._snapshot() for name in curr - prev: completion = process_result_dir(self._path, name) if completion: mark_tests_completed([completion]) coe = update_all_p2p_coe_pairs_sql() p3p = update_all_p3p_pairs_sql() print(f"[watcher] SMB: marked completed {name} (coe={coe}, p3p={p3p})") for name in prev - curr: test_id, device = parse_deleted_result_dir_name(name) if test_id: reset_by_file_id_and_device(test_id, device) coe = update_all_p2p_coe_pairs_sql() p3p = update_all_p3p_pairs_sql() print(f"[watcher] SMB: cleared completed {name} (coe={coe}, p3p={p3p})") prev = curr except Exception as exc: print(f"[watcher] SMB poll error {self._path}: {exc}") def _snapshot(self): try: return {e.name for e in smbclient.scandir(self._path) if e.is_dir()} except Exception as exc: print(f"[watcher] SMB scandir error {self._path}: {exc}") return set() def _register_smb_session(unc_path): server = unc_path[2:].split("\\", 1)[0] if not server: return username = os.getenv("SMB_USERNAME", "").strip() password = os.getenv("SMB_PASSWORD", "") domain = os.getenv("SMB_DOMAIN", "").strip() if username and domain and "\\" not in username and "@" not in username: username = f"{domain}\\{username}" try: if username: smbclient.register_session(server, username=username, password=password) else: smbclient.register_session(server) except Exception as exc: print(f"[watcher] SMB session registration failed for {server}: {exc}") def stop_results_watchers(): global _WATCHERS with _WATCHERS_LOCK: if not _WATCHERS: return for w in _WATCHERS: try: w.stop() except Exception: pass for w in _WATCHERS: try: w.join(timeout=5) except Exception: pass _WATCHERS = [] print("[watcher] Stopped results directory watcher(s)") def start_results_watchers(results_dir, results_dir_ref): global _WATCHERS with _WATCHERS_LOCK: stop_results_watchers() local_paths = [] smb_paths = [] seen = set() for raw_dir in (results_dir, results_dir_ref): if not raw_dir: continue path = resolve_runtime_path(raw_dir) if not path: continue path = str(path) if path.startswith("\\\\") or path.startswith("//"): unc = "\\\\" + path.lstrip("/\\").replace("/", "\\") key = unc.lower() if key not in seen: seen.add(key) smb_paths.append(unc) else: p = Path(path) key = os.path.normcase(str(p)) if key not in seen: seen.add(key) local_paths.append(p) if not local_paths and not smb_paths: print("[watcher] No results directories configured, watcher not started") return watchers = [] if local_paths: observer = PollingObserver(timeout=1.0) scheduled = 0 for p in local_paths: try: observer.schedule(ResultsDirEventHandler(p), str(p), recursive=False) scheduled += 1 print(f"[watcher] Watching local directory: {p}") except Exception as exc: print(f"[watcher] Skipping local watch path {p}: {exc}") if scheduled: try: observer.start() watchers.append(observer) except Exception as exc: print(f"[watcher] Failed to start local watcher: {exc}") try: observer.stop() observer.join(timeout=5) except Exception: pass for unc in smb_paths: _register_smb_session(unc) poller = _SmbPollerThread(unc) poller.start() watchers.append(poller) print(f"[watcher] Watching SMB directory: {unc}") if not watchers: print("[watcher] No valid watch paths available, watcher not started") return _WATCHERS = watchers print(f"[watcher] Results directory watcher(s) started ({len(watchers)})")