348 lines
11 KiB
Python
348 lines
11 KiB
Python
"""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)
|