From 8b9f9c60885f6c09fbca6ae396c8bbeebee4f99d Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Wed, 4 Feb 2026 20:01:29 +0000 Subject: [PATCH] Add config models --- confsync/models/config_models.py | 347 +++++++++++++++++++++++++++++++ 1 file changed, 347 insertions(+) create mode 100644 confsync/models/config_models.py diff --git a/confsync/models/config_models.py b/confsync/models/config_models.py new file mode 100644 index 0000000..26a7d5e --- /dev/null +++ b/confsync/models/config_models.py @@ -0,0 +1,347 @@ +"""Data models for configuration files and manifests.""" + +import hashlib +import uuid +from dataclasses import dataclass, field +from datetime import datetime +from enum import Enum +from typing import Optional, Dict, List, Any + + +class ConfigCategory(str, Enum): + """Categories of configuration files.""" + + EDITOR = "editor" + TERMINAL = "terminal" + SHELL = "shell" + GIT = "git" + SSH = "ssh" + DOCKER = "docker" + TMUX = "tmux" + MISC = "misc" + + +@dataclass +class ConfigFile: + """Represents a configuration file.""" + + path: str + name: str + category: ConfigCategory + tool_name: str + content: str = "" + file_hash: str = field(default_factory=lambda: calculate_file_hash("")) + size: int = 0 + created_at: datetime = field(default_factory=datetime.now) + modified_at: datetime = field(default_factory=datetime.now) + metadata: Dict[str, Any] = field(default_factory=dict) + tags: List[str] = field(default_factory=list) + id: str = field(default_factory=lambda: str(uuid.uuid4())) + + def __post_init__(self): + if not self.file_hash: + self.file_hash = calculate_file_hash(self.content) + + def update_content(self, new_content: str) -> None: + """Update the file content and recalculate hash.""" + self.content = new_content + self.file_hash = calculate_file_hash(new_content) + self.size = len(new_content.encode('utf-8')) + self.modified_at = datetime.now() + + def to_dict(self) -> Dict[str, Any]: + """Convert to dictionary for serialization.""" + return { + "id": self.id, + "path": self.path, + "name": self.name, + "category": self.category.value, + "tool_name": self.tool_name, + "content": self.content, + "file_hash": self.file_hash, + "size": self.size, + "created_at": self.created_at.isoformat(), + "modified_at": self.modified_at.isoformat(), + "metadata": self.metadata, + "tags": self.tags, + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "ConfigFile": + """Create from dictionary.""" + data["category"] = ConfigCategory(data["category"]) + data["created_at"] = datetime.fromisoformat(data["created_at"]) + data["modified_at"] = datetime.fromisoformat(data["modified_at"]) + return cls(**data) + + +def calculate_file_hash(content: str) -> str: + """Calculate SHA256 hash of content.""" + return hashlib.sha256(content.encode('utf-8')).hexdigest() + + +@dataclass +class ManifestEntry: + """Entry in the configuration manifest.""" + + config_file: ConfigFile + priority: int = 0 + merge_strategy: str = "keep_local" + ignore_patterns: List[str] = field(default_factory=list) + custom_rules: Dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + """Convert to dictionary.""" + return { + "config_file": self.config_file.to_dict(), + "priority": self.priority, + "merge_strategy": self.merge_strategy, + "ignore_patterns": self.ignore_patterns, + "custom_rules": self.custom_rules, + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "ManifestEntry": + """Create from dictionary.""" + return cls( + config_file=ConfigFile.from_dict(data["config_file"]), + priority=data.get("priority", 0), + merge_strategy=data.get("merge_strategy", "keep_local"), + ignore_patterns=data.get("ignore_patterns", []), + custom_rules=data.get("custom_rules", {}), + ) + + +@dataclass +class Manifest: + """Unified manifest of configuration files.""" + + version: str = "1.0.0" + created_at: datetime = field(default_factory=datetime.now) + updated_at: datetime = field(default_factory=datetime.now) + entries: Dict[str, ManifestEntry] = field(default_factory=dict) + metadata: Dict[str, Any] = field(default_factory=dict) + + def add_entry(self, entry: ManifestEntry) -> None: + """Add an entry to the manifest.""" + self.entries[entry.config_file.id] = entry + self.updated_at = datetime.now() + + def remove_entry(self, config_id: str) -> bool: + """Remove an entry from the manifest.""" + if config_id in self.entries: + del self.entries[config_id] + self.updated_at = datetime.now() + return True + return False + + def get_by_category(self, category: ConfigCategory) -> List[ManifestEntry]: + """Get all entries in a category.""" + return [ + entry for entry in self.entries.values() + if entry.config_file.category == category + ] + + def get_by_tool(self, tool_name: str) -> List[ManifestEntry]: + """Get all entries for a specific tool.""" + return [ + entry for entry in self.entries.values() + if entry.config_file.tool_name.lower() == tool_name.lower() + ] + + def get_changed_files(self, other_manifest: "Manifest") -> Dict[str, ManifestEntry]: + """Get files that have changed between manifests.""" + changed = {} + for entry in self.entries.values(): + other_entry = other_manifest.entries.get(entry.config_file.id) + if other_entry is None: + changed[entry.config_file.id] = entry + elif entry.config_file.file_hash != other_entry.config_file.file_hash: + changed[entry.config_file.id] = entry + return changed + + def to_dict(self) -> Dict[str, Any]: + """Convert to dictionary for YAML serialization.""" + return { + "version": self.version, + "created_at": self.created_at.isoformat(), + "updated_at": self.updated_at.isoformat(), + "entries": {k: v.to_dict() for k, v in self.entries.items()}, + "metadata": self.metadata, + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "Manifest": + """Create from dictionary.""" + manifest = cls( + version=data.get("version", "1.0.0"), + created_at=datetime.fromisoformat(data["created_at"]), + updated_at=datetime.fromisoformat(data["updated_at"]), + metadata=data.get("metadata", {}), + ) + for entry_id, entry_data in data.get("entries", {}).items(): + manifest.entries[entry_id] = ManifestEntry.from_dict(entry_data) + return manifest + + +class SyncOperation(str, Enum): + """Types of sync operations.""" + + PUSH = "push" + PULL = "pull" + SYNC = "sync" + STATUS = "status" + + +@dataclass +class SyncMetadata: + """Metadata for sync operations.""" + + operation: SyncOperation + source: str + destination: str + timestamp: datetime = field(default_factory=datetime.now) + committed_files: List[str] = field(default_factory=list) + conflicts: List[str] = field(default_factory=list) + merged_files: List[str] = field(default_factory=list) + status: str = "pending" + error_message: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + """Convert to dictionary.""" + return { + "operation": self.operation.value, + "source": self.source, + "destination": self.destination, + "timestamp": self.timestamp.isoformat(), + "committed_files": self.committed_files, + "conflicts": self.conflicts, + "merged_files": self.merged_files, + "status": self.status, + "error_message": self.error_message, + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "SyncMetadata": + """Create from dictionary.""" + return cls( + operation=SyncOperation(data["operation"]), + source=data["source"], + destination=data["destination"], + timestamp=datetime.fromisoformat(data["timestamp"]), + committed_files=data.get("committed_files", []), + conflicts=data.get("conflicts", []), + merged_files=data.get("merged_files", []), + status=data.get("status", "pending"), + error_message=data.get("error_message"), + ) + + +class Severity(str, Enum): + """Validation severity levels.""" + + ERROR = "error" + WARNING = "warning" + INFO = "info" + + +@dataclass +class ValidationIssue: + """Represents a validation issue.""" + + rule: str + message: str + severity: Severity + file_path: Optional[str] = None + line_number: Optional[int] = None + suggestion: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + """Convert to dictionary.""" + return { + "rule": self.rule, + "message": self.message, + "severity": self.severity.value, + "file_path": self.file_path, + "line_number": self.line_number, + "suggestion": self.suggestion, + } + + +@dataclass +class ValidationResult: + """Result of configuration validation.""" + + is_valid: bool + issues: List[ValidationIssue] = field(default_factory=list) + validated_files: int = 0 + checked_at: datetime = field(default_factory=datetime.now) + + def add_issue(self, issue: ValidationIssue) -> None: + """Add a validation issue.""" + self.issues.append(issue) + if issue.severity == Severity.ERROR: + self.is_valid = False + + def to_dict(self) -> Dict[str, Any]: + """Convert to dictionary.""" + return { + "is_valid": self.is_valid, + "issues": [issue.to_dict() for issue in self.issues], + "validated_files": self.validated_files, + "checked_at": self.checked_at.isoformat(), + } + + +@dataclass +class MergeResult: + """Result of a merge operation.""" + + success: bool + merged_files: Dict[str, str] = field(default_factory=dict) + conflicts: Dict[str, str] = field(default_factory=dict) + unresolved: List[str] = field(default_factory=list) + strategy_used: str = "" + message: str = "" + + def to_dict(self) -> Dict[str, Any]: + """Convert to dictionary.""" + return { + "success": self.success, + "merged_files": self.merged_files, + "conflicts": self.conflicts, + "unresolved": self.unresolved, + "strategy_used": self.strategy_used, + "message": self.message, + } + + +@dataclass +class HistoryEntry: + """Entry in the configuration change history.""" + + id: str = field(default_factory=lambda: str(uuid.uuid4())) + timestamp: datetime = field(default_factory=datetime.now) + operation: str = "" + files_changed: List[str] = field(default_factory=list) + commit_message: str = "" + diff: Optional[str] = None + tags: List[str] = field(default_factory=list) + + def to_dict(self) -> Dict[str, Any]: + """Convert to dictionary.""" + return { + "id": self.id, + "timestamp": self.timestamp.isoformat(), + "operation": self.operation, + "files_changed": self.files_changed, + "commit_message": self.commit_message, + "diff": self.diff, + "tags": self.tags, + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "HistoryEntry": + """Create from dictionary.""" + data["timestamp"] = datetime.fromisoformat(data["timestamp"]) + return cls(**data)