239 lines
7.3 KiB
Python
239 lines
7.3 KiB
Python
|
"""Manifest dataclass for vault metadata."""
|
||
|
|
|
||
|
|
import json
|
||
|
|
from dataclasses import dataclass, field
|
||
|
|
from datetime import datetime
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Literal
|
||
|
|
from uuid import uuid4
|
||
|
|
|
||
|
|
from src.core.file_entry import FileEntry
|
||
|
|
|
||
|
|
|
||
|
|
LocationStatus = Literal["active", "unreachable"]
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class Location:
|
||
|
|
"""Represents a vault replica location.
|
||
|
|
|
||
|
|
Attributes:
|
||
|
|
path: Absolute path to the .vault file
|
||
|
|
last_seen: Last time this location was accessible
|
||
|
|
status: Current status (active/unreachable)
|
||
|
|
"""
|
||
|
|
|
||
|
|
path: str
|
||
|
|
last_seen: datetime
|
||
|
|
status: LocationStatus = "active"
|
||
|
|
|
||
|
|
def to_dict(self) -> dict:
|
||
|
|
"""Convert to dictionary for JSON serialization."""
|
||
|
|
return {
|
||
|
|
"path": self.path,
|
||
|
|
"last_seen": self.last_seen.isoformat(),
|
||
|
|
"status": self.status,
|
||
|
|
}
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def from_dict(cls, data: dict) -> "Location":
|
||
|
|
"""Create Location from dictionary."""
|
||
|
|
return cls(
|
||
|
|
path=data["path"],
|
||
|
|
last_seen=datetime.fromisoformat(data["last_seen"]),
|
||
|
|
status=data["status"],
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class Manifest:
|
||
|
|
"""Vault manifest containing all metadata.
|
||
|
|
|
||
|
|
This is stored as .vault/manifest.json inside each vault image.
|
||
|
|
|
||
|
|
Attributes:
|
||
|
|
vault_id: Unique identifier for the vault (UUID)
|
||
|
|
vault_name: Human-readable name
|
||
|
|
version: Manifest format version
|
||
|
|
created: Creation timestamp
|
||
|
|
last_modified: Last modification timestamp
|
||
|
|
image_size_mb: Size of the vault image in MB
|
||
|
|
locations: List of all replica locations
|
||
|
|
files: List of all files in the vault
|
||
|
|
"""
|
||
|
|
|
||
|
|
vault_id: str
|
||
|
|
vault_name: str
|
||
|
|
image_size_mb: int
|
||
|
|
created: datetime
|
||
|
|
last_modified: datetime
|
||
|
|
version: int = 1
|
||
|
|
locations: list[Location] = field(default_factory=list)
|
||
|
|
files: list[FileEntry] = field(default_factory=list)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def create_new(cls, vault_name: str, image_size_mb: int, location_path: str) -> "Manifest":
|
||
|
|
"""Create a new manifest for a fresh vault.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
vault_name: Human-readable name for the vault
|
||
|
|
image_size_mb: Size of the vault image in MB
|
||
|
|
location_path: Path to the first .vault file
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
New Manifest instance
|
||
|
|
"""
|
||
|
|
now = datetime.now()
|
||
|
|
resolved = str(Path(location_path).resolve())
|
||
|
|
return cls(
|
||
|
|
vault_id=str(uuid4()),
|
||
|
|
vault_name=vault_name,
|
||
|
|
image_size_mb=image_size_mb,
|
||
|
|
created=now,
|
||
|
|
last_modified=now,
|
||
|
|
locations=[Location(path=resolved, last_seen=now, status="active")],
|
||
|
|
files=[],
|
||
|
|
)
|
||
|
|
|
||
|
|
def to_dict(self) -> dict:
|
||
|
|
"""Convert to dictionary for JSON serialization."""
|
||
|
|
return {
|
||
|
|
"vault_id": self.vault_id,
|
||
|
|
"vault_name": self.vault_name,
|
||
|
|
"version": self.version,
|
||
|
|
"created": self.created.isoformat(),
|
||
|
|
"last_modified": self.last_modified.isoformat(),
|
||
|
|
"image_size_mb": self.image_size_mb,
|
||
|
|
"locations": [loc.to_dict() for loc in self.locations],
|
||
|
|
"files": [f.to_dict() for f in self.files],
|
||
|
|
}
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def from_dict(cls, data: dict) -> "Manifest":
|
||
|
|
"""Create Manifest from dictionary."""
|
||
|
|
return cls(
|
||
|
|
vault_id=data["vault_id"],
|
||
|
|
vault_name=data["vault_name"],
|
||
|
|
version=data["version"],
|
||
|
|
created=datetime.fromisoformat(data["created"]),
|
||
|
|
last_modified=datetime.fromisoformat(data["last_modified"]),
|
||
|
|
image_size_mb=data["image_size_mb"],
|
||
|
|
locations=[Location.from_dict(loc) for loc in data["locations"]],
|
||
|
|
files=[FileEntry.from_dict(f) for f in data["files"]],
|
||
|
|
)
|
||
|
|
|
||
|
|
def save(self, mount_point: Path) -> None:
|
||
|
|
"""Save manifest to .vault/manifest.json.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
mount_point: Path where the vault is mounted
|
||
|
|
"""
|
||
|
|
vault_dir = mount_point / ".vault"
|
||
|
|
vault_dir.mkdir(exist_ok=True)
|
||
|
|
|
||
|
|
manifest_path = vault_dir / "manifest.json"
|
||
|
|
with open(manifest_path, "w", encoding="utf-8") as f:
|
||
|
|
json.dump(self.to_dict(), f, indent=2)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def load(cls, mount_point: Path) -> "Manifest":
|
||
|
|
"""Load manifest from .vault/manifest.json.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
mount_point: Path where the vault is mounted
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Loaded Manifest instance
|
||
|
|
|
||
|
|
Raises:
|
||
|
|
FileNotFoundError: If manifest doesn't exist
|
||
|
|
"""
|
||
|
|
manifest_path = mount_point / ".vault" / "manifest.json"
|
||
|
|
with open(manifest_path, "r", encoding="utf-8") as f:
|
||
|
|
data = json.load(f)
|
||
|
|
return cls.from_dict(data)
|
||
|
|
|
||
|
|
def add_location(self, path: str) -> None:
|
||
|
|
"""Add a new replica location.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
path: Absolute path to the .vault file
|
||
|
|
"""
|
||
|
|
resolved = str(Path(path).resolve())
|
||
|
|
# Don't add duplicate locations
|
||
|
|
for loc in self.locations:
|
||
|
|
if str(Path(loc.path).resolve()) == resolved:
|
||
|
|
loc.status = "active"
|
||
|
|
loc.last_seen = datetime.now()
|
||
|
|
self.last_modified = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
self.locations.append(
|
||
|
|
Location(path=resolved, last_seen=datetime.now(), status="active")
|
||
|
|
)
|
||
|
|
self.last_modified = datetime.now()
|
||
|
|
|
||
|
|
def update_location_status(self, path: str, status: LocationStatus) -> None:
|
||
|
|
"""Update status of a location.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
path: Path to the location
|
||
|
|
status: New status
|
||
|
|
"""
|
||
|
|
resolved = str(Path(path).resolve())
|
||
|
|
for loc in self.locations:
|
||
|
|
if str(Path(loc.path).resolve()) == resolved:
|
||
|
|
loc.status = status
|
||
|
|
if status == "active":
|
||
|
|
loc.last_seen = datetime.now()
|
||
|
|
break
|
||
|
|
self.last_modified = datetime.now()
|
||
|
|
|
||
|
|
def add_file(self, file_entry: FileEntry) -> None:
|
||
|
|
"""Add or update a file entry.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
file_entry: File entry to add/update
|
||
|
|
"""
|
||
|
|
# Remove existing entry with same path
|
||
|
|
self.files = [f for f in self.files if f.path != file_entry.path]
|
||
|
|
self.files.append(file_entry)
|
||
|
|
self.last_modified = datetime.now()
|
||
|
|
|
||
|
|
def add_file_from_path(self, base_path: Path, file_path: Path) -> FileEntry:
|
||
|
|
"""Add a file entry from a file path.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
base_path: Base path (mount point) for relative path calculation
|
||
|
|
file_path: Absolute path to the file
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
The created FileEntry
|
||
|
|
"""
|
||
|
|
entry = FileEntry.from_path(base_path, file_path)
|
||
|
|
self.add_file(entry)
|
||
|
|
return entry
|
||
|
|
|
||
|
|
def remove_file(self, path: str) -> None:
|
||
|
|
"""Remove a file entry by path.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
path: Relative path of the file to remove
|
||
|
|
"""
|
||
|
|
self.files = [f for f in self.files if f.path != path]
|
||
|
|
self.last_modified = datetime.now()
|
||
|
|
|
||
|
|
def get_file(self, path: str) -> FileEntry | None:
|
||
|
|
"""Get file entry by path.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
path: Relative path of the file
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
FileEntry if found, None otherwise
|
||
|
|
"""
|
||
|
|
for f in self.files:
|
||
|
|
if f.path == path:
|
||
|
|
return f
|
||
|
|
return None
|