# License: Apache-2.0 License # Created by: Patrick Lumbantobing, VertoX-AI # Copyright (c) 2026 VertoX-AI. All rights reserved. # # This work is licensed under the Apache-2.0 License. # To view a copy of this license, visit # https://www.apache.org/licenses/LICENSE-2.0 """Read and atomically enrich metadata CSVs without changing source cell values.""" from __future__ import annotations import csv import os import tempfile from dataclasses import dataclass from decimal import Decimal from pathlib import Path from typing import Final from speech_text_data_aligners.core.errors import MetadataError from speech_text_data_aligners.core.extraction import BOUNDARY_COLUMNS SOURCE_COLUMNS: Final = ( "line_id", "audio_id", "en_transcript", "ru_translation", ) @dataclass(frozen=True, slots=True) class CsvFormat: """Capture deterministic CSV serialization details discovered from an input. Parameters ---------- delimiter: One-character column delimiter. quotechar: One-character field quote marker. lineterminator: Original dominant line ending. encoding: ``utf-8`` or ``utf-8-sig`` according to input BOM presence. """ delimiter: str = "," quotechar: str = '"' lineterminator: str = "\n" encoding: str = "utf-8" @dataclass(frozen=True, slots=True) class MetadataTable: """Store CSV headers and cells immutably in source order. Parameters ---------- source_path: Path from which the table was read. fieldnames: Exact ordered source header cells. rows: Ordered rows, each with exactly one value per field. csv_format: Detected deterministic read/write policy. """ source_path: Path fieldnames: tuple[str, ...] rows: tuple[tuple[str, ...], ...] csv_format: CsvFormat def row_dict(self, index: int) -> dict[str, str]: """Return one source row as a new insertion-ordered dictionary. Parameters ---------- index: Zero-based row ordinal. Returns ------- A new mapping that preserves header order and exact cell strings. Raises ------ IndexError: If ``index`` is outside the table. """ return dict(zip(self.fieldnames, self.rows[index], strict=True)) def _detect_format(raw: bytes) -> CsvFormat: """Detect BOM, delimiter, quote character, and dominant line ending. Parameters ---------- raw: Complete CSV bytes. Returns ------- A bounded format policy with safe fallbacks for small valid files. """ encoding = "utf-8-sig" if raw.startswith(b"\xef\xbb\xbf") else "utf-8" text = raw.decode(encoding) lineterminator = "\r\n" if "\r\n" in text else "\n" try: dialect = csv.Sniffer().sniff(text[:8192], delimiters=",\t;|") delimiter = dialect.delimiter quotechar = dialect.quotechar or '"' except csv.Error: delimiter = "," quotechar = '"' return CsvFormat(delimiter, quotechar, lineterminator, encoding) def read_metadata(path: Path) -> MetadataTable: """Read a metadata/reference CSV while preserving headers, cells, and row order. Parameters ---------- path: Existing UTF-8 or UTF-8-with-BOM CSV path. Returns ------- Immutable table with exact source values and detected format policy. Raises ------ MetadataError: If the file is unreadable, malformed, non-UTF-8, or lacks each required source column exactly once. """ try: raw = path.read_bytes() csv_format = _detect_format(raw) text = raw.decode(csv_format.encoding) except (OSError, UnicodeError) as error: raise MetadataError(f"cannot read metadata CSV: {path}") from error try: reader = csv.reader( text.splitlines(keepends=True), delimiter=csv_format.delimiter, quotechar=csv_format.quotechar, strict=True, ) records = list(reader) except csv.Error as error: raise MetadataError(f"malformed metadata CSV: {path}") from error if not records: raise MetadataError("metadata CSV is empty") header = tuple(records[0]) if any(not name for name in header) or len(header) != len(set(header)): raise MetadataError("metadata header contains an empty or duplicate column") missing = [name for name in SOURCE_COLUMNS if header.count(name) != 1] if missing: raise MetadataError(f"metadata CSV is missing required columns: {missing}") rows: list[tuple[str, ...]] = [] for ordinal, record in enumerate(records[1:]): if len(record) != len(header): raise MetadataError(f"metadata row {ordinal} has the wrong column count") rows.append(tuple(record)) return MetadataTable(path, header, tuple(rows), csv_format) def _format_seconds(value: str | float | None) -> str: """Render a boundary value without scientific notation or redundant zeros. Parameters ---------- value: Text, finite seconds, or ``None`` for an empty failure cell. Returns ------- Original text or deterministic fixed-point decimal representation. """ if value is None: return "" if isinstance(value, str): return value rendered = format(Decimal(str(value)), "f") if "." in rendered: rendered = rendered.rstrip("0").rstrip(".") return rendered or "0" def write_annotated_metadata( table: MetadataTable, output_csv: Path, annotations: tuple[dict[str, str | float | None], ...], *, overwrite: bool = False, ) -> None: """Atomically append exactly six boundary columns to every ordered source row. Existing input values are serialized unchanged. The completed temporary file is flushed and fsynced before same-directory atomic replacement; no partial output is published on failure. Parameters ---------- table: Source table returned by :func:`read_metadata`. output_csv: New publication path, which must differ from the source path. annotations: One exact six-key mapping per source row in source order. overwrite: Explicitly allow replacement of an existing output file. Raises ------ MetadataError: If targets are unsafe, annotations violate the schema, or serialization/publication fails. Side Effects: Creates or atomically replaces ``output_csv`` after complete serialization. """ try: same_target = table.source_path.resolve() == output_csv.resolve() except OSError as error: raise MetadataError("cannot resolve metadata input/output paths") from error if same_target: raise MetadataError("source metadata CSV cannot be used as the output target") if any(column in table.fieldnames for column in BOUNDARY_COLUMNS): raise MetadataError("source metadata already contains an alignment column") if len(annotations) != len(table.rows): raise MetadataError("annotation count does not match source row count") expected_keys = set(BOUNDARY_COLUMNS) if any(set(annotation) != expected_keys for annotation in annotations): raise MetadataError("each annotation must contain exactly six boundary fields") if output_csv.exists() and not overwrite: raise MetadataError("output CSV exists and overwrite was not authorized") output_csv.parent.mkdir(parents=True, exist_ok=True) temporary_path: Path | None = None try: with tempfile.NamedTemporaryFile( mode="w", encoding=table.csv_format.encoding, newline="", dir=output_csv.parent, prefix=f".{output_csv.name}.", suffix=".tmp", delete=False, ) as stream: temporary_path = Path(stream.name) writer = csv.writer( stream, delimiter=table.csv_format.delimiter, quotechar=table.csv_format.quotechar, lineterminator=table.csv_format.lineterminator, quoting=csv.QUOTE_MINIMAL, ) writer.writerow((*table.fieldnames, *BOUNDARY_COLUMNS)) for row, annotation in zip(table.rows, annotations, strict=True): writer.writerow( ( *row, *( _format_seconds(annotation[name]) for name in BOUNDARY_COLUMNS ), ) ) stream.flush() os.fsync(stream.fileno()) os.replace(temporary_path, output_csv) except OSError as error: if temporary_path is not None: temporary_path.unlink(missing_ok=True) raise MetadataError(f"cannot publish annotated CSV: {output_csv}") from error