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