77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
"""Protected inventory and restoration of automatic recovery snapshots."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
|
|
from core.config import CONFIG_DIR
|
|
from core.database import char_put, reh_put
|
|
from core.validation import _safe_child_path
|
|
|
|
router = APIRouter()
|
|
_BACKUP_ROOT = CONFIG_DIR / "backups"
|
|
_BACKUP_KINDS = {"characters", "rehearsals"}
|
|
|
|
|
|
def _backup_path(kind: str, filename: str) -> Path:
|
|
if kind not in _BACKUP_KINDS or not filename or Path(filename).name != filename or not filename.endswith(".json"):
|
|
raise HTTPException(400, "Invalid backup reference")
|
|
directory = _BACKUP_ROOT / kind
|
|
path = _safe_child_path(directory, directory / filename)
|
|
if not path.is_file():
|
|
raise HTTPException(404, "Backup not found")
|
|
return path
|
|
|
|
|
|
def _read_backup(path: Path) -> dict:
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError) as exc:
|
|
raise HTTPException(422, f"Backup cannot be read: {exc}")
|
|
if not isinstance(data, dict) or not isinstance(data.get("record"), dict):
|
|
raise HTTPException(422, "Invalid backup format")
|
|
return data
|
|
|
|
|
|
@router.get("/api/backups")
|
|
async def list_backups():
|
|
items: list[dict] = []
|
|
for kind in sorted(_BACKUP_KINDS):
|
|
directory = _BACKUP_ROOT / kind
|
|
if not directory.is_dir():
|
|
continue
|
|
for path in sorted(directory.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True):
|
|
try:
|
|
data = _read_backup(path)
|
|
record = data["record"]
|
|
except HTTPException:
|
|
continue
|
|
items.append({
|
|
"kind": kind,
|
|
"file": path.name,
|
|
"saved_at": data.get("saved_at", ""),
|
|
"reason": data.get("reason", ""),
|
|
"id": record.get("id"),
|
|
"name": record.get("name") or record.get("title", ""),
|
|
})
|
|
return {"backups": items[:500]}
|
|
|
|
|
|
@router.post("/api/backups/restore")
|
|
async def restore_backup(request: Request):
|
|
try:
|
|
data = await request.json()
|
|
except Exception:
|
|
raise HTTPException(400, "Invalid JSON")
|
|
kind = str(data.get("kind") or "").strip().lower()
|
|
path = _backup_path(kind, str(data.get("file") or "").strip())
|
|
payload = _read_backup(path)
|
|
record = payload["record"]
|
|
if kind == "characters":
|
|
restored = char_put(record)
|
|
return {"ok": True, "kind": kind, "id": restored.get("id"), "restored_from": path.name}
|
|
restored = reh_put(record)
|
|
return {"ok": True, "kind": kind, "id": restored.get("id"), "restored_from": path.name}
|