Add core modules: manifest, merger, validator

confsync/core/manifest.py

@@ -0,0 +1,207 @@
+"""Manifest builder and management for ConfSync."""
+
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+import yaml
+
+from confsync.models.config_models import (
+    Manifest,
+    ManifestEntry,
+    ConfigFile,
+    ConfigCategory,
+)
+
+
+class ManifestBuilder:
+    """Builds and manages configuration manifests."""
+
+    def __init__(self, ignore_patterns: Optional[List[str]] = None):
+        self.ignore_patterns = ignore_patterns or []
+        self._load_default_ignore_patterns()
+
+    def _load_default_ignore_patterns(self) -> None:
+        """Load default patterns to ignore."""
+        default_patterns = [
+            "*.swp",
+            "*.swo",
+            "*~",
+            ".DS_Store",
+            "*.bak",
+            "*.orig",
+            "*.rej",
+            "confsync",
+            ".git",
+            ".venv",
+            "venv",
+            "__pycache__",
+        ]
+        for pattern in default_patterns:
+            if pattern not in self.ignore_patterns:
+                self.ignore_patterns.append(pattern)
+
+    def should_ignore(self, path: str) -> bool:
+        """Check if a path should be ignored."""
+        path_obj = Path(path)
+        path_str = str(path_obj)
+        for pattern in self.ignore_patterns:
+            if path_obj.match(pattern):
+                return True
+            if pattern.startswith('.'):
+                if path_obj.name == pattern:
+                    return True
+                if pattern in path_obj.parts:
+                    return True
+            elif pattern in path_str.split('/'):
+                return True
+        return False
+
+    def build_from_detected(
+        self,
+        detected_configs: List[ConfigFile],
+        metadata: Optional[Dict] = None
+    ) -> Manifest:
+        """Build a manifest from detected configuration files."""
+        manifest = Manifest(
+            version="1.0.0",
+            created_at=datetime.now(),
+            updated_at=datetime.now(),
+            metadata=metadata or {},
+        )
+
+        for config_file in detected_configs:
+            if self.should_ignore(config_file.path):
+                continue
+
+            entry = ManifestEntry(
+                config_file=config_file,
+                priority=self._get_default_priority(config_file.category),
+                merge_strategy=self._get_default_merge_strategy(config_file),
+            )
+            manifest.add_entry(entry)
+
+        return manifest
+
+    def _get_default_priority(self, category: ConfigCategory) -> int:
+        """Get default priority for a category."""
+        priorities = {
+            ConfigCategory.GIT: 10,
+            ConfigCategory.SHELL: 9,
+            ConfigCategory.TERMINAL: 8,
+            ConfigCategory.EDITOR: 7,
+            ConfigCategory.SSH: 6,
+            ConfigCategory.DOCKER: 5,
+            ConfigCategory.TMUX: 4,
+            ConfigCategory.MISC: 1,
+        }
+        return priorities.get(category, 1)
+
+    def _get_default_merge_strategy(self, config_file: ConfigFile) -> str:
+        """Get default merge strategy for a config file."""
+        ext_strategy_map = {
+            ".json": "three_way",
+            ".yaml": "three_way",
+            ".yml": "three_way",
+            ".toml": "three_way",
+            ".ini": "keep_local",
+            ".cfg": "keep_local",
+            ".conf": "keep_local",
+            ".sh": "keep_common",
+            ".bash": "keep_common",
+            ".zsh": "keep_common",
+            ".fish": "keep_common",
+        }
+        suffix = Path(config_file.path).suffix.lower()
+        return ext_strategy_map.get(suffix, "keep_local")
+
+    def save_manifest(self, manifest: Manifest, path: str) -> None:
+        """Save manifest to a YAML file."""
+        path_obj = Path(path)
+        path_obj.parent.mkdir(parents=True, exist_ok=True)
+        with open(path_obj, 'w', encoding='utf-8') as f:
+            yaml.safe_dump(manifest.to_dict(), f, default_flow_style=False, allow_unicode=True)
+
+    def load_manifest(self, path: str) -> Manifest:
+        """Load manifest from a YAML file."""
+        with open(path, 'r', encoding='utf-8') as f:
+            data = yaml.safe_load(f)
+        if data is None:
+            return Manifest()
+        return Manifest.from_dict(data)
+
+    def merge_manifests(
+        self,
+        base: Manifest,
+        remote: Manifest,
+        local: Manifest,
+        strategy: str = "local_wins"
+    ) -> Manifest:
+        """Merge multiple manifests with the specified strategy."""
+        result = Manifest(
+            version=base.version,
+            created_at=base.created_at,
+            metadata=base.metadata,
+        )
+
+        all_ids = set(base.entries.keys()) | set(remote.entries.keys()) | set(local.entries.keys())
+
+        for entry_id in all_ids:
+            base_entry = base.entries.get(entry_id)
+            remote_entry = remote.entries.get(entry_id)
+            local_entry = local.entries.get(entry_id)
+
+            merged_entry = self._merge_entry(
+                base_entry, remote_entry, local_entry, strategy
+            )
+            if merged_entry:
+                result.entries[entry_id] = merged_entry
+
+        result.updated_at = datetime.now()
+        return result
+
+    def _merge_entry(
+        self,
+        base: Optional[ManifestEntry],
+        remote: Optional[ManifestEntry],
+        local: Optional[ManifestEntry],
+        strategy: str
+    ) -> Optional[ManifestEntry]:
+        """Merge a single manifest entry."""
+        if local and not remote:
+            return local
+
+        if remote and not local:
+            return remote
+
+        if local and remote:
+            if local.config_file.file_hash == remote.config_file.file_hash:
+                return local
+
+            if base and base.config_file.file_hash == local.config_file.file_hash:
+                return remote
+            if base and base.config_file.file_hash == remote.config_file.file_hash:
+                return local
+
+            if strategy == "remote_wins":
+                return remote
+            elif strategy == "local_wins":
+                return local
+
+        return local
+
+    def get_summary(self, manifest: Manifest) -> Dict[str, Any]:
+        """Get a summary of the manifest contents."""
+        summary: Dict[str, Any] = {
+            "total_files": len(manifest.entries),
+            "by_category": {},
+            "by_tool": {},
+        }
+
+        for entry in manifest.entries.values():
+            category = entry.config_file.category.value
+            tool = entry.config_file.tool_name
+
+            summary["by_category"][category] = summary["by_category"].get(category, 0) + 1
+            summary["by_tool"][tool] = summary["by_tool"].get(tool, 0) + 1
+
+        return summary