| """Small IO helpers shared across pipeline stages.""" | |
| from __future__ import annotations | |
| import json | |
| from pathlib import Path | |
| from typing import Any, Iterable, Iterator | |
| def read_jsonl(path: str | Path) -> Iterator[dict[str, Any]]: | |
| with open(path, "r") as f: | |
| for line in f: | |
| line = line.strip() | |
| if line: | |
| yield json.loads(line) | |
| def write_jsonl(path: str | Path, rows: Iterable[dict[str, Any]]) -> None: | |
| path = Path(path) | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| with open(path, "w") as f: | |
| for row in rows: | |
| f.write(json.dumps(row) + "\n") | |