File size: 8,975 Bytes
51e84d2 fc1dcbc | 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 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 | # 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
|