File size: 1,441 Bytes
ce209f5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 | from __future__ import annotations
import csv
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Mapping
ROOT = Path(__file__).resolve().parents[1]
def write_json(data: Mapping, path: str | Path) -> Path:
path = Path(path)
if not path.is_absolute():
path = ROOT / path
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", encoding="utf-8") as f:
json.dump(data, f, indent=2, sort_keys=True)
return path
def append_result(row: Mapping, path: str | Path = "results/results.csv") -> Path:
path = Path(path)
if not path.is_absolute():
path = ROOT / path
path.parent.mkdir(parents=True, exist_ok=True)
payload = dict(row)
payload.setdefault("timestamp_utc", datetime.now(timezone.utc).isoformat())
exists = path.exists()
fieldnames = sorted(payload)
if exists:
with path.open("r", newline="", encoding="utf-8") as f:
existing = csv.DictReader(f)
fieldnames = sorted(set(existing.fieldnames or []) | set(fieldnames))
rows = []
if exists:
with path.open("r", newline="", encoding="utf-8") as f:
rows = list(csv.DictReader(f))
rows.append(payload)
with path.open("w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(rows)
return path
|