from __future__ import annotations import logging import os import threading import time from dataclasses import dataclass from pathlib import Path from typing import Any try: import smbclient # type: ignore[import-not-found] except ModuleNotFoundError: smbclient = None from watchdog.events import FileSystemEvent, FileSystemEventHandler, FileSystemMovedEvent from watchdog.observers.polling import PollingObserver from scanner import resolve_runtime_path, scan_results LOGGER = logging.getLogger("scheduler.watcher") @dataclass(frozen=True) class WatchConfig: dut_dir: str ref_dir: str smb_username: str smb_password: str smb_domain: str class _ResultDirEventHandler(FileSystemEventHandler): def __init__(self, on_new_directory): super().__init__() self._on_new_directory = on_new_directory def on_created(self, event: FileSystemEvent) -> None: if event.is_directory: self._on_new_directory(event.src_path) def on_moved(self, event: FileSystemMovedEvent) -> None: if event.is_directory: self._on_new_directory(event.dest_path) class _SmbPollerThread(threading.Thread): """Poll a UNC results directory and trigger a scan when entries change.""" def __init__(self, unc_path: str, on_change, interval_seconds: float = 5.0): super().__init__(daemon=True, name=f"smb-poller-{unc_path}") self._path = unc_path.rstrip("\\") self._on_change = on_change self._interval_seconds = max(1.0, float(interval_seconds)) self._stop_evt = threading.Event() def stop(self) -> None: self._stop_evt.set() def run(self) -> None: prev = self._snapshot() while not self._stop_evt.wait(self._interval_seconds): try: curr = self._snapshot() added = curr - prev removed = prev - curr if added or removed: LOGGER.info( "SMB directory change detected (%s): +%d -%d", self._path, len(added), len(removed), ) self._on_change("smb-directory-change") prev = curr except Exception: LOGGER.exception("SMB poll error for %s", self._path) def _snapshot(self) -> set[str]: if smbclient is None: LOGGER.error("smbclient is not installed; SMB polling is unavailable for %s", self._path) return set() try: return {entry.name for entry in smbclient.scandir(self._path) if entry.is_dir()} except Exception: LOGGER.exception("SMB scandir failed for %s", self._path) return set() class ResultDirectoryWatcher: """Watch DUT/REF result directories and trigger scans when new result folders appear.""" def __init__(self, poll_interval_seconds: float = 2.0, min_scan_interval_seconds: float = 1.0): self._poll_interval_seconds = poll_interval_seconds self._min_scan_interval_seconds = min_scan_interval_seconds self._lock = threading.RLock() self._scan_lock = threading.Lock() self._watchers: list[Any] = [] self._current_config: WatchConfig | None = None self._last_scan_monotonic = 0.0 def configure_from_settings(self, settings: dict[str, Any]) -> None: dut_dir = _normalize_watch_path(settings.get("dutResultDir")) ref_dir = _normalize_watch_path(settings.get("refResultDir")) new_config = WatchConfig( dut_dir=dut_dir, ref_dir=ref_dir, smb_username=str(settings.get("smbUsername") or "").strip(), smb_password=str(settings.get("smbPassword") or ""), smb_domain=str(settings.get("smbDomain") or "").strip(), ) with self._lock: if new_config == self._current_config and self._watchers: return self._stop_locked() self._current_config = new_config if not dut_dir or not ref_dir: LOGGER.info("result watcher disabled: DUT/REF result directories are not both set") return smb_creds = { "username": new_config.smb_username, "password": new_config.smb_password, "domain": new_config.smb_domain, } local_paths: list[Path] = [] smb_paths: list[str] = [] seen: set[str] = set() for result_dir, label in ((dut_dir, "DUT"), (ref_dir, "REF")): is_watchable, reason = _validate_watch_path(result_dir, smb_creds) if not is_watchable: LOGGER.warning( "%s result directory is not watchable, skipping watch: %s (reason: %s)", label, result_dir, reason, ) continue if _is_unc_path(result_dir): unc = _to_unc_path(result_dir) key = unc.lower() if key not in seen: seen.add(key) smb_paths.append(unc) else: p = Path(result_dir) key = os.path.normcase(str(p)) if key not in seen: seen.add(key) local_paths.append(p) watchers: list[Any] = [] if local_paths: handler = _ResultDirEventHandler(self._on_new_directory) observer = PollingObserver(timeout=self._poll_interval_seconds) scheduled_count = 0 scheduled_paths: list[str] = [] for path in local_paths: try: observer.schedule(handler, str(path), recursive=False) scheduled_count += 1 scheduled_paths.append(str(path)) except Exception: LOGGER.exception("failed to watch local result directory: %s", path) if scheduled_count: try: observer.start() watchers.append(observer) for path in scheduled_paths: print(f"[watcher] Watching local directory: {path}") LOGGER.info("watcher successfully watching local path: %s", path) except Exception: LOGGER.exception("failed to start local result observer") try: observer.stop() observer.join(timeout=5) except Exception: pass for unc in smb_paths: registered, reason = _register_smb_session(unc, smb_creds) if not registered: LOGGER.warning("SMB watch skipped for %s (reason: %s)", unc, reason) continue poller = _SmbPollerThread( unc, on_change=lambda reason: self.scan_now(reason=reason), interval_seconds=max(2.0, self._poll_interval_seconds), ) poller.start() watchers.append(poller) LOGGER.info("watcher successfully watching SMB path: %s", unc) print(f"[watcher] Watching SMB directory: {unc}") if not watchers: LOGGER.warning("result watcher not started: no valid result directories to watch") return self._watchers = watchers LOGGER.info("result watcher started for DUT=%s REF=%s", dut_dir, ref_dir) self.scan_now(reason="startup") def _on_new_directory(self, directory_path: str) -> None: LOGGER.info("new result directory detected: %s", directory_path) self.scan_now(reason="directory-created") def scan_now(self, reason: str = "manual") -> None: with self._lock: config = self._current_config if config is None or not config.dut_dir or not config.ref_dir: return now = time.monotonic() if now - self._last_scan_monotonic < self._min_scan_interval_seconds: return self._last_scan_monotonic = now if not self._scan_lock.acquire(blocking=False): return def _run_scan() -> None: credentials = { "username": config.smb_username, "password": config.smb_password, "domain": config.smb_domain, } try: scan_results(config.dut_dir, config.ref_dir, smb_credentials=credentials) LOGGER.info("result scan finished (%s)", reason) except Exception: LOGGER.exception("result scan failed (%s)", reason) finally: self._scan_lock.release() threading.Thread(target=_run_scan, daemon=True).start() def stop(self) -> None: with self._lock: self._stop_locked() def _stop_locked(self) -> None: watchers = self._watchers self._watchers = [] for watcher in watchers: try: watcher.stop() except Exception: pass for watcher in watchers: try: watcher.join(timeout=5) except Exception: pass if watchers: LOGGER.info("result watcher stopped") _WATCHER = ResultDirectoryWatcher() def configure_result_watcher(settings: dict[str, Any]) -> None: _WATCHER.configure_from_settings(settings) def stop_result_watcher() -> None: _WATCHER.stop() def _normalize_watch_path(path_value: Any) -> str: if path_value is None: return "" cleaned = str(path_value).strip() if not cleaned: return "" return str(resolve_runtime_path(cleaned)).strip() def _is_unc_path(path_value: str) -> bool: path = str(path_value) return path.startswith("\\\\") or path.startswith("//") def _to_unc_path(path_value: str) -> str: path = str(path_value).replace("/", "\\") if path.startswith("\\\\"): return path if path.startswith("//"): return "\\\\" + path.lstrip("/\\") return "\\\\" + path.lstrip("/\\") def _validate_watch_path(path_value: str, smb_credentials: dict[str, str]) -> tuple[bool, str]: path = str(path_value).strip() if not path: return False, "path is empty" if _is_unc_path(path): if smbclient is None: return False, "smbclient is not installed" unc = _to_unc_path(path) registered, reason = _register_smb_session(unc, smb_credentials) if not registered: return False, reason try: # Access one directory entry to validate permissions/connectivity. next(iter(smbclient.scandir(unc)), None) except Exception as exc: return False, f"{type(exc).__name__}: {exc}" return True, "ok" if not os.path.exists(path): return False, "path does not exist" if not os.path.isdir(path): return False, "path is not a directory" try: with os.scandir(path): pass except Exception as exc: return False, f"{type(exc).__name__}: {exc}" return True, "ok" def _register_smb_session(unc_path: str, smb_credentials: dict[str, str]) -> tuple[bool, str]: if smbclient is None: return False, "smbclient is not installed" server = unc_path[2:].split("\\", 1)[0] if not server: return False, "invalid UNC path (missing server)" username = str(smb_credentials.get("username") or "").strip() password = str(smb_credentials.get("password") or "") domain = str(smb_credentials.get("domain") or "").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) return True, "ok" except Exception as exc: return False, f"SMB session registration failed for {server}: {type(exc).__name__}: {exc}"