68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
STATE_FILENAME = ".copienator-gui.json"
|
|
|
|
|
|
class StateStore:
|
|
def __init__(self) -> None:
|
|
self.evaluation: Path | None = None
|
|
self.data: dict[str, Any] = self._empty_data()
|
|
|
|
@staticmethod
|
|
def _empty_data() -> dict[str, Any]:
|
|
return {"version": 1, "steps": {}, "history": []}
|
|
|
|
def load(self, evaluation: Path) -> None:
|
|
self.evaluation = evaluation
|
|
path = evaluation / STATE_FILENAME
|
|
try:
|
|
loaded = json.loads(path.read_text(encoding="utf-8"))
|
|
self.data = loaded if isinstance(loaded, dict) else self._empty_data()
|
|
except (OSError, json.JSONDecodeError):
|
|
self.data = self._empty_data()
|
|
self.data.setdefault("version", 1)
|
|
self.data.setdefault("steps", {})
|
|
self.data.setdefault("history", [])
|
|
|
|
def save(self) -> None:
|
|
if not self.evaluation:
|
|
return
|
|
path = self.evaluation / STATE_FILENAME
|
|
temporary = path.with_suffix(path.suffix + ".tmp")
|
|
temporary.write_text(
|
|
json.dumps(self.data, indent=2, ensure_ascii=False), encoding="utf-8"
|
|
)
|
|
temporary.replace(path)
|
|
|
|
def step(self, step_id: str) -> dict[str, Any]:
|
|
return self.data["steps"].setdefault(step_id, {})
|
|
|
|
def update_step(self, step_id: str, **changes: Any) -> None:
|
|
entry = self.step(step_id)
|
|
entry.update(changes)
|
|
entry["updated_at"] = datetime.now(timezone.utc).isoformat()
|
|
self.save()
|
|
|
|
def add_history(self, entry: dict[str, Any]) -> None:
|
|
entry = dict(entry)
|
|
entry["timestamp"] = datetime.now(timezone.utc).isoformat()
|
|
self.data["history"].append(entry)
|
|
self.data["history"] = self.data["history"][-200:]
|
|
self.save()
|
|
|
|
def invalidate_after(self, ordered_ids: list[str], current_id: str) -> None:
|
|
try:
|
|
index = ordered_ids.index(current_id)
|
|
except ValueError:
|
|
return
|
|
for step_id in ordered_ids[index + 1 :]:
|
|
entry = self.step(step_id)
|
|
if entry.get("status") in {"success", "detected", "skipped"}:
|
|
entry["status"] = "stale"
|
|
self.save()
|