diff --git a/dev_env_sync/managers/dotfile.py b/dev_env_sync/managers/dotfile.py new file mode 100644 index 0000000..95a481f --- /dev/null +++ b/dev_env_sync/managers/dotfile.py @@ -0,0 +1,315 @@ +"""Dotfiles management and symlink handling.""" + +import fnmatch +import os +from pathlib import Path +from typing import Dict, List, Optional, Set +from dataclasses import dataclass + +from dev_env_sync.core.config import DotfileConfig, ConfigSchema +from dev_env_sync.utils.file_ops import FileOps, path_expand +from dev_env_sync.utils.logging import get_logger +from dev_env_sync.core.platform import PlatformDetector + + +class SymlinkError(Exception): + """Raised when a symlink operation fails.""" + pass + + +@dataclass +class SyncResult: + """Result of a dotfile sync operation.""" + success: bool + source: Path + target: Path + action: str + message: str + backup_path: Optional[Path] = None + + +class SymlinkHandler: + """Handles creation and management of symbolic links.""" + + def __init__(self, dry_run: bool = False, logger=None): + self.dry_run = dry_run + self.logger = logger or get_logger(__name__) + self.file_ops = FileOps(dry_run=dry_run, logger=logger) + + def create( + self, + source: Path, + target: Path, + backup_existing: bool = True, + backup_dir: Optional[Path] = None, + ) -> SyncResult: + """Create a symlink from source to target.""" + source = path_expand(str(source)) + target = path_expand(str(target)) + + if not source.exists(): + return SyncResult( + success=False, + source=source, + target=target, + action="create_symlink", + message=f"Source does not exist: {source}", + ) + + action = "create" + backup_path = None + + if target.is_symlink(): + existing_source = target.resolve() + if existing_source == source.resolve(): + return SyncResult( + success=True, + source=source, + target=target, + action="skip", + message="Symlink already points to correct source", + ) + else: + target.unlink() + action = "replace_symlink" + elif target.exists(): + if backup_existing: + if backup_dir is None: + backup_dir = target.parent + backup_path = self._backup(target, backup_dir) + action = "replace_with_backup" + else: + target.unlink() + action = "replace" + + success = self.file_ops.create_symlink( + source=source, + target=target, + force=True, + backup=False, + ) + + if success: + message = f"Created symlink: {target} -> {source}" + if action != "create": + message = f"{action.capitalize()}: {message}" + return SyncResult( + success=True, + source=source, + target=target, + action=action, + message=message, + backup_path=backup_path, + ) + else: + return SyncResult( + success=False, + source=source, + target=target, + action="create_symlink", + message=f"Failed to create symlink: {target}", + ) + + def remove(self, target: Path, backup: bool = False, backup_dir: Optional[Path] = None) -> SyncResult: + """Remove a symlink or file.""" + target = path_expand(str(target)) + + if not target.exists(): + return SyncResult( + success=False, + source=Path(""), + target=target, + action="remove", + message=f"Target does not exist: {target}", + ) + + backup_path = None + if backup and target.is_file() and not target.is_symlink(): + if backup_dir is None: + backup_dir = target.parent + backup_path = self._backup(target, backup_dir) + + success = self.file_ops.remove_file(target, force=True) + + if success: + return SyncResult( + success=True, + source=Path(""), + target=target, + action="remove", + message=f"Removed: {target}", + backup_path=backup_path, + ) + else: + return SyncResult( + success=False, + source=Path(""), + target=target, + action="remove", + message=f"Failed to remove: {target}", + ) + + def _backup(self, path: Path, backup_dir: Path) -> Path: + """Create a backup of a file.""" + backup_dir = path_expand(str(backup_dir)) + backup_dir.mkdir(parents=True, exist_ok=True) + + timestamp = int(path.stat().st_mtime) + backup_name = f"{path.name}.{timestamp}" + backup_path = backup_dir / backup_name + + self.file_ops.copy_file(path, backup_path, preserve_metadata=True) + return backup_path + + def check_status(self, source: Path, target: Path) -> str: + """Check the status of a symlink.""" + source = path_expand(str(source)) + target = path_expand(str(target)) + + if not target.exists(): + return "missing" + + if target.is_symlink(): + actual_target = target.resolve() + source_resolved = source.resolve() + if actual_target == source_resolved: + return "ok" + else: + return "broken" + else: + return "file_exists" + + def verify(self, source: Path, target: Path) -> bool: + """Verify that a symlink is correctly set up.""" + status = self.check_status(source, target) + return status == "ok" + + +class DotfileManager: + """Manages dotfiles synchronization.""" + + def __init__( + self, + config: ConfigSchema, + dry_run: bool = False, + backup_dir: Optional[Path] = None, + logger=None, + ): + self.config = config + self.dry_run = dry_run + self.backup_dir = path_expand(str(backup_dir)) if backup_dir else None + self.logger = logger or get_logger(__name__) + self.symlink_handler = SymlinkHandler(dry_run=dry_run, logger=logger) + self.file_ops = FileOps(dry_run=dry_run, logger=logger) + self.results: List[SyncResult] = [] + + def sync(self, dotfiles: Optional[Set[str]] = None) -> List[SyncResult]: + """Sync dotfiles based on configuration.""" + self.results = [] + + if self.backup_dir: + self.file_ops.create_directory(self.backup_dir, exist_ok=True) + + dotfiles_to_sync = dotfiles or set(self.config.dotfiles.keys()) + + for name in dotfiles_to_sync: + if name not in self.config.dotfiles: + self.logger.warning(f"Dotfile '{name}' not found in configuration") + continue + + config = self.config.dotfiles[name] + result = self._sync_dotfile(name, config) + self.results.append(result) + + return self.results + + def _sync_dotfile(self, name: str, config: DotfileConfig) -> SyncResult: + """Sync a single dotfile.""" + source = Path(config.source) + target = Path(config.target) + + if not source.exists(): + self.logger.error(f"Source dotfile not found: {source}") + return SyncResult( + success=False, + source=source, + target=target, + action="sync", + message=f"Source not found: {source}", + ) + + if not self._should_ignore(name, config.ignore): + return self.symlink_handler.create( + source=source, + target=target, + backup_existing=config.backup, + backup_dir=self.backup_dir, + ) + else: + return SyncResult( + success=True, + source=source, + target=target, + action="skip", + message=f"Ignored: {name}", + ) + + def _should_ignore(self, name: str, ignore_patterns: Optional[List[str]]) -> bool: + """Check if a dotfile should be ignored.""" + if not ignore_patterns: + return False + + for pattern in ignore_patterns: + if fnmatch.fnmatch(name, pattern): + return True + return False + + def verify(self) -> Dict[str, bool]: + """Verify all dotfiles are correctly linked.""" + status = {} + for name, config in self.config.dotfiles.items(): + source = Path(config.source) + target = Path(config.target) + status[name] = self.symlink_handler.verify(source, target) + return status + + def sync_all(self) -> List[SyncResult]: + """Sync all dotfiles in the configuration.""" + return self.sync() + + def get_summary(self) -> Dict[str, int]: + """Get a summary of sync operations.""" + summary = { + "total": len(self.results), + "success": 0, + "failed": 0, + "skipped": 0, + } + + for result in self.results: + if result.success: + if result.action == "skip": + summary["skipped"] += 1 + else: + summary["success"] += 1 + else: + summary["failed"] += 1 + + return summary + + def print_summary(self): + """Print a summary of sync operations.""" + summary = self.get_summary() + + self.logger.info(f"Sync Summary:") + self.logger.info(f" Total: {summary['total']}") + self.logger.info(f" Success: {summary['success']}") + self.logger.info(f" Failed: {summary['failed']}") + self.logger.info(f" Skipped: {summary['skipped']}") + + for result in self.results: + if result.success: + status = "OK" if result.action != "skip" else "SKIP" + self.logger.info(f" [{status}] {result.target} -> {result.source}") + else: + self.logger.error(f" [FAIL] {result.message}")