62 lines
2.3 KiB
Python
62 lines
2.3 KiB
Python
"""Locations and non-destructive retention for local operational reports."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from tools.rag.common import DATA_DIR
|
|
|
|
CORE_JSON = {"documents.json", "chunks.json", "index_version.json"}
|
|
|
|
|
|
def run_id() -> str:
|
|
return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "-" + uuid4().hex[:8]
|
|
|
|
|
|
def report_output(requested: Path | None, category: str, *, data_dir: Path = DATA_DIR) -> Path:
|
|
if not category.replace("_", "").isalnum():
|
|
raise ValueError("Invalid report category")
|
|
if requested is None:
|
|
target = data_dir / "reports" / category / (run_id() + ".json")
|
|
else:
|
|
target = requested.resolve()
|
|
if target.parent == data_dir.resolve():
|
|
if target.name in CORE_JSON or target.name.startswith(".") or target.suffix != ".json":
|
|
raise ValueError("Report cannot overwrite operational data")
|
|
target = data_dir / "reports" / category / target.name
|
|
if target.exists():
|
|
raise FileExistsError(f"Report already exists: {target}")
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
return target
|
|
|
|
|
|
def report_log(report: Path, *, data_dir: Path = DATA_DIR) -> Path:
|
|
try:
|
|
relative = report.resolve().relative_to((data_dir / "reports").resolve())
|
|
except ValueError:
|
|
identity = hashlib.sha256(str(report.absolute()).encode()).hexdigest()[:8]
|
|
relative = Path("custom") / (report.stem + "-" + identity + ".json")
|
|
target = data_dir / "logs" / relative.with_suffix(".pytest.log")
|
|
if target.exists():
|
|
raise FileExistsError(f"Log already exists: {target}")
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
return target
|
|
|
|
|
|
def resolve_report_input(path: Path, *, data_dir: Path = DATA_DIR) -> Path:
|
|
if path.exists():
|
|
return path
|
|
mappings = {}
|
|
for record in sorted((data_dir / "reports/migrations").glob("*.json")):
|
|
for move in json.loads(record.read_text()).get("moves", []):
|
|
mappings[Path(move["source"]).resolve()] = Path(move["destination"])
|
|
visited = set()
|
|
while not path.exists() and path.resolve() in mappings and path.resolve() not in visited:
|
|
visited.add(path.resolve())
|
|
path = mappings[path.resolve()]
|
|
return path
|