from __future__ import annotations from dataclasses import dataclass from difflib import unified_diff from fnmatch import fnmatch from hashlib import sha256 from pathlib import Path from engine.models import BaselineSnapshot, TaskSpec @dataclass(frozen=True) class ArtifactManager: task: TaskSpec def resolve_paths(self) -> list[Path]: root_dir = self.task.root_dir resolved: set[Path] = set() for pattern in self.task.artifacts.include: for path in root_dir.glob(pattern): if not path.is_file(): continue relative_path = path.relative_to(root_dir).as_posix() if any(fnmatch(relative_path, exclude) for exclude in self.task.artifacts.exclude): continue resolved.add(path) return sorted(resolved) def snapshot(self) -> BaselineSnapshot: file_contents: dict[Path, str] = {} file_hashes: dict[Path, str] = {} for path in self.resolve_paths(): with path.open("r", encoding="utf-8", newline="") as handle: content = handle.read() file_contents[path] = content file_hashes[path] = sha256(content.encode("utf-8")).hexdigest() return BaselineSnapshot(file_contents=file_contents, file_hashes=file_hashes) def restore(self, snapshot: BaselineSnapshot) -> None: current_paths = set(self.resolve_paths()) snapshot_paths = set(snapshot.file_contents) for path in current_paths - snapshot_paths: path.unlink() for path, content in snapshot.file_contents.items(): path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8", newline="") as handle: handle.write(content) def diff_summary(self, snapshot: BaselineSnapshot) -> str: lines: list[str] = [] current_contents: dict[Path, str] = {} for path in self.resolve_paths(): with path.open("r", encoding="utf-8", newline="") as handle: current_contents[path] = handle.read() all_paths = sorted(set(snapshot.file_contents) | set(current_contents)) for path in all_paths: before = snapshot.file_contents.get(path, "") after = current_contents.get(path, "") if before == after: continue diff = unified_diff( before.splitlines(keepends=True), after.splitlines(keepends=True), fromfile=f"{path.as_posix()} (before)", tofile=f"{path.as_posix()} (after)", ) lines.extend(diff) return "".join(lines)