""" Directory Sync Engine - Unidirectional Mirror Synchronization Mirrors source directory to destination using mtime+size comparison """ import os import shutil import fnmatch from pathlib import Path from typing import Optional, Callable, Dict, Any, List from datetime import datetime from loguru import logger class DirectorySync: """ Unidirectional directory synchronization engine Mirrors source -> destination: - Copies new/modified files (based on mtime + size) - Deletes files from destination that don't exist in source - Preserves directory structure - Handles network paths (e.g., Z:\) """ def __init__(self, source: str, destination: str, exclude_patterns: List[str] = None, protect_excluded: bool = False): """ Initialize sync engine Args: source: Source directory path (can be network path) destination: Destination directory path (can be network path) exclude_patterns: List of patterns to exclude (e.g., ['venv/', '*.log', '__pycache__/']) protect_excluded: If True, don't delete excluded files in destination (for deploy scenarios) If False, delete excluded files from destination (for clean mirror) """ self.source = Path(source) self.destination = Path(destination) self.exclude_patterns = exclude_patterns or [] self.protect_excluded = protect_excluded # Validate paths if not self.source.exists(): raise ValueError(f"Source path does not exist: {source}") if not self.source.is_dir(): raise ValueError(f"Source path is not a directory: {source}") # Create destination if doesn't exist self.destination.mkdir(parents=True, exist_ok=True) if self.exclude_patterns: logger.info(f"DirectorySync initialized: {source} -> {destination} (excluding {len(self.exclude_patterns)} patterns)") else: logger.info(f"DirectorySync initialized: {source} -> {destination}") def sync(self, progress_callback: Optional[Callable] = None) -> Dict[str, Any]: """ Perform synchronization Args: progress_callback: Optional callback for progress updates Called with dict: {'phase': str, 'current': int, 'total': int, 'file': str} Returns: Dict with statistics: { 'files_scanned': int, 'files_copied': int, 'files_deleted': int, 'bytes_total': int, 'errors': List[str] } """ stats = { 'files_scanned': 0, 'files_copied': 0, 'files_deleted': 0, 'bytes_total': 0, 'errors': [] } try: # Phase 1: Build source file map if progress_callback: progress_callback({'phase': 'scanning', 'current': 0, 'total': 0, 'file': ''}) source_files = self._scan_directory(self.source) stats['files_scanned'] = len(source_files) logger.info(f"Scanned {len(source_files)} files in source") # Phase 2: Sync files (copy new/modified) if progress_callback: progress_callback({'phase': 'syncing', 'current': 0, 'total': len(source_files), 'file': ''}) for idx, (rel_path, source_file) in enumerate(source_files.items()): dest_file = self.destination / rel_path try: if self._needs_sync(source_file, dest_file): self._copy_file(source_file, dest_file) stats['files_copied'] += 1 stats['bytes_total'] += source_file.stat().st_size if progress_callback: progress_callback({ 'phase': 'syncing', 'current': idx + 1, 'total': len(source_files), 'file': str(rel_path) }) except Exception as e: error_msg = f"Error syncing {rel_path}: {e}" logger.error(error_msg) stats['errors'].append(error_msg) # Phase 3: Delete files not in source if progress_callback: progress_callback({'phase': 'cleaning', 'current': 0, 'total': 0, 'file': ''}) deleted_count = self._cleanup_destination(source_files) stats['files_deleted'] = deleted_count logger.info(f"Sync completed: {stats['files_copied']} copied, {stats['files_deleted']} deleted") except Exception as e: error_msg = f"Critical sync error: {e}" logger.exception(error_msg) stats['errors'].append(error_msg) raise return stats def _should_exclude(self, rel_path: Path) -> bool: """ Check if path matches any exclusion pattern Args: rel_path: Relative path to check Returns: True if path should be excluded """ if not self.exclude_patterns: return False path_str = str(rel_path).replace('\\', '/') path_parts = rel_path.parts for pattern in self.exclude_patterns: pattern = pattern.strip() if not pattern or pattern.startswith('#'): continue # Directory pattern (ends with /) if pattern.endswith('/'): dir_pattern = pattern.rstrip('/') # Check if any part of the path matches the directory pattern for part in path_parts: if fnmatch.fnmatch(part, dir_pattern): return True else: # File pattern - check filename if fnmatch.fnmatch(rel_path.name, pattern): return True # Also check full relative path for patterns like "subdir/*.txt" if fnmatch.fnmatch(path_str, pattern): return True return False def _scan_directory(self, directory: Path, apply_exclusions: bool = True) -> Dict[Path, Path]: """ Scan directory and build file map Args: directory: Directory to scan apply_exclusions: Whether to apply exclusion patterns (default True) Returns: Dict mapping relative paths to absolute paths """ file_map = {} excluded_count = 0 try: for item in directory.rglob('*'): if item.is_file(): try: rel_path = item.relative_to(directory) # Check exclusions if apply_exclusions and self._should_exclude(rel_path): excluded_count += 1 continue file_map[rel_path] = item except (OSError, PermissionError) as e: logger.warning(f"Cannot access {item}: {e}") continue except Exception as e: logger.error(f"Error scanning directory {directory}: {e}") raise if excluded_count > 0: logger.info(f"Excluded {excluded_count} files matching exclusion patterns") return file_map def _needs_sync(self, source_file: Path, dest_file: Path) -> bool: """ Check if file needs to be synced using size comparison only Note: mtime comparison disabled because WebDAV/network drives don't preserve timestamps correctly with shutil.copy2() Args: source_file: Source file path dest_file: Destination file path Returns: True if file needs to be copied """ # If destination doesn't exist, needs sync if not dest_file.exists(): return True try: source_stat = source_file.stat() dest_stat = dest_file.stat() # Compare size only (mtime not reliable on WebDAV) if source_stat.st_size != dest_stat.st_size: logger.debug(f"Size mismatch: {source_file.name}") return True # Files have same size - consider them synced return False except (OSError, PermissionError) as e: logger.warning(f"Cannot compare {source_file}: {e}") # If we can't compare, assume it needs sync return True def _copy_file(self, source_file: Path, dest_file: Path): """ Copy file from source to destination, preserving metadata Args: source_file: Source file path dest_file: Destination file path """ try: # Create parent directory if needed dest_file.parent.mkdir(parents=True, exist_ok=True) # Copy file with metadata (preserves mtime) shutil.copy2(source_file, dest_file) logger.debug(f"Copied: {source_file.name}") except (OSError, PermissionError, shutil.Error) as e: logger.error(f"Failed to copy {source_file} to {dest_file}: {e}") raise def _cleanup_destination(self, source_files: Dict[Path, Path]) -> int: """ Delete files from destination that don't exist in source Behavior depends on protect_excluded flag: - protect_excluded=True: Don't delete files matching exclusion patterns (for deploy) - protect_excluded=False: Delete all files not in source, including excluded (clean mirror) Args: source_files: Dict of relative paths from source Returns: Number of files deleted """ deleted_count = 0 skipped_excluded = 0 try: # Get all files in destination (without applying exclusions) dest_files = self._scan_directory(self.destination, apply_exclusions=False) # Find files to delete (in dest but not in source) for rel_path in dest_files: if rel_path not in source_files: # Only skip excluded files if protect_excluded is enabled if self.protect_excluded and self._should_exclude(rel_path): skipped_excluded += 1 continue dest_file = self.destination / rel_path try: # Try normal delete first dest_file.unlink() logger.debug(f"Deleted: {rel_path}") deleted_count += 1 except PermissionError: # Try removing read-only attribute and retry try: os.chmod(dest_file, 0o777) dest_file.unlink() logger.debug(f"Deleted (forced): {rel_path}") deleted_count += 1 except (OSError, PermissionError) as e: logger.error(f"Failed to delete {dest_file}: {e}") except OSError as e: logger.error(f"Failed to delete {dest_file}: {e}") if skipped_excluded > 0: logger.info(f"Skipped deletion of {skipped_excluded} excluded files in destination (protect_excluded=True)") # Clean up empty directories self._cleanup_empty_dirs(self.destination) except Exception as e: logger.error(f"Error during cleanup: {e}") return deleted_count def _cleanup_empty_dirs(self, directory: Path): """ Remove empty directories from destination Args: directory: Root directory to clean """ try: # Walk from bottom up to handle nested empty dirs for dirpath, dirnames, filenames in os.walk(str(directory), topdown=False): dir_path = Path(dirpath) # Skip the root directory if dir_path == directory: continue # Check if directory is empty try: if not any(dir_path.iterdir()): dir_path.rmdir() logger.debug(f"Removed empty directory: {dir_path}") except (OSError, PermissionError): pass except Exception as e: logger.warning(f"Error cleaning empty directories: {e}")