| |
| |
| |
| |
| |
| |
| |
|
|
| """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 |
|
|
