scheduler/backend/watcher.py

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}"