src/turn_detection/data/manifest.py

"""Portable JSONL manifest I/O.

Audit manifests contain the following stable keys:

``record_id, source_file, source_row, language, endpoint, midfiller,
endfiller, synthetic, dataset, audio_path, audio_sha256, audio_num_bytes,
audio_format, sample_rate, num_channels, num_frames, bits_per_sample,
duration_seconds, spoken_text, group_keys, group_id, validation_errors,
validation_warnings``.

Split manifests retain those keys and add ``split``.  Audio bytes are never
embedded in a manifest; ``source_file`` and ``source_row`` preserve provenance.
"""

from __future__ import annotations

import json
import os
from pathlib import Path
import tempfile
from typing import Any, Iterable, Iterator, Mapping


MANIFEST_SCHEMA_VERSION = "1.0"


def read_manifest(path: str | Path, *, limit: int | None = None) -> Iterator[dict[str, Any]]:
    """Stream a JSONL manifest using only the Python standard library."""

    if limit is not None and limit < 0:
        raise ValueError("limit cannot be negative")
    manifest_path = Path(path)
    with manifest_path.open("r", encoding="utf-8") as handle:
        emitted = 0
        for line_number, line in enumerate(handle, start=1):
            if not line.strip():
                continue
            if limit is not None and emitted >= limit:
                return
            try:
                value = json.loads(line)
            except json.JSONDecodeError as exc:
                raise ValueError(f"invalid JSON at {manifest_path}:{line_number}: {exc}") from exc
            if not isinstance(value, dict):
                raise ValueError(f"manifest row at {manifest_path}:{line_number} is not an object")
            yield value
            emitted += 1


def write_manifest(path: str | Path, rows: Iterable[Mapping[str, Any]]) -> int:
    """Atomically write JSONL rows and return the number written."""

    output_path = Path(path)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary_name = tempfile.mkstemp(
        prefix=f".{output_path.name}.", suffix=".tmp", dir=str(output_path.parent)
    )
    count = 0
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
            for row in rows:
                handle.write(json.dumps(dict(row), sort_keys=True, ensure_ascii=False, allow_nan=False))
                handle.write("\n")
                count += 1
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary_name, output_path)
    except BaseException:
        try:
            os.unlink(temporary_name)
        except FileNotFoundError:
            pass
        raise
    return count


def write_json(path: str | Path, payload: Mapping[str, Any]) -> None:
    """Atomically write a deterministic UTF-8 JSON document."""

    output_path = Path(path)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary_name = tempfile.mkstemp(
        prefix=f".{output_path.name}.", suffix=".tmp", dir=str(output_path.parent)
    )
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
            json.dump(dict(payload), handle, indent=2, sort_keys=True, ensure_ascii=False, allow_nan=False)
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary_name, output_path)
    except BaseException:
        try:
            os.unlink(temporary_name)
        except FileNotFoundError:
            pass
        raise