Spaces:
Paused
Paused
File size: 1,914 Bytes
86974b4 | 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 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 | """Transcript read/write helpers.
JSONL is the canonical format for long-running append-heavy collection jobs.
JSON array format is still supported for backward compatibility.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Iterable
def _ensure_parent(path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
def load_records(path: str | Path) -> list[dict]:
p = Path(path)
if not p.exists():
return []
if p.suffix.lower() == ".jsonl":
records: list[dict] = []
with p.open("r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
records.append(json.loads(line))
return records
with p.open("r", encoding="utf-8") as f:
payload = json.load(f)
if isinstance(payload, list):
return payload
raise ValueError(f"Expected list payload in {p}, got {type(payload).__name__}")
def save_records(path: str | Path, records: Iterable[dict]) -> None:
p = Path(path)
_ensure_parent(p)
items = list(records)
if p.suffix.lower() == ".jsonl":
with p.open("w", encoding="utf-8") as f:
for item in items:
f.write(json.dumps(item, ensure_ascii=True) + "\n")
return
with p.open("w", encoding="utf-8") as f:
json.dump(items, f, indent=4)
def append_records(path: str | Path, new_records: Iterable[dict]) -> None:
p = Path(path)
_ensure_parent(p)
items = list(new_records)
if not items:
return
if p.suffix.lower() == ".jsonl":
with p.open("a", encoding="utf-8") as f:
for item in items:
f.write(json.dumps(item, ensure_ascii=True) + "\n")
return
existing = load_records(p)
existing.extend(items)
save_records(p, existing)
|