"""DSCons Real-Time "Sync on Save" Hot-Watcher daemon. Monitors local source directories (app, deployment, docker, db, scripts, docs) and immediately syncs modified files to the Cloud Production VM (/opt/dscons) via OpenSSH / SCP. Triggers Docker backend uvicorn hot-reload on the Cloud in real-time (< 0.5s). """ from __future__ import annotations import argparse from pathlib import Path import subprocess import sys import threading import time from typing import Set # Ensure utf-8 output on Windows consoles if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8", errors="replace") if hasattr(sys.stderr, "reconfigure"): sys.stderr.reconfigure(encoding="utf-8", errors="replace") from watchdog.events import FileSystemEvent, FileSystemEventHandler from watchdog.observers import Observer LOCAL_ROOT = Path(__file__).resolve().parent.parent REMOTE_HOST = "dscons" REMOTE_ROOT = "/opt/dscons" SSH_KEY = Path.home() / ".ssh" / "google_compute_engine" WATCH_DIRS = ["app", "deployment", "docker", "db", "scripts", "docs"] ROOT_FILES = { "docker-compose.prod.yml", "docker-compose.yml", "Dockerfile", "pyproject.toml", "README.md", } IGNORE_PARTS = { ".git", ".venv", "__pycache__", ".ruff_cache", ".vscode", ".idea", "qdrant-data", "storage", "logs", "drawings", "cassettes", } IGNORE_EXTENSIONS = { ".pyc", ".pyo", ".pyd", ".swp", ".tmp", ".bak", ".lock", ".log", } def is_ignored(rel_path: Path) -> bool: """Check if relative path matches ignore patterns.""" parts = set(rel_path.parts) if parts & IGNORE_PARTS: return True if rel_path.suffix.lower() in IGNORE_EXTENSIONS: return True if rel_path.name.startswith((".", "~")): return True return False def is_watched(rel_path: Path) -> bool: """Check if relative path is inside watched scope.""" if is_ignored(rel_path): return False first_part = rel_path.parts[0] if rel_path.parts else "" if first_part in WATCH_DIRS: return True if len(rel_path.parts) == 1 and rel_path.name in ROOT_FILES: return True return False class SyncHandler(FileSystemEventHandler): def __init__(self, debounce_sec: float = 0.25): super().__init__() self.debounce_sec = debounce_sec self._pending: dict[str, threading.Timer] = {} self._known_remote_dirs: Set[str] = set() self._lock = threading.Lock() def on_modified(self, event: FileSystemEvent) -> None: if not event.is_directory: self._schedule_sync(event.src_path) def on_created(self, event: FileSystemEvent) -> None: if not event.is_directory: self._schedule_sync(event.src_path) def _schedule_sync(self, file_path_str: str) -> None: file_path = Path(file_path_str) try: rel_path = file_path.relative_to(LOCAL_ROOT) except ValueError: return if not is_watched(rel_path): return posix_rel = rel_path.as_posix() with self._lock: if posix_rel in self._pending: self._pending[posix_rel].cancel() timer = threading.Timer( self.debounce_sec, self._do_sync, args=[file_path, posix_rel] ) self._pending[posix_rel] = timer timer.start() def _ensure_remote_dir(self, parent_rel_posix: str) -> None: if not parent_rel_posix or parent_rel_posix in self._known_remote_dirs: return remote_dir = f"{REMOTE_ROOT}/{parent_rel_posix}" ssh_cmd = [ "ssh", "-i", str(SSH_KEY), "-o", "StrictHostKeyChecking=accept-new", REMOTE_HOST, f"mkdir -p '{remote_dir}'", ] try: subprocess.run(ssh_cmd, check=True, capture_output=True, timeout=5) self._known_remote_dirs.add(parent_rel_posix) except Exception as e: print(f"[MKDIR FAILED] {remote_dir}: {e}") def _do_sync(self, local_path: Path, posix_rel: str) -> None: with self._lock: self._pending.pop(posix_rel, None) if not local_path.exists(): return start_time = time.time() parent_rel = Path(posix_rel).parent.as_posix() if parent_rel and parent_rel != ".": self._ensure_remote_dir(parent_rel) remote_target = f"{REMOTE_HOST}:{REMOTE_ROOT}/{posix_rel}" local_src = f"./{posix_rel}" scp_cmd = [ "scp", "-i", str(SSH_KEY), "-o", "StrictHostKeyChecking=accept-new", local_src, remote_target, ] try: res = subprocess.run( scp_cmd, capture_output=True, text=True, timeout=15, cwd=str(LOCAL_ROOT), ) elapsed = time.time() - start_time now_str = time.strftime("%H:%M:%S") if res.returncode == 0: print(f"[{now_str}] [SYNCED] {posix_rel} -> {remote_target} ({elapsed:.2f}s)", flush=True) else: err_msg = res.stderr.strip() or res.stdout.strip() print(f"[{now_str}] [ERROR] {posix_rel}: {err_msg}", flush=True) except Exception as e: print(f"[{time.strftime('%H:%M:%S')}] [EXCEPTION] {posix_rel}: {e}", flush=True) def initial_full_sync() -> None: """Perform a one-time push of all watched source files.""" print(">> Dang kiem tra va dong bo toan dien (Initial Sync) app/ len Cloud...", flush=True) start_time = time.time() cmd = [ "scp", "-r", "-i", str(SSH_KEY), "-o", "StrictHostKeyChecking=accept-new", "./app", f"{REMOTE_HOST}:{REMOTE_ROOT}/", ] res = subprocess.run(cmd, capture_output=True, text=True, cwd=str(LOCAL_ROOT)) if res.returncode == 0: print(f"[OK] Thu muc app/ da duoc dong bo hoan tat ({time.time() - start_time:.2f}s)", flush=True) else: print(f"[WARN] Canh bao dong bo app/: {res.stderr}", flush=True) def main() -> None: parser = argparse.ArgumentParser(description="DSCons Sync on Save Daemon") parser.add_argument( "--sync-now", action="store_true", help="Run full initial sync of app/ before watching" ) args = parser.parse_args() print("=" * 65) print(" DSCons Real-Time \"Sync on Save\" Hot-Watcher") print(f" Local Source : {LOCAL_ROOT}") print(f" Cloud Server : {REMOTE_HOST} ({REMOTE_ROOT})") print(f" Watched Dirs : {', '.join(WATCH_DIRS)}") print("=" * 65) if args.sync_now: initial_full_sync() handler = SyncHandler() observer = Observer() observer.schedule(handler, str(LOCAL_ROOT), recursive=True) observer.start() print(">> Hot-Watcher dang chay... Moi khi bam Ctrl+S, code se tu day len Cloud!") print(" (Bam Ctrl+C de dung)") try: while True: time.sleep(1) except KeyboardInterrupt: print("\n>> Dung Hot-Watcher.") observer.stop() observer.join() if __name__ == "__main__": main()