From bf9910452e53b988968ad52872c42754e26fe3a2 Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Wed, 4 Feb 2026 20:03:05 +0000 Subject: [PATCH] Add core modules: manifest, merger, validator --- confsync/core/manifest.py | 207 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 207 insertions(+) create mode 100644 confsync/core/manifest.py diff --git a/confsync/core/manifest.py b/confsync/core/manifest.py new file mode 100644 index 0000000..7f4df78 --- /dev/null +++ b/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