"""S1 — ingestion of YODAS utterances: listing, audio checks, S1/S2 signal measurements.

YODAS ships one 16 kHz wav per subtitle cue (``<video>-<seg>-<start_cs>-<end_cs>``), so a
YODAS utterance *is* a segment candidate: no long-file segmentation is needed. Every
utterance becomes either a manifest row or a quarantine record with a reason.
"""

from __future__ import annotations

import math
from collections.abc import Mapping
from dataclasses import dataclass, fields
from pathlib import Path
from typing import TYPE_CHECKING

from nida.bench.clips import YodasUtt, parse_yodas_utt_id
from nida.data import acoustic
from nida.data.types import QuarantineRecord, segment_id
from nida.text.turkish import nfc

if TYPE_CHECKING:
    import pyarrow as pa  # type: ignore[import-untyped]  # no stubs

__all__ = ["STAGE", "IngestConfig", "ingest_schema", "ingest_utterance"]

STAGE = "ingest"
SOURCE_ID = "yodas_tr"


@dataclass(frozen=True)
class IngestConfig:
    """configs/data/ingest.yaml (all required)."""

    expected_sample_rate_hz: int
    duration_tolerance_ms: int
    min_duration_ms: int
    max_duration_ms: int
    max_clipping_ratio: float
    max_silence_ratio: float

    def __post_init__(self) -> None:
        for name in (
            "expected_sample_rate_hz",
            "duration_tolerance_ms",
            "min_duration_ms",
            "max_duration_ms",
        ):
            value = getattr(self, name)
            if isinstance(value, bool) or not isinstance(value, int) or value < 0:
                raise ValueError(f"{name} must be a non-negative int, got {value!r}")
        if self.expected_sample_rate_hz == 0:
            raise ValueError("expected_sample_rate_hz must be positive")
        if not 0 < self.min_duration_ms < self.max_duration_ms:
            raise ValueError("need 0 < min_duration_ms < max_duration_ms")
        for name in ("max_clipping_ratio", "max_silence_ratio"):
            value = getattr(self, name)
            if isinstance(value, bool) or not isinstance(value, int | float):
                raise ValueError(f"{name} must be a number")
            if not (math.isfinite(value) and 0.0 <= value <= 1.0):
                raise ValueError(f"{name} must be in [0, 1], got {value!r}")

    @classmethod
    def from_mapping(cls, raw: Mapping[str, object]) -> IngestConfig:
        expected = {f.name for f in fields(cls)}
        if set(raw) != expected:
            raise ValueError(
                f"ingest config keys: missing={sorted(expected - set(raw))} "
                f"unknown={sorted(set(raw) - expected)}"
            )
        return cls(**raw)  # type: ignore[arg-type]  # values validated in __post_init__


def ingest_schema() -> pa.Schema:
    import pyarrow as pa

    from nida.data.manifest import schema_from_fields

    return schema_from_fields(
        [
            ("segment_id", pa.string()),
            ("source_id", pa.string()),
            ("utt_id", pa.string()),
            ("video_id", pa.string()),
            ("shard", pa.string()),
            ("caption", pa.string()),
            ("audio_relpath", pa.string()),
            ("duration_ms", pa.int64()),
            ("sample_rate_hz", pa.int64()),
            ("bandwidth_hz", pa.float64()),
            ("loudness_dbfs", pa.float64()),
            ("clipping_ratio", pa.float64()),
            ("silence_ratio", pa.float64()),
        ]
    )


def _drop(utt_id: str, reason: str, **details: object) -> QuarantineRecord:
    return QuarantineRecord(utt_id, STAGE, reason, dict(details))


def ingest_utterance(
    utt: YodasUtt,
    audio_root: Path,
    cfg: IngestConfig,
    acfg: acoustic.AcousticConfig,
) -> dict[str, object] | QuarantineRecord:
    """Read ``<audio_root>/<shard>/<utt_id>.wav``, check it and measure S1/S2 signals.

    Quarantine reasons: ``bad_utt_id``, ``unreadable_audio``, ``not_mono``,
    ``unexpected_sample_rate``, ``empty_audio``, ``non_finite_audio``, ``duration_mismatch``,
    ``too_short``, ``too_long``, ``silent``, ``clipping``, ``mostly_silence``.
    The quarantine ``item_id`` is the YODAS utterance id (the segment id needs a valid
    duration, which a broken file does not have).
    """
    import numpy as np
    import soundfile  # type: ignore[import-untyped]  # bench extra, untyped

    if any(bad in utt.utt_id for bad in ("/", "\\", "..", chr(0))):
        return _drop(utt.utt_id, "bad_utt_id", error="path characters in utterance id")
    try:
        video_id, start_ms, end_ms = parse_yodas_utt_id(utt.utt_id)
    except ValueError as err:
        return _drop(utt.utt_id, "bad_utt_id", error=str(err))
    if not utt.shard:
        raise ValueError(f"utterance {utt.utt_id} has no shard")
    relpath = f"{utt.shard}/{utt.utt_id}.wav"
    try:
        audio, rate_hz = soundfile.read(str(audio_root / relpath), dtype="float64", always_2d=True)
    except (RuntimeError, soundfile.LibsndfileError) as err:
        return _drop(utt.utt_id, "unreadable_audio", error=str(err))
    if audio.shape[1] != 1:
        return _drop(utt.utt_id, "not_mono", channels=int(audio.shape[1]))
    if rate_hz != cfg.expected_sample_rate_hz:
        return _drop(utt.utt_id, "unexpected_sample_rate", sample_rate_hz=int(rate_hz))
    mono = np.ascontiguousarray(audio[:, 0])
    if mono.shape[0] == 0:
        return _drop(utt.utt_id, "empty_audio")
    if not np.all(np.isfinite(mono)):
        return _drop(utt.utt_id, "non_finite_audio")
    duration_ms = round(1000 * mono.shape[0] / rate_hz)
    listed = [utt.duration_ms, end_ms - start_ms]
    if any(abs(duration_ms - d) > cfg.duration_tolerance_ms for d in listed):
        return _drop(utt.utt_id, "duration_mismatch", audio_ms=duration_ms, listed_ms=listed)
    if duration_ms < cfg.min_duration_ms:
        return _drop(utt.utt_id, "too_short", duration_ms=duration_ms)
    if duration_ms > cfg.max_duration_ms:
        return _drop(utt.utt_id, "too_long", duration_ms=duration_ms)
    try:
        bandwidth_hz = acoustic.estimate_bandwidth_hz(
            mono, int(rate_hz), floor_db=acfg.bandwidth_floor_db, params=acfg.bandwidth
        )
    except acoustic.SilentAudioError:
        return _drop(utt.utt_id, "silent")
    clipping = acoustic.clipping_ratio(mono, threshold=acfg.clipping_threshold)
    if clipping > cfg.max_clipping_ratio:
        return _drop(utt.utt_id, "clipping", clipping_ratio=clipping)
    silence = acoustic.silence_ratio(
        mono, int(rate_hz), frame_ms=acfg.silence_frame_ms, silence_dbfs=acfg.silence_dbfs
    )
    if silence > cfg.max_silence_ratio:
        return _drop(utt.utt_id, "mostly_silence", silence_ratio=silence)
    return {
        "segment_id": segment_id(SOURCE_ID, utt.utt_id, 0, duration_ms),
        "source_id": SOURCE_ID,
        "utt_id": utt.utt_id,
        "video_id": video_id,
        "shard": utt.shard,
        "caption": nfc(utt.caption),
        "audio_relpath": relpath,
        "duration_ms": duration_ms,
        "sample_rate_hz": int(rate_hz),
        "bandwidth_hz": float(bandwidth_hz),
        "loudness_dbfs": float(acoustic.loudness_dbfs(mono)),
        "clipping_ratio": float(clipping),
        "silence_ratio": float(silence),
    }