54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import secrets
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
PROJECT_DIR = Path(__file__).resolve().parent.parent
|
|
configured_data_dir = os.environ.get("DATA_DIR", "").strip()
|
|
if configured_data_dir:
|
|
configured_path = Path(configured_data_dir).expanduser()
|
|
if not configured_path.is_absolute():
|
|
configured_path = PROJECT_DIR / configured_path
|
|
DATA_DIR = configured_path.resolve()
|
|
else:
|
|
DATA_DIR = (PROJECT_DIR / "data").resolve()
|
|
|
|
RUNS_DIR = DATA_DIR / "runs"
|
|
RUNS_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def create_run_id() -> str:
|
|
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
return f"{stamp}-{secrets.token_hex(4)}"
|
|
|
|
|
|
def run_dir(run_id: str) -> Path:
|
|
if not run_id or any(
|
|
ch not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_"
|
|
for ch in run_id
|
|
):
|
|
raise ValueError("Ungültige Run-ID.")
|
|
path = RUNS_DIR / run_id
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
return path
|
|
|
|
|
|
def write_json(run_id: str, filename: str, payload: Any) -> Path:
|
|
path = run_dir(run_id) / filename
|
|
temporary = path.with_suffix(path.suffix + ".tmp")
|
|
temporary.write_text(
|
|
json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
temporary.replace(path)
|
|
return path
|
|
|
|
|
|
def read_json(run_id: str, filename: str) -> Any:
|
|
return json.loads((run_dir(run_id) / filename).read_text(encoding="utf-8"))
|