60 lines
2.3 KiB
Python
60 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from fnmatch import fnmatch
|
|
from hashlib import sha256
|
|
from pathlib import Path
|
|
from difflib import unified_diff
|
|
|
|
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():
|
|
content = path.read_text(encoding="utf-8")
|
|
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:
|
|
for path, content in snapshot.file_contents.items():
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content, encoding="utf-8")
|
|
|
|
def diff_summary(self, snapshot: BaselineSnapshot) -> str:
|
|
lines: list[str] = []
|
|
current_contents = {path: path.read_text(encoding="utf-8") for path in self.resolve_paths()}
|
|
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)
|