"""EARS -> latent dataset builder (ADR-0007, ``nida data-ears``).

EARS (Expressive Anechoic Recordings of Speech, CC BY-NC 4.0) provides English emotional,
whispered and non-verbal speech. This module turns the selected files into a *latent dataset
directory* in exactly the format of :mod:`nida.train.data` (records carry the optional key
``"language": "en"``), so it can be trained together with Turkish datasets
(``train --data A --data B``).

What is used (rules and mappings live in ``configs/data/ears.yaml``):

* ``emo_<emotion>_sentences``: reading transcript, emotion mapped (unmapped -> ``null``);
* ``rainbow_NN_<style>`` / ``sentences_NN_<style>``: ``whisper`` -> the whole text wrapped in a
  whisper span (rendered by :mod:`nida.text.tags`), the plain styles -> plain text;
* laughter/crying/screaming/coughing/throat-clearing files: just the event tag;
* everything else is skipped and counted by reason.

Per speaker (one remote zip of ~564 MB, read with HTTP Range requests, see
:mod:`nida.data.remote_zip`): list the archive -> decide on every member (skip / text
quarantine, before any download) -> extract the selected members, convert 48 kHz -> 16 kHz mono
-> measure. The speakers are handled in *waves* of ``speakers_per_wave``: one AudioVAE run
(the same ``vae_encode`` worker and environment as the pilot, so the model is loaded once per
wave, not once per speaker) encodes the wave's audio, then every speaker of the wave gets its
float16 shard, index records and ``done.json``. Layout::

    <root>/plan.json                       seed, config digest, VAE pin, transcripts SHA-256
    <root>/source/transcripts.json         pinned reading transcripts
    <root>/speakers/pNNN/records.jsonl     index rows of the speaker
    <root>/speakers/pNNN/quarantine.jsonl  every quarantined item with its reason
    <root>/speakers/pNNN/done.json         written last: the speaker is finished
    <root>/workers/vae/                    worker requests/results of the current wave
    <root>/dataset/                        latent dataset: index.jsonl, meta.json, shard-pNNN.npy
    <root>/report.md, <root>/stats.json
    <cache>/...                            raw members, 16 kHz wavs and latent arrays (local disk)

Resumable and idempotent: a speaker with ``done.json`` is skipped without any network access, an
interrupted wave continues (worker results are kept, already converted audio is reused), and
the dataset directory is reassembled from all finished speakers at the end of every run, so
it is usable after any run. A run on a root made with another seed, config digest or VAE pin,
or with different transcripts, is refused.
"""

from __future__ import annotations

import hashlib
import io
import json
import logging
import math
import re
import shutil
import string
import time
from bisect import bisect_right
from collections import Counter
from collections.abc import Callable, Mapping, Sequence
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Literal

import numpy as np
import numpy.typing as npt

# scipy ships no type stubs; the ignore is scoped to this single import (as in acoustic.py).
from scipy import signal  # type: ignore[import-untyped]

from nida.bench import envs as bench_envs
from nida.bench import fetch
from nida.bench.b1 import MAX_ERROR_FRACTION, _stream_runner
from nida.bench.protocol import (
    WorkRequest,
    WorkResult,
    drop_results,
    load_results,
    read_summary,
    write_requests,
)
from nida.bench.registry import load_sources, require_approved
from nida.data import acoustic, envs
from nida.data.ingest import IngestConfig
from nida.data.pilot import DatasetConfig
from nida.data.quarantine import QuarantineLog
from nida.data.registry import DATA_MODELS
from nida.data.remote_zip import RemoteZip, ZipMember
from nida.data.splits import assign_splits
from nida.data.types import QuarantineRecord, segment_id
from nida.model.vocab import encode_text
from nida.text import tags
from nida.text.turkish import nfc
from nida.train.data import (
    EMOTIONS,
    REJECT_TEXT_TOO_LONG,
    SPLITS,
    DataFormatError,
    load_latent_dataset,
)
from nida.utils.io import atomic_write_bytes, atomic_write_json, atomic_write_text

__all__ = [
    "SOURCE_ID",
    "AudioFacts",
    "Ears",
    "EarsConfig",
    "EarsSettings",
    "ItemRejected",
    "PlannedItem",
    "Selection",
    "SelectionPlan",
    "Skip",
    "TextRejected",
    "bucket_of",
    "build_text",
    "classify_stem",
    "convert_to_target_rate",
    "en_lower",
    "load_settings",
    "measure_audio",
    "normalise_english",
    "parse_speakers",
    "rate_chars_per_s",
    "resample",
    "select_members",
    "speaker_split",
]

log = logging.getLogger(__name__)

SOURCE_ID = "ears"
STAGE = "ears"
LANGUAGE = "en"
VAE_KEY = "voxcpm2_audiovae_v2"
DEFAULT_SEED = 20261009
FLOAT16_MAX_ABS = 6.0e4  # latents beyond this would overflow the float16 shards (as in pilot.py)
CONFIG_FILE = "ears.yaml"
INGEST_FILE = "ingest.yaml"  # duration limits shared with the Turkish pilot
DATASET_FILE = "dataset.yaml"  # fixed condition-bucket edges
_ASCII_LOWER = str.maketrans(string.ascii_uppercase, string.ascii_lowercase)
_SHA256_RE = re.compile(r"[0-9a-f]{64}")
_SPEAKER_RE = re.compile(r"p(\d{3})")
_WORD_RE = re.compile(r"[a-z]+")
_STEM_RE = re.compile(r"[a-z0-9_]+")
_EMO_RE = re.compile(r"emo_([a-z]+)_(sentences|freeform)")
_READ_RE = re.compile(r"([a-z]+)_(\d+)_([a-z]+)")
_TAG_CHARS = frozenset("[]<>")

Kind = Literal["emotion", "read", "whisper", "event"]
FloatArray = npt.NDArray[np.float64]


# --------------------------------------------------------------------------------------
# Config
# --------------------------------------------------------------------------------------


def _load_yaml(path: Path) -> Mapping[str, object]:
    import yaml  # type: ignore[import-untyped]  # PyYAML: bench extra, no stubs installed

    obj = yaml.safe_load(path.read_text(encoding="utf-8"))
    if not isinstance(obj, dict):
        raise ValueError(f"{path}: expected a mapping")
    return obj


def _non_negative_int(raw: Mapping[str, object], key: str) -> int:
    value = raw[key]
    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
        raise ValueError(f"{key} must be a non-negative int, got {value!r}")
    return value


def _positive_int(raw: Mapping[str, object], key: str) -> int:
    value = raw[key]
    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
        raise ValueError(f"{key} must be a positive int, got {value!r}")
    return value


def _fraction(raw: Mapping[str, object], key: str) -> float:
    value = raw[key]
    if isinstance(value, bool) or not isinstance(value, int | float):
        raise ValueError(f"{key} must be a number, got {value!r}")
    if not (math.isfinite(value) and 0.0 <= value < 1.0):
        raise ValueError(f"{key} must be in [0, 1), got {value!r}")
    return float(value)


def _words(raw: Mapping[str, object], key: str) -> tuple[str, ...]:
    value = raw[key]
    if (
        not isinstance(value, list)
        or not value
        or not all(isinstance(x, str) and _WORD_RE.fullmatch(x) for x in value)
        or len(set(value)) != len(value)
    ):
        raise ValueError(f"{key} must be a non-empty list of distinct lowercase words")
    return tuple(value)


def _str_map(raw: Mapping[str, object], key: str) -> dict[str, str]:
    value = raw[key]
    if not isinstance(value, dict) or not all(
        isinstance(k, str) and isinstance(v, str) for k, v in value.items()
    ):
        raise ValueError(f"{key} must be a mapping of strings")
    return dict(value)


@dataclass(frozen=True)
class EarsConfig:
    """``configs/data/ears.yaml`` (all keys required; see the file for the meaning)."""

    transcripts_url: str
    transcripts_sha256: str | None
    archive_url_template: str
    n_speakers: int
    expected_sample_rate_hz: int
    target_sample_rate_hz: int
    val_speaker_fraction: float
    test_speaker_fraction: float
    speakers_per_wave: int
    read_prefixes: tuple[str, ...]
    whisper_styles: tuple[str, ...]
    plain_styles: tuple[str, ...]
    max_read_per_speaker: int
    emotion_map: dict[str, str]  # plain dicts (not frozen views): ``asdict`` feeds the digest
    event_map: dict[str, str]
    char_map: dict[str, str]

    def __post_init__(self) -> None:
        for name in ("transcripts_url", "archive_url_template"):
            if not getattr(self, name).lower().startswith(("http://", "https://")):
                raise ValueError(f"{name} must be an http(s) URL")
        if "{speaker}" not in self.archive_url_template:
            raise ValueError("archive_url_template must contain {speaker}")
        if self.transcripts_sha256 is not None and not _SHA256_RE.fullmatch(
            self.transcripts_sha256
        ):
            raise ValueError("transcripts_sha256 must be null or 64 lowercase hex characters")
        if self.val_speaker_fraction + self.test_speaker_fraction >= 1.0:
            raise ValueError("val and test speaker fractions must leave room for train")
        if set(self.whisper_styles) & set(self.plain_styles):
            raise ValueError("a style cannot be both whisper and plain")
        for emotion, mapped in self.emotion_map.items():
            if not _WORD_RE.fullmatch(emotion) or mapped not in EMOTIONS:
                raise ValueError(f"emotion_map {emotion!r}: {mapped!r} is not one of {EMOTIONS}")
        for stem, event in self.event_map.items():
            if not _STEM_RE.fullmatch(stem) or event not in tags.EVENTS:
                raise ValueError(f"event_map {stem!r}: {event!r} is not a tag-spec event id")
        for char, replacement in self.char_map.items():
            if len(char) != 1 or not replacement:
                raise ValueError(f"char_map entries map one character to a string: {char!r}")

    @classmethod
    def from_mapping(cls, raw: Mapping[str, object]) -> EarsConfig:
        expected = set(cls.__dataclass_fields__)
        if set(raw) != expected:
            raise ValueError(
                f"ears config keys: missing={sorted(expected - set(raw))} "
                f"unknown={sorted(set(raw) - expected)}"
            )
        for key in ("transcripts_url", "archive_url_template"):
            if not isinstance(raw[key], str):
                raise ValueError(f"{key} must be a string")
        sha = raw["transcripts_sha256"]
        if sha is not None and not isinstance(sha, str):
            raise ValueError("transcripts_sha256 must be null or a string")
        return cls(
            transcripts_url=str(raw["transcripts_url"]),
            transcripts_sha256=sha,
            archive_url_template=str(raw["archive_url_template"]),
            n_speakers=_positive_int(raw, "n_speakers"),
            expected_sample_rate_hz=_positive_int(raw, "expected_sample_rate_hz"),
            target_sample_rate_hz=_positive_int(raw, "target_sample_rate_hz"),
            val_speaker_fraction=_fraction(raw, "val_speaker_fraction"),
            test_speaker_fraction=_fraction(raw, "test_speaker_fraction"),
            speakers_per_wave=_positive_int(raw, "speakers_per_wave"),
            read_prefixes=_words(raw, "read_prefixes"),
            whisper_styles=_words(raw, "whisper_styles"),
            plain_styles=_words(raw, "plain_styles"),
            max_read_per_speaker=_non_negative_int(raw, "max_read_per_speaker"),
            emotion_map=_str_map(raw, "emotion_map"),
            event_map=_str_map(raw, "event_map"),
            char_map=_str_map(raw, "char_map"),
        )

    def archive_url(self, speaker: str) -> str:
        return self.archive_url_template.replace("{speaker}", speaker)


@dataclass(frozen=True)
class EarsSettings:
    """Everything the builder reads from ``configs/data``."""

    ears: EarsConfig
    min_duration_ms: int
    max_duration_ms: int
    dataset: DatasetConfig
    digest: str


def load_settings(config_dir: Path) -> EarsSettings:
    """Load ``ears.yaml`` plus the duration limits (``ingest.yaml``) and bucket edges
    (``dataset.yaml``) it shares with the Turkish pilot.

    ``digest`` names everything a finished speaker depends on; it ignores
    ``transcripts_sha256`` so that pinning the hash in the config later is allowed.
    """
    ears = EarsConfig.from_mapping(_load_yaml(config_dir / CONFIG_FILE))
    ingest_cfg = IngestConfig.from_mapping(_load_yaml(config_dir / INGEST_FILE))
    dataset = DatasetConfig.from_mapping(_load_yaml(config_dir / DATASET_FILE))
    body = asdict(ears)
    del body["transcripts_sha256"]
    del body["speakers_per_wave"]  # operational: does not change any output
    canonical = json.dumps(
        {
            "ears": body,
            "min_duration_ms": ingest_cfg.min_duration_ms,
            "max_duration_ms": ingest_cfg.max_duration_ms,
            "dataset": asdict(dataset),
        },
        sort_keys=True,
        ensure_ascii=False,
    )
    return EarsSettings(
        ears=ears,
        min_duration_ms=ingest_cfg.min_duration_ms,
        max_duration_ms=ingest_cfg.max_duration_ms,
        dataset=dataset,
        digest=hashlib.sha1(canonical.encode("utf-8")).hexdigest()[:12],
    )


def parse_speakers(spec: str, n_speakers: int) -> list[str]:
    """``"all"`` or ``"p001,p007"`` -> speaker ids.

    Raises:
        ValueError: on an empty list, a malformed or out-of-range id, or a repeated id.
    """
    if spec.strip() == "all":
        return [f"p{i:03d}" for i in range(1, n_speakers + 1)]
    ids = [part.strip() for part in spec.split(",")]
    for speaker in ids:
        match = _SPEAKER_RE.fullmatch(speaker)
        if match is None or not 1 <= int(match.group(1)) <= n_speakers:
            raise ValueError(f"invalid speaker {speaker!r}: use p001..p{n_speakers:03d} or 'all'")
    if len(set(ids)) != len(ids):
        raise ValueError(f"repeated speaker in {spec!r}")
    return ids


# --------------------------------------------------------------------------------------
# Text
# --------------------------------------------------------------------------------------


class TextRejected(ValueError):
    """The text of an item cannot be used; ``reason`` is the quarantine reason."""

    def __init__(self, reason: str, detail: str) -> None:
        super().__init__(f"{reason}: {detail}")
        self.reason = reason
        self.detail = detail


def en_lower(text: str) -> str:
    """Lower-case ``A``-``Z`` only (English). Never use ``tr_lower`` (``I`` -> ``ı``) or
    ``str.lower`` (``İ`` -> ``i`` + combining dot) on English text: every other character is
    returned unchanged."""
    return text.translate(_ASCII_LOWER)


def normalise_english(text: str, char_map: Mapping[str, str]) -> str:
    """NFC, ``char_map`` replacements, ASCII lower-case and collapsed whitespace.

    Returns ``""`` for text without any non-space character.
    """
    mapped = nfc(text).translate({ord(k): v for k, v in char_map.items()})
    return " ".join(en_lower(mapped).split())


@dataclass(frozen=True)
class BuiltText:
    """Model text (tag-spec surface) and the plain words the speaker says (may be empty)."""

    tagged: str
    plain: str


def build_text(selection: Selection, transcript: str | None, cfg: EarsConfig) -> BuiltText:
    """Build the training text of one selected file.

    Raises:
        TextRejected: with reason ``no_transcript``, ``empty_text``, ``digits``,
            ``tag_characters``, ``non_ascii_letter`` or ``text_encoding`` (the model
            vocabulary cannot encode the result).
    """
    if selection.kind == "event":
        assert selection.event is not None  # classify_stem always sets it for events
        tagged = tags.render([tags.Event(selection.event)])
        plain = ""
    else:
        if transcript is None:
            raise TextRejected("no_transcript", f"no entry for {selection.stem!r}")
        plain = normalise_english(transcript, cfg.char_map)
        if not plain:
            raise TextRejected("empty_text", "transcript has no characters")
        if any(ch.isdigit() for ch in plain):
            raise TextRejected("digits", plain[:80])
        if any(ch in _TAG_CHARS for ch in plain):
            raise TextRejected("tag_characters", plain[:80])
        if any(ch.isalpha() and not ch.isascii() for ch in plain):
            raise TextRejected("non_ascii_letter", plain[:80])
        if selection.kind == "whisper":
            try:
                span = tags.Span("whisper", tuple(tags.Word(w) for w in plain.split()))
            except ValueError as err:
                raise TextRejected("text_encoding", str(err)) from err
            tagged = tags.render([span])
        else:
            tagged = plain
    try:
        ids = encode_text(tagged)
    except ValueError as err:  # unknown character, bad tag
        raise TextRejected("text_encoding", str(err)) from err
    if not ids:
        raise TextRejected("empty_text", "text encodes to no tokens")
    return BuiltText(tagged=tagged, plain=plain)


# --------------------------------------------------------------------------------------
# Selection
# --------------------------------------------------------------------------------------


@dataclass(frozen=True)
class Selection:
    """A file that is used. ``emotion`` is set for ``emotion`` items (may be ``None``),
    ``event`` for ``event`` items."""

    stem: str
    kind: Kind
    emotion: str | None = None
    event: str | None = None

    @property
    def label(self) -> str:
        """Report category: ``emotion``, ``read``, ``whisper`` or ``event:<id>``."""
        return f"event:{self.event}" if self.kind == "event" else self.kind


@dataclass(frozen=True)
class Skip:
    """A file that is not used, with the counted reason."""

    stem: str
    reason: str


def classify_stem(stem: str, cfg: EarsConfig) -> Selection | Skip:
    """Decide on one file stem (ADR-0007 table). Pure; thresholds and maps come from ``cfg``.

    Skip reasons: ``freeform``, ``interjection``, ``melodic``, ``event_unmapped``,
    ``style_unsupported``, ``unknown_stem``.
    """
    emo = _EMO_RE.fullmatch(stem)
    if emo is not None:
        if emo.group(2) != "sentences":
            return Skip(stem, "freeform")
        return Selection(stem, "emotion", emotion=cfg.emotion_map.get(emo.group(1)))
    event = cfg.event_map.get(stem)
    if event is not None:
        return Selection(stem, "event", event=event)
    read = _READ_RE.fullmatch(stem)
    if read is not None and read.group(1) in cfg.read_prefixes:
        style = read.group(3)
        if style in cfg.whisper_styles:
            return Selection(stem, "whisper")
        if style in cfg.plain_styles:
            return Selection(stem, "read")
        return Skip(stem, "style_unsupported")
    if stem.startswith("freeform_speech"):
        return Skip(stem, "freeform")
    for prefix, reason in (
        ("interjection_", "interjection"),
        ("melodic_", "melodic"),
        ("nonverbal_", "event_unmapped"),
        ("vegetative_", "event_unmapped"),
    ):
        if stem.startswith(prefix):
            return Skip(stem, reason)
    return Skip(stem, "unknown_stem")


@dataclass(frozen=True)
class PlannedItem:
    """A selected archive member with its final text."""

    stem: str
    member: ZipMember
    selection: Selection
    text: BuiltText


@dataclass
class SelectionPlan:
    """Result of :func:`select_members`: nothing is lost, every member is in one of three places."""

    items: list[PlannedItem] = field(default_factory=list)
    skipped: Counter[str] = field(default_factory=Counter)
    quarantine: list[QuarantineRecord] = field(default_factory=list)


def select_members(
    speaker: str,
    members: Sequence[ZipMember],
    transcripts: Mapping[str, str],
    cfg: EarsConfig,
) -> SelectionPlan:
    """Decide on every archive member before anything is downloaded.

    ``items`` come back in archive order (ascending ``header_offset``) so extraction profits
    from the read-ahead window. Skipped members are counted by reason; text problems become
    quarantine records (stage ``ears``, item id ``<speaker>/<stem>``).
    """
    plan = SelectionPlan()
    seen: set[str] = set()
    for member in sorted(members, key=lambda m: m.header_offset):
        if member.is_dir:
            plan.skipped["directory"] += 1
            continue
        parts = member.name.split("/")
        if len(parts) != 2 or parts[0] != speaker:
            plan.skipped["outside_speaker_folder"] += 1
            continue
        if not parts[1].endswith(".wav"):
            plan.skipped["not_wav"] += 1
            continue
        stem = parts[1][: -len(".wav")]
        decision = classify_stem(stem, cfg)
        if isinstance(decision, Skip):
            plan.skipped[decision.reason] += 1
            continue
        item_id = f"{speaker}/{stem}"
        if stem in seen:
            plan.quarantine.append(
                QuarantineRecord(item_id, STAGE, "duplicate_member", {"name": member.name})
            )
            continue
        seen.add(stem)
        try:
            text = build_text(decision, transcripts.get(stem), cfg)
        except TextRejected as err:
            plan.quarantine.append(
                QuarantineRecord(item_id, STAGE, err.reason, {"detail": err.detail})
            )
            continue
        plan.items.append(PlannedItem(stem, member, decision, text))
    _cap_reads(plan, speaker, cfg.max_read_per_speaker)
    return plan


def _cap_reads(plan: SelectionPlan, speaker: str, cap: int) -> None:
    """Keep at most ``cap`` plain reading items per speaker (0 = no limit; ADR-0007: plain
    reading is a low share). The kept ones are a deterministic hash choice of the stems;
    the rest are counted as skipped ``read_cap``."""
    reads = [item for item in plan.items if item.selection.kind == "read"]
    if cap == 0 or len(reads) <= cap:
        return
    ranked = sorted(reads, key=lambda it: hashlib.sha1(f"{speaker}/{it.stem}".encode()).hexdigest())
    dropped = {id(item) for item in ranked[cap:]}
    plan.items = [item for item in plan.items if id(item) not in dropped]
    plan.skipped["read_cap"] += len(dropped)


# --------------------------------------------------------------------------------------
# Audio
# --------------------------------------------------------------------------------------


class ItemRejected(Exception):
    """An audio file cannot be used; ``reason`` is the quarantine reason."""

    def __init__(self, reason: str, **details: object) -> None:
        super().__init__(f"{reason}: {details}")
        self.reason = reason
        self.details = details


@dataclass(frozen=True)
class AudioFacts:
    """Measurements of a converted file: ``duration_ms`` of the 16 kHz audio, RMS level."""

    duration_ms: int
    loudness_dbfs: float


def resample(audio: FloatArray, rate_hz: int, target_hz: int) -> FloatArray:
    """Polyphase resampling of a mono signal (anti-aliased) from ``rate_hz`` to ``target_hz``.

    Args:
        audio: ``(num_samples,)`` float signal.
        rate_hz: Source rate.
        target_hz: Target rate.

    Returns:
        ``(ceil(num_samples * target_hz / rate_hz),)`` float64 signal.
    """
    if rate_hz == target_hz:
        return np.asarray(audio, dtype=np.float64)
    divisor = math.gcd(rate_hz, target_hz)
    out = signal.resample_poly(audio, target_hz // divisor, rate_hz // divisor)
    return np.asarray(out, dtype=np.float64)


def convert_to_target_rate(raw: Path, dest: Path, cfg: EarsConfig) -> None:
    """Read a raw EARS wav, check it, resample to ``target_sample_rate_hz`` mono and write a
    32-bit float wav atomically to ``dest``.

    Raises:
        ItemRejected: ``unreadable_audio``, ``not_mono``, ``unexpected_sample_rate``,
            ``empty_audio`` or ``non_finite_audio``.
    """
    import soundfile  # type: ignore[import-untyped]  # bench extra, untyped

    try:
        audio, rate_hz = soundfile.read(str(raw), dtype="float64", always_2d=True)
    except (RuntimeError, soundfile.LibsndfileError) as err:
        raise ItemRejected("unreadable_audio", error=str(err)) from err
    if audio.shape[1] != 1:
        raise ItemRejected("not_mono", channels=int(audio.shape[1]))
    if rate_hz != cfg.expected_sample_rate_hz:
        raise ItemRejected("unexpected_sample_rate", sample_rate_hz=int(rate_hz))
    mono = np.ascontiguousarray(audio[:, 0])
    if mono.shape[0] == 0:
        raise ItemRejected("empty_audio")
    if not np.all(np.isfinite(mono)):
        raise ItemRejected("non_finite_audio")
    converted = resample(mono, int(rate_hz), cfg.target_sample_rate_hz)
    buffer = io.BytesIO()
    soundfile.write(buffer, converted, cfg.target_sample_rate_hz, format="WAV", subtype="FLOAT")
    atomic_write_bytes(dest, buffer.getvalue())


def measure_audio(
    path: Path, cfg: EarsConfig, min_duration_ms: int, max_duration_ms: int
) -> AudioFacts:
    """Duration and loudness of a converted wav; enforces the duration limits.

    The loudness is :func:`nida.data.acoustic.loudness_dbfs`, the measure the pilot ingest
    stores, here on the 16 kHz signal the VAE hears.

    Raises:
        ItemRejected: ``unreadable_audio``, ``too_short``, ``too_long`` or ``silent``.
        ValueError: if the file is not at ``target_sample_rate_hz`` (the cache is inconsistent
            with the config).
    """
    import soundfile

    try:
        audio, rate_hz = soundfile.read(str(path), dtype="float64", always_2d=True)
    except (RuntimeError, soundfile.LibsndfileError) as err:
        raise ItemRejected("unreadable_audio", error=str(err)) from err
    if rate_hz != cfg.target_sample_rate_hz or audio.shape[1] != 1:
        raise ValueError(
            f"{path}: expected {cfg.target_sample_rate_hz} Hz mono, got {rate_hz} Hz x "
            f"{audio.shape[1]}; delete the cache folder"
        )
    mono = np.ascontiguousarray(audio[:, 0])
    duration_ms = round(1000 * mono.shape[0] / rate_hz)
    if duration_ms < min_duration_ms:
        raise ItemRejected("too_short", duration_ms=duration_ms)
    if duration_ms > max_duration_ms:
        raise ItemRejected("too_long", duration_ms=duration_ms)
    if float(np.max(np.abs(mono))) == 0.0:
        raise ItemRejected("silent")
    return AudioFacts(duration_ms=duration_ms, loudness_dbfs=float(acoustic.loudness_dbfs(mono)))


# --------------------------------------------------------------------------------------
# Buckets and splits
# --------------------------------------------------------------------------------------


def bucket_of(value: float, edges: Sequence[float]) -> int:
    """Bucket index of ``value``: the number of ``edges`` that are ``<= value``
    (``numpy.searchsorted(edges, value, side="right")``, as the pilot dataset stage)."""
    return bisect_right(edges, value)


def rate_chars_per_s(plain_text: str, duration_ms: int) -> float:
    """Alphanumeric characters of the plain text per second of audio (pilot formula)."""
    return sum(ch.isalnum() for ch in plain_text) / (duration_ms / 1000)


def speaker_split(group: str, val_fraction: float, test_fraction: float, seed: int) -> str:
    """Split of a speaker group: a pure function of ``(seed, group)`` (:mod:`nida.data.splits`)."""
    fractions = {
        "train": 1.0 - val_fraction - test_fraction,
        "val": val_fraction,
        "test": test_fraction,
    }
    return assign_splits({"speaker": group}, fractions, str(seed))["speaker"]


# --------------------------------------------------------------------------------------
# The builder
# --------------------------------------------------------------------------------------


@dataclass(frozen=True)
class _Measured:
    item: PlannedItem
    segment_id: str
    wav: Path
    facts: AudioFacts


@dataclass
class _Prepared:
    """A speaker whose audio is ready for the VAE: nothing is written for it yet."""

    plan: SelectionPlan
    quarantine: list[QuarantineRecord]
    measured: list[_Measured]


@dataclass
class Ears:
    """EARS -> latent dataset builder (see the module docstring).

    Attributes:
        root: Durable directory (Drive on Colab).
        cache: Local scratch for raw members, 16 kHz audio and latent arrays.
        envs_root: Where the worker virtual environments live.
        repo_dir: Repository root (environment builds need it).
        sources_path: ``configs/sources.yaml``.
        config_dir: ``configs/data``.
        seed: Split seed (recorded in ``plan.json``).
        runner: Worker command runner (injectable).
        opener: Network access (injectable).
        sleep: Backoff sleep for retries (injectable).
    """

    root: Path
    cache: Path
    envs_root: Path
    repo_dir: Path
    sources_path: Path
    config_dir: Path
    seed: int = DEFAULT_SEED
    uv: str | None = None
    runner: Callable[[Sequence[str]], int] = _stream_runner
    opener: fetch.Opener = fetch.default_opener
    sleep: Callable[[float], None] = time.sleep
    _python: Path | None = field(default=None, init=False, repr=False)

    # ---------------------------------------------------------------- paths

    @property
    def dataset_dir(self) -> Path:
        return self.root / "dataset"

    def _speaker_dir(self, speaker: str) -> Path:
        return self.root / "speakers" / speaker

    def _npy(self, segment: str) -> Path:
        return self.cache / "arrays" / "vae" / f"{segment}.npy"

    # ---------------------------------------------------------------- run

    def run(self, speakers: Sequence[str]) -> dict[str, object]:
        """Build (or continue) the dataset for ``speakers``.

        Finished speakers are skipped. The dataset directory, ``report.md`` and
        ``stats.json`` are rewritten from *all* finished speakers of the root at the end.

        Returns:
            Summary with ``dataset``, ``speakers_done``, ``segments``, ``hours``,
            ``skipped`` and ``quarantined`` counts, and ``failed_speakers``.

        Raises:
            SourceNotApprovedError: before any download, if a source is not approved.
            ValueError: on a plan/config/transcripts mismatch with the root.
            RuntimeError: if a worker fails, a speaker could not be fetched (after all other
                speakers were processed and the dataset was assembled), or nothing was built.
        """
        settings = load_settings(self.config_dir)
        model = DATA_MODELS[VAE_KEY]
        require_approved({SOURCE_ID, model.source_id}, load_sources(self.sources_path))
        wanted = list(speakers)
        for speaker in wanted:
            parse_speakers(speaker, settings.ears.n_speakers)
        transcripts, transcripts_sha = self._prepare_plan(settings)
        failed: dict[str, str] = {}
        pending: list[str] = []
        for speaker in dict.fromkeys(wanted):
            if (self._speaker_dir(speaker) / "done.json").is_file():
                log.info("EARS %s already done", speaker)
            else:
                pending.append(speaker)
        wave_size = settings.ears.speakers_per_wave
        for start in range(0, len(pending), wave_size):
            wave = pending[start : start + wave_size]
            prepared: dict[str, _Prepared] = {}
            for speaker in wave:
                log.info("EARS %s (%d/%d)", speaker, pending.index(speaker) + 1, len(pending))
                try:
                    prepared[speaker] = self._prepare_speaker(speaker, settings, transcripts)
                except fetch.FetchError as err:  # network or archive problem: others still run
                    log.error("EARS %s failed: %s", speaker, err)
                    failed[speaker] = str(err)
            results = self._encode([m for entry in prepared.values() for m in entry.measured])
            for speaker, entry in prepared.items():
                self._finish_speaker(speaker, settings, entry, results)
        summary = self._assemble(settings, transcripts_sha, failed)
        if failed:
            raise RuntimeError(
                f"{len(failed)} speaker(s) failed and can be retried by running again: "
                + "; ".join(f"{s}: {e[:200]}" for s, e in sorted(failed.items()))
            )
        return summary

    # ---------------------------------------------------------------- plan and transcripts

    def _prepare_plan(self, settings: EarsSettings) -> tuple[dict[str, str], str]:
        """Check/write ``plan.json`` and fetch the pinned transcripts."""
        model = DATA_MODELS[VAE_KEY]
        core: dict[str, object] = {
            "seed": self.seed,
            "config_digest": settings.digest,
            "archive_url_template": settings.ears.archive_url_template,
            "transcripts_url": settings.ears.transcripts_url,
            "vae": [model.ref.model_id, model.ref.revision],
        }
        plan_path = self.root / "plan.json"
        previous: dict[str, object] | None = None
        if plan_path.is_file():
            loaded = json.loads(plan_path.read_text(encoding="utf-8"))
            previous = dict(loaded)
            recorded_sha = previous.pop("transcripts_sha256", None)
            if previous != core:
                raise ValueError(
                    f"{plan_path} was written for a different plan: {previous} != {core}"
                )
            if (
                settings.ears.transcripts_sha256 is not None
                and recorded_sha is not None
                and settings.ears.transcripts_sha256 != recorded_sha
            ):
                raise ValueError(
                    f"config pins transcripts {settings.ears.transcripts_sha256} but the root "
                    f"was built with {recorded_sha}"
                )
            expected = recorded_sha if isinstance(recorded_sha, str) else None
        else:
            expected = None
        expected = settings.ears.transcripts_sha256 or expected
        downloaded = fetch.download(
            settings.ears.transcripts_url,
            self.root / "source" / "transcripts.json",
            expected_sha256=expected,
            opener=self.opener,
            sleep=self.sleep,
        )
        if previous is None:
            atomic_write_json(plan_path, {**core, "transcripts_sha256": downloaded.sha256})
        return _parse_transcripts(downloaded.path), downloaded.sha256

    # ---------------------------------------------------------------- speaker

    def _python_exe(self) -> Path:
        if self._python is None:
            if self.uv is None:
                self.uv = bench_envs.find_uv(bench_envs.subprocess_runner)
            spec = envs.data_env_for_worker(DATA_MODELS[VAE_KEY].worker)
            self._python = bench_envs.ensure_env(spec, self.envs_root, self.repo_dir, uv=self.uv)
        return self._python

    def _prepare_speaker(
        self, speaker: str, settings: EarsSettings, transcripts: Mapping[str, str]
    ) -> _Prepared:
        """List the archive, decide on every member, fetch + convert + measure the selected."""
        cfg = settings.ears
        QuarantineLog(self._speaker_dir(speaker)).path.unlink(missing_ok=True)  # rebuilt below
        archive = RemoteZip(cfg.archive_url(speaker), opener=self.opener, sleep=self.sleep)
        members = archive.members()
        plan = select_members(speaker, members, transcripts, cfg)
        quarantine = list(plan.quarantine)
        log.info(
            "EARS %s: %d members, %d selected, %d quarantined, skipped %s",
            speaker,
            len(members),
            len(plan.items),
            len(quarantine),
            dict(plan.skipped),
        )
        measured: list[_Measured] = []
        for item in plan.items:  # archive order: one read-ahead window serves neighbours
            item_id = f"{speaker}/{item.stem}"
            wav = self.cache / "audio16k" / speaker / f"{item.stem}.wav"
            try:
                if not wav.is_file():
                    raw = archive.extract(
                        item.member, self.cache / "raw" / speaker / f"{item.stem}.wav"
                    )
                    convert_to_target_rate(raw, wav, cfg)
                    raw.unlink()
                facts = measure_audio(wav, cfg, settings.min_duration_ms, settings.max_duration_ms)
            except ItemRejected as err:
                quarantine.append(QuarantineRecord(item_id, STAGE, err.reason, dict(err.details)))
                continue
            sid = segment_id(SOURCE_ID, item_id, 0, facts.duration_ms)
            measured.append(_Measured(item, sid, wav, facts))
        return _Prepared(plan, quarantine, measured)

    def _finish_speaker(
        self,
        speaker: str,
        settings: EarsSettings,
        prepared: _Prepared,
        results: Mapping[str, WorkResult],
    ) -> None:
        """Write the speaker's outputs, then drop its scratch files."""
        self._write_speaker(
            speaker, settings, prepared.measured, results, prepared.plan, prepared.quarantine
        )
        for sub in ("raw", "audio16k"):
            shutil.rmtree(self.cache / sub / speaker, ignore_errors=True)
        for entry in prepared.measured:
            self._npy(entry.segment_id).unlink(missing_ok=True)

    def _encode(self, measured: Sequence[_Measured]) -> dict[str, WorkResult]:
        """Run the pilot's ``vae_encode`` worker once over the audio of a wave (resumable)."""
        if not measured:
            return {}
        model = DATA_MODELS[VAE_KEY]
        python = self._python_exe()
        out_dir = self.root / "workers" / "vae"
        results_path = out_dir / "results.jsonl"
        wanted = {m.segment_id for m in measured}
        summary_path = out_dir / "results.summary.json"
        if results_path.exists() and summary_path.exists():
            previous = read_summary(summary_path)
            if (previous.model_id, previous.revision) != (model.ref.model_id, model.ref.revision):
                raise RuntimeError(
                    f"existing latents come from {previous.model_id}@"
                    f"{previous.revision}, not {model.ref.model_id}@{model.ref.revision}; "
                    "move them away first"
                )
        if results_path.exists():
            existing = load_results(results_path, repair=True)
            stale = [
                cid
                for cid, res in existing.items()
                if cid not in wanted
                or res.output is None  # failed before: try again (a stuck item fails again)
                or not Path(str(res.output.get("out_path", ""))).is_file()
            ]
            if stale:
                log.info("dropping %d stray, failed or lost worker results", len(stale))
                drop_results(results_path, stale)
        requests = [
            WorkRequest(m.segment_id, str(m.wav), {"out_path": str(self._npy(m.segment_id))})
            for m in measured
        ]
        write_requests(out_dir / "requests.jsonl", requests)
        command = [
            str(python),
            "-m",
            envs.worker_module_name(model.worker),
            "--requests",
            str(out_dir / "requests.jsonl"),
            "--results",
            str(results_path),
            "--model-id",
            model.ref.model_id,
            "--revision",
            model.ref.revision,
            "--batch-size",
            str(model.batch_size),
            "--device",
            "cuda",
        ]
        code = self.runner(command)
        if code != 0:
            raise RuntimeError(f"vae_encode worker exited with code {code}")
        summary = read_summary(summary_path)
        total = summary.n_ok + summary.n_error
        if total != len(requests):
            raise RuntimeError(f"{total} latent results for {len(requests)} requests")
        if summary.n_error > MAX_ERROR_FRACTION * total:
            raise RuntimeError(f"{summary.n_error}/{total} items failed in the worker")
        return load_results(results_path)

    def _write_speaker(
        self,
        speaker: str,
        settings: EarsSettings,
        measured: Sequence[_Measured],
        results: Mapping[str, WorkResult],
        plan: SelectionPlan,
        quarantine: list[QuarantineRecord],
    ) -> dict[str, object]:
        """Shard, records, quarantine log and finally the done marker of one speaker."""
        cfg = settings.ears
        group = f"ears-{speaker}"
        split = speaker_split(group, cfg.val_speaker_fraction, cfg.test_speaker_fraction, self.seed)
        rate_edges = list(settings.dataset.rate_bucket_edges_chars_per_s)
        loud_edges = list(settings.dataset.loudness_bucket_edges_dbfs)
        shard_name = f"shard-{speaker}.npy"
        rows: list[dict[str, object]] = []
        arrays: list[npt.NDArray[np.float32]] = []
        kinds: dict[str, dict[str, int]] = {}
        latent_dim: int | None = None
        frame_hop_ms: float | None = None
        frames = 0
        for entry in sorted(measured, key=lambda m: m.item.stem):
            item_id = f"{speaker}/{entry.item.stem}"
            result = results.get(entry.segment_id)
            if result is None:
                raise RuntimeError(f"{speaker}: the worker returned no result for {item_id}")
            if result.output is None:
                quarantine.append(
                    QuarantineRecord(item_id, STAGE, "vae_error", {"error": str(result.error)})
                )
                continue
            array = np.load(str(result.output["out_path"]), allow_pickle=False)
            hop = float(str(result.output["frame_hop_ms"]))
            if array.ndim != 2 or array.shape[0] < 1:
                raise ValueError(f"{item_id}: latents must be 2-D with frames, got {array.shape}")
            if frame_hop_ms is not None and hop != frame_hop_ms:
                raise ValueError(f"{item_id}: frame hop {hop} != {frame_hop_ms}")
            if latent_dim is not None and int(array.shape[1]) != latent_dim:
                raise ValueError(f"{item_id}: latent_dim {array.shape[1]} != {latent_dim}")
            frame_hop_ms, latent_dim = hop, int(array.shape[1])
            if not np.all(np.isfinite(array)) or float(np.max(np.abs(array))) > FLOAT16_MAX_ABS:
                quarantine.append(QuarantineRecord(item_id, STAGE, "latents_not_float16", {}))
                continue
            num_frames = int(array.shape[0])
            n_tokens = len(encode_text(entry.item.text.tagged))
            if n_tokens > num_frames:
                quarantine.append(
                    QuarantineRecord(
                        item_id,
                        STAGE,
                        REJECT_TEXT_TOO_LONG,
                        {"text_tokens": n_tokens, "num_frames": num_frames},
                    )
                )
                continue
            plain = entry.item.text.plain
            rows.append(
                {
                    "segment_id": entry.segment_id,
                    "shard": shard_name,
                    "offset": frames,
                    "num_frames": num_frames,
                    "text": entry.item.text.tagged,
                    "emotion": entry.item.selection.emotion,
                    "intensity": None,
                    "rate_bucket": (
                        bucket_of(rate_chars_per_s(plain, entry.facts.duration_ms), rate_edges)
                        if plain
                        else None
                    ),
                    "loudness_bucket": bucket_of(entry.facts.loudness_dbfs, loud_edges),
                    "group": group,
                    "split": split,
                    "language": LANGUAGE,
                }
            )
            arrays.append(array.astype(np.float32))
            frames += num_frames
            stat = kinds.setdefault(entry.item.selection.label, {"segments": 0, "audio_ms": 0})
            stat["segments"] += 1
            stat["audio_ms"] += entry.facts.duration_ms

        spk_dir = self._speaker_dir(speaker)
        if arrays:
            buffer = io.BytesIO()
            np.save(buffer, np.concatenate(arrays).astype(np.float16), allow_pickle=False)
            atomic_write_bytes(self.dataset_dir / shard_name, buffer.getvalue())
        atomic_write_text(
            spk_dir / "records.jsonl",
            "".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows),
        )
        QuarantineLog(spk_dir).append(quarantine)
        marker: dict[str, object] = {
            "speaker": speaker,
            "group": group,
            "split": split,
            "segments": len(rows),
            "frames": frames,
            "audio_ms": sum(s["audio_ms"] for s in kinds.values()),
            "latent_dim": latent_dim if rows else None,
            "frame_hop_ms": frame_hop_ms if rows else None,
            "shard": shard_name if rows else None,
            "kinds": kinds,
            "skipped": dict(sorted(plan.skipped.items())),
            "quarantined": dict(sorted(Counter(q.reason for q in quarantine).items())),
        }
        atomic_write_json(spk_dir / "done.json", marker)  # written last: the speaker is done
        return marker

    # ---------------------------------------------------------------- assembly

    def _assemble(
        self, settings: EarsSettings, transcripts_sha: str, failed: Mapping[str, str]
    ) -> dict[str, object]:
        """Index, meta, report and stats from every finished speaker of the root."""
        markers: list[dict[str, object]] = []
        speakers_dir = self.root / "speakers"
        for done_path in sorted(speakers_dir.glob("p*/done.json")) if speakers_dir.is_dir() else []:
            markers.append(json.loads(done_path.read_text(encoding="utf-8")))
        with_data = [m for m in markers if m["segments"]]
        if not with_data:
            raise RuntimeError(
                f"no EARS segment built (speakers done: {len(markers)}, failed: {sorted(failed)})"
            )
        hops = {float(str(m["frame_hop_ms"])) for m in with_data}
        dims = {int(str(m["latent_dim"])) for m in with_data}
        if len(hops) != 1 or len(dims) != 1:
            raise ValueError(
                f"speakers disagree on frame hop {sorted(hops)} or latent_dim {sorted(dims)}"
            )
        index_lines: list[str] = []
        for marker in with_data:
            lines = (
                (self._speaker_dir(str(marker["speaker"])) / "records.jsonl")
                .read_text(encoding="utf-8")
                .splitlines(keepends=True)
            )
            if len(lines) != marker["segments"]:
                raise ValueError(f"{marker['speaker']}: records.jsonl and done.json disagree")
            if not (self.dataset_dir / str(marker["shard"])).is_file():
                raise ValueError(f"{marker['speaker']}: shard {marker['shard']} is missing")
            index_lines.extend(lines)
        total_segments = len(index_lines)
        total_frames = sum(int(str(m["frames"])) for m in with_data)
        atomic_write_text(self.dataset_dir / "index.jsonl", "".join(index_lines))
        source = load_sources(self.sources_path)[SOURCE_ID]
        model = DATA_MODELS[VAE_KEY]
        meta: dict[str, object] = {
            "segments": total_segments,
            "frames": total_frames,
            "latent_dim": dims.pop(),
            "frame_hop_ms": hops.pop(),
            "rate_bucket_edges_chars_per_s": list(settings.dataset.rate_bucket_edges_chars_per_s),
            "loudness_bucket_edges_dbfs": list(settings.dataset.loudness_bucket_edges_dbfs),
            "source": SOURCE_ID,
            "license": source.get("license"),
            "language": LANGUAGE,
            "transcripts_sha256": transcripts_sha,
            "transcripts_url": settings.ears.transcripts_url,
            "config_digest": settings.digest,
            "seed": self.seed,
            "vae": [model.ref.model_id, model.ref.revision],
            "speakers": {
                str(m["speaker"]): {"split": m["split"], "segments": m["segments"]} for m in markers
            },
        }
        atomic_write_json(self.dataset_dir / "meta.json", meta)

        loader = self._check_with_loader()
        stats = _stats(markers, loader, settings, failed, meta)
        atomic_write_json(self.root / "stats.json", stats)
        atomic_write_text(self.root / "report.md", _render_report(stats))
        rejected = {s: c for s, c in loader.items() if c["rejected"]}
        if rejected:
            raise RuntimeError(f"nida.train.data rejects records the builder accepted: {rejected}")
        return {
            "dataset": str(self.dataset_dir),
            "speakers_done": len(markers),
            "segments": total_segments,
            "hours": stats["hours"],
            "skipped": stats["skipped"],
            "quarantined": stats["quarantined"],
            "failed_speakers": dict(failed),
        }

    def _check_with_loader(self) -> dict[str, dict[str, object]]:
        """Load every split with the training loader; counts per split (must not reject)."""
        out: dict[str, dict[str, object]] = {}
        for split in SPLITS:
            try:
                dataset, report = load_latent_dataset(self.dataset_dir, split)
            except DataFormatError as err:
                raise RuntimeError(f"the assembled dataset is invalid: {err}") from err
            dataset.close()
            out[split] = {
                "records": report.num_in_split,
                "accepted": report.num_accepted,
                "rejected": report.num_rejected,
                "rejection_counts": dict(report.rejection_counts),
            }
        return out


# --------------------------------------------------------------------------------------
# Reporting
# --------------------------------------------------------------------------------------


def _parse_transcripts(path: Path) -> dict[str, str]:
    obj = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(obj, dict) or not obj:
        raise ValueError(f"{path}: expected a non-empty mapping stem -> text")
    bad = [k for k, v in obj.items() if not isinstance(k, str) or not isinstance(v, str)]
    if bad:
        raise ValueError(f"{path}: entries must map strings to strings, e.g. {bad[:3]}")
    return dict(obj)


def _sum_counts(markers: Sequence[Mapping[str, object]], key: str) -> dict[str, int]:
    total: Counter[str] = Counter()
    for marker in markers:
        value = marker[key]
        assert isinstance(value, dict)
        total.update({str(k): int(v) for k, v in value.items()})
    return dict(sorted(total.items(), key=lambda kv: (-kv[1], kv[0])))


def _stats(
    markers: Sequence[Mapping[str, object]],
    loader: Mapping[str, Mapping[str, object]],
    settings: EarsSettings,
    failed: Mapping[str, str],
    meta: Mapping[str, object],
) -> dict[str, object]:
    kinds: dict[str, dict[str, int]] = {}
    for marker in markers:
        value = marker["kinds"]
        assert isinstance(value, dict)
        for label, stat in value.items():
            into = kinds.setdefault(label, {"segments": 0, "audio_ms": 0})
            into["segments"] += int(stat["segments"])
            into["audio_ms"] += int(stat["audio_ms"])
    splits: dict[str, dict[str, int]] = {s: {"speakers": 0, "segments": 0} for s in SPLITS}
    for marker in markers:
        entry = splits[str(marker["split"])]
        entry["speakers"] += 1
        entry["segments"] += int(str(marker["segments"]))
    audio_ms = sum(int(str(m["audio_ms"])) for m in markers)
    return {
        "speakers_done": len(markers),
        "speakers_total": settings.ears.n_speakers,
        "segments": meta["segments"],
        "frames": meta["frames"],
        "hours": round(audio_ms / 3_600_000, 3),
        "kinds": dict(sorted(kinds.items())),
        "splits": splits,
        "skipped": _sum_counts(markers, "skipped"),
        "quarantined": _sum_counts(markers, "quarantined"),
        "loader": {k: dict(v) for k, v in loader.items()},
        "failed_speakers": dict(failed),
        "transcripts_sha256": meta["transcripts_sha256"],
        "config_digest": settings.digest,
    }


def _render_report(stats: Mapping[str, object]) -> str:
    kinds = stats["kinds"]
    splits = stats["splits"]
    loader = stats["loader"]
    assert isinstance(kinds, dict) and isinstance(splits, dict) and isinstance(loader, dict)
    lines = [
        "# EARS latent veri seti raporu",
        "",
        f"- Konuşmacı: {stats['speakers_done']}/{stats['speakers_total']} tamamlandı",
        f"- Segment: {stats['segments']}, süre: {stats['hours']} sa, kare: {stats['frames']}",
        f"- Yapılandırma özeti: `{stats['config_digest']}`; transkript SHA-256: "
        f"`{stats['transcripts_sha256']}`",
        "",
        "## Seçilen türler",
        "",
        *(
            f"- {label}: {v['segments']} segment, {v['audio_ms'] / 3_600_000:.3f} sa"
            for label, v in kinds.items()
        ),
        "",
        "## Bölmeler (konuşmacı-ayrık)",
        "",
        *(
            f"- {name}: {v['speakers']} konuşmacı, {v['segments']} segment"
            for name, v in splits.items()
        ),
        "",
        "## Atlananlar (neden: sayı)",
        "",
        json.dumps(stats["skipped"], ensure_ascii=False),
        "",
        "## Karantina (neden: sayı)",
        "",
        json.dumps(stats["quarantined"], ensure_ascii=False),
        "",
        "## Eğitim yükleyicisi doğrulaması (`nida.train.data`)",
        "",
        *(
            f"- {name}: {v['accepted']}/{v['records']} kabul, ret: {v['rejection_counts']}"
            for name, v in loader.items()
        ),
    ]
    failed = stats["failed_speakers"]
    if failed:
        lines += ["", "## Başarısız konuşmacılar (yeniden çalıştırın)", ""]
        assert isinstance(failed, dict)
        lines += [f"- {speaker}: {error[:300]}" for speaker, error in failed.items()]
    return "\n".join(lines) + "\n"