"""Snapshots + undo stack for transcript edits.
Each mutating operation calls snapshot() BEFORE the mutation, capturing
the pre-mutation JSON. _history[] in the main JSON references each
snapshot. cleanup_snapshots enforces SNAPSHOT_CAP.
"""
import json
import shutil
import time
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
SNAPSHOT_CAP = 20
def _now_iso() -> str:
return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
def _load(json_path: str) -> dict[str, Any]:
data: dict[str, Any] = json.loads(Path(json_path).read_text(encoding="utf-8"))
return data
def _save(json_path: str, data: dict[str, Any]) -> None:
Path(json_path).write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
def snapshot(json_path: str, action: str, metric: str) -> str:
"""Save a pre-mutation snapshot and append a _history entry.
Returns absolute path to the snapshot file.
Snapshot is written BEFORE the caller mutates json_path, so it reflects
the state to revert TO if undone.
"""
path = Path(json_path)
snap_dir = path.parent / "snapshots"
snap_dir.mkdir(exist_ok=True)
# Use millisecond-precision timestamp so rapid successive snapshots don't collide
ts = int(time.time() * 1000)
snap_file = snap_dir / f"{ts}.json"
# Ensure unique filename if same ms-tick
while snap_file.exists():
ts += 1
snap_file = snap_dir / f"{ts}.json"
shutil.copy2(path, snap_file)
data = _load(json_path)
data.setdefault("_history", []).append(
{
"ts": _now_iso(),
"action": action,
"metric": metric,
"snapshot": f"snapshots/{snap_file.name}",
}
)
_save(json_path, data)
return str(snap_file)
def cleanup_snapshots(json_path: str) -> int:
"""Delete oldest snapshots beyond SNAPSHOT_CAP. Returns count deleted."""
data = _load(json_path)
history: list[dict[str, Any]] = data.get("_history", [])
if len(history) <= SNAPSHOT_CAP:
return 0
excess = len(history) - SNAPSHOT_CAP
to_remove = history[:excess]
parent = Path(json_path).parent
deleted = 0
for entry in to_remove:
snap_rel = entry.get("snapshot", "")
if snap_rel:
snap_path = parent / snap_rel
if snap_path.exists():
snap_path.unlink()
deleted += 1
data["_history"] = history[excess:]
_save(json_path, data)
return deleted
def undo_last(json_path: str) -> dict[str, Any] | None:
"""Restore the latest snapshot and pop the matching _history entry.
Returns the popped history entry, or None if _history is empty.
Snapshot file is deleted after restoration.
Note on history bookkeeping: snapshot() writes the file BEFORE appending
to _history. So the snapshot content lacks the just-appended history
entry. After restoring from snapshot, no further pop is needed.
"""
data = _load(json_path)
history: list[dict[str, Any]] = data.get("_history", [])
if not history:
return None
entry = history[-1]
parent = Path(json_path).parent
snap_rel = entry.get("snapshot", "")
if not snap_rel:
history.pop()
_save(json_path, data)
return entry
snap_path = parent / snap_rel
if snap_path.exists():
shutil.copy2(snap_path, json_path) # restores pre-mutation state
snap_path.unlink()
else:
# Snapshot missing — fall back to popping history entry only
history.pop()
_save(json_path, data)
return entry
def list_history(json_path: str) -> list[dict[str, Any]]:
"""Return _history entries (oldest first). Empty list if absent."""
if not Path(json_path).exists():
return []
data = _load(json_path)
history: list[dict[str, Any]] = list(data.get("_history", []))
return history