316 lines
10 KiB
Python
316 lines
10 KiB
Python
"""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}")
|