"""File-based import of an UmbraAudio metadata bundle.

The audio node is the master for the audiobooks. This module reads the
machine-readable bundle it exports (``bundle.json`` plus ``manifest.jsonl``),
checks it, and turns it into an :class:`ImportPlan` — the exact set of rows the
controller would write. Nothing here touches a database; the plan is a value,
which is what makes the dry run a real dry run rather than a rolled-back write.

Two things this module refuses to do, both on purpose:

* **It never fetches a render's bytes.** Every artifact it plans carries
  ``integrity='unverified'``. A path and a checksum are a claim that somebody
  said a file exists there; only reading the bytes back turns that into a fact,
  and that is a different operation with a different name.

* **It never derives a chapter's identity from its order.** The chapter row id is
  the identity; ``chapter_index`` is reading order and a hint; the number in a
  distribution filename is neither. The export carries ``numbering_basis`` for
  exactly this reason, and a chapter without one is a bundle this module rejects.

The mapping this implements is ``docs/audio-system-mapping-v1.md``.
"""

from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Mapping, Protocol, runtime_checkable
from uuid import UUID, uuid4

from .security import utc_now

#: The only bundle format this module understands. A future format is a new
#: constant and a branch, not a guess about what an unknown field means.
BUNDLE_FORMAT_VERSION = 1

RECORD_KINDS = ("book", "chapter", "speaker", "utterance", "render")

UTTERANCE_TYPES = frozenset({"narration", "dialogue", "scene_break"})

#: ``audio_format.container`` (measured) to an IANA media type. Keyed on the
#: container and never on the file extension: 33 files in the reference bundle
#: are named ``.wav`` and are MPEG audio, and the extension is the one thing that
#: cannot be trusted.
CONTAINER_MEDIA_TYPES: Mapping[str, str] = {
    "wav": "audio/wav",
    "mp3": "audio/mpeg",
    "aac": "audio/aac",
    "mp4": "audio/mp4",
    "m4a": "audio/mp4",
    "m4b": "audio/mp4",
    "ogg": "audio/ogg",
    "opus": "audio/opus",
    "flac": "audio/flac",
    "webm": "audio/webm",
}

#: Used only when the container is genuinely unknown. It is deliberately not
#: ``audio/wav`` or any other guess: a wrong media type is worse than an honest
#: "I do not know", because the wrong one is believed.
UNKNOWN_MEDIA_TYPE = "application/octet-stream"

#: A render role that represents finished output rather than an intermediate.
FINAL_RENDER_ROLES = frozenset({"distribution_track", "assembled_book"})

DEFAULT_SYSTEM = "umbraaudio"


class BundleError(Exception):
    """A bundle that cannot be read, or that must not be trusted."""

    def __init__(self, message: str) -> None:
        super().__init__(message)
        self.message = message


# ---------------------------------------------------------------------------
# Reading the bundle
# ---------------------------------------------------------------------------


@dataclass(frozen=True, slots=True)
class Bundle:
    """A bundle as read from disk, before any judgement is made about it."""

    path: Path
    envelope: Mapping[str, Any]
    records: tuple[Mapping[str, Any], ...]
    manifest_sha256: str
    declared_content_hash: str | None

    def of_kind(self, kind: str) -> tuple[Mapping[str, Any], ...]:
        return tuple(record for record in self.records if record.get("kind") == kind)

    @property
    def content_hash_matches(self) -> bool:
        return self.declared_content_hash is not None and self.declared_content_hash == self.manifest_sha256


def read_bundle(path: str | Path) -> Bundle:
    """Read a bundle directory. Raises :class:`BundleError` if it cannot be read.

    Reading is separated from judging. A bundle that parses and a bundle that is
    acceptable are different questions, and collapsing them is how a malformed
    bundle comes back as "mostly fine, I think".
    """

    root = Path(path)
    if not root.is_dir():
        raise BundleError(f"not a directory: {root}")
    envelope_path = root / "bundle.json"
    manifest_path = root / "manifest.jsonl"
    for required in (envelope_path, manifest_path):
        if not required.is_file():
            raise BundleError(f"missing required file: {required.name}")

    try:
        envelope = json.loads(envelope_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise BundleError(f"bundle.json is not readable JSON: {error}") from error
    if not isinstance(envelope, dict):
        raise BundleError("bundle.json must be a JSON object")

    manifest_bytes = manifest_path.read_bytes()
    manifest_sha256 = hashlib.sha256(manifest_bytes).hexdigest()

    records: list[Mapping[str, Any]] = []
    for number, line in enumerate(manifest_bytes.decode("utf-8").splitlines(), start=1):
        if not line.strip():
            continue
        try:
            record = json.loads(line)
        except json.JSONDecodeError as error:
            raise BundleError(f"manifest.jsonl:{number}: not valid JSON ({error.msg})") from error
        if not isinstance(record, dict):
            raise BundleError(f"manifest.jsonl:{number}: a record, not a {type(record).__name__}")
        records.append(record)
    if not records:
        raise BundleError("manifest.jsonl contains no records")

    declared = envelope.get("content_hash")
    declared_value = declared.get("value") if isinstance(declared, Mapping) else None

    return Bundle(
        path=root,
        envelope=envelope,
        records=tuple(records),
        manifest_sha256=manifest_sha256,
        declared_content_hash=declared_value if isinstance(declared_value, str) else None,
    )


# ---------------------------------------------------------------------------
# Checking the bundle
# ---------------------------------------------------------------------------


def validate_bundle(bundle: Bundle) -> list[str]:
    """Return every structural or invariant problem, or an empty list.

    Every problem is reported, not the first: a checker that stops at one turns a
    round trip into a correspondence. This is the pure-Python half of the gate;
    the JSON Schema half is ``docs/validate-audio-manifest.py``, run separately.
    """

    problems: list[str] = []

    version = bundle.envelope.get("bundle_format_version")
    if version != BUNDLE_FORMAT_VERSION:
        problems.append(f"bundle_format_version is {version!r}, expected {BUNDLE_FORMAT_VERSION}")
    if not bundle.content_hash_matches:
        problems.append(
            "content_hash mismatch: bundle.json declares "
            f"{bundle.declared_content_hash!r} but manifest.jsonl hashes to {bundle.manifest_sha256}"
        )

    by_id: dict[str, Mapping[str, Any]] = {}
    for record in bundle.records:
        rid = record.get("id")
        if isinstance(rid, str):
            by_id[rid] = record

    chapters = {c.get("id") for c in bundle.of_kind("chapter")}
    speakers = {s.get("id") for s in bundle.of_kind("speaker")}
    narrator_ids = {s.get("id") for s in bundle.of_kind("speaker") if s.get("is_narrator")}

    books = bundle.of_kind("book")
    if len(books) != 1:
        problems.append(f"expected exactly one book record, found {len(books)}")
    if not narrator_ids:
        problems.append("no speaker record has is_narrator=true")

    for chapter in bundle.of_kind("chapter"):
        cid = chapter.get("id")
        if "numbering_basis" not in chapter:
            problems.append(f"chapter {cid!r}: no numbering_basis; identity cannot be told from order")
        if not isinstance(chapter.get("chapter_index"), int):
            problems.append(f"chapter {cid!r}: chapter_index is not an integer")
        if chapter.get("book_id") not in by_id:
            problems.append(f"chapter {cid!r}: book_id {chapter.get('book_id')!r} does not resolve")

    for utterance in bundle.of_kind("utterance"):
        uid = utterance.get("id")
        resolution = utterance.get("speaker_resolution")
        raw = utterance.get("speaker_name_raw")
        if utterance.get("chapter_id") not in chapters:
            problems.append(f"utterance {uid!r}: chapter_id {utterance.get('chapter_id')!r} does not resolve")
        if utterance.get("speaker_id") not in speakers:
            problems.append(f"utterance {uid!r}: speaker_id {utterance.get('speaker_id')!r} does not resolve")
        if utterance.get("utterance_type") not in UTTERANCE_TYPES:
            problems.append(f"utterance {uid!r}: bad utterance_type {utterance.get('utterance_type')!r}")
        if resolution == "implicit_narrator":
            if raw is not None:
                problems.append(
                    f"utterance {uid!r}: implicit_narrator with a non-null speaker_name_raw ({raw!r}); "
                    "null is what implicit means"
                )
            if utterance.get("speaker_id") not in narrator_ids:
                problems.append(f"utterance {uid!r}: implicit narrator does not resolve to a narrator speaker")
        elif resolution == "explicit":
            if not isinstance(raw, str) or not raw:
                problems.append(f"utterance {uid!r}: explicit attribution with speaker_name_raw {raw!r}")
        else:
            problems.append(f"utterance {uid!r}: unknown speaker_resolution {resolution!r}")
        render_id = utterance.get("render_id")
        if render_id is not None and render_id not in by_id:
            problems.append(f"utterance {uid!r}: render_id {render_id!r} does not resolve")

    for render in bundle.of_kind("render"):
        rid = render.get("id")
        fmt = render.get("audio_format")
        if fmt is not None and not isinstance(fmt, Mapping):
            problems.append(f"render {rid!r}: audio_format is not an object")
        book_id = render.get("book_id")
        if book_id is not None and book_id not in by_id:
            problems.append(f"render {rid!r}: book_id {book_id!r} does not resolve")
        chapter_id = render.get("chapter_id")
        if chapter_id is not None and chapter_id not in chapters:
            problems.append(f"render {rid!r}: chapter_id {chapter_id!r} does not resolve")

    return problems


# ---------------------------------------------------------------------------
# The plan
# ---------------------------------------------------------------------------


@dataclass(frozen=True, slots=True)
class ImportWarning:
    """Something a human should see before or after an import. Not an error."""

    code: str
    subject: str
    message: str


@dataclass(frozen=True, slots=True)
class PlannedProject:
    organization_id: UUID
    external_id: str
    name: str
    slug: str
    title: str | None
    source_status: str | None
    languages: tuple[str, ...]
    numbering: Mapping[str, Any]
    system: str
    node_id: str
    product_version: str | None
    book_directory: str | None
    content_hash: str


@dataclass(frozen=True, slots=True)
class PlannedChapter:
    external_id: str
    reading_order: int
    matter: str  # "novel" | "matter"
    title: str | None
    source_hash: str
    source_row_id: int
    numbering_basis: Mapping[str, Any]
    utterance_count: int
    source_status: str | None


@dataclass(frozen=True, slots=True)
class PlannedSpeaker:
    external_id: str
    display_name: str
    role: str  # "narrator" | "character"
    aliases: tuple[str, ...]
    collision_group: str | None
    has_profile: bool
    is_group: bool
    unidentified: bool
    edge_voice: str | None
    tts_engine: str | None
    voice_review_status: str | None
    #: The canonical narrator a merged spelling resolves to, else ``None``.
    merged_into: str | None


@dataclass(frozen=True, slots=True)
class PlannedUtterance:
    external_id: str
    chapter_external_id: str
    index: int
    kind: str
    text: str
    source_hash: str
    speaker_external_id: str
    speaker_name_raw: str | None
    speaker_name_resolved: str
    attribution: str
    needs_review: bool
    render_external_id: str | None
    source_row_id: int | None


@dataclass(frozen=True, slots=True)
class PlannedArtifact:
    source_render_id: str
    kind: str
    status: str
    content_hash: str
    byte_size: int
    media_type: str
    storage_key: str
    storage_uri: str
    integrity: str
    derivation: Mapping[str, Any]


@dataclass(frozen=True, slots=True)
class ImportPlan:
    project: PlannedProject
    chapters: tuple[PlannedChapter, ...]
    speakers: tuple[PlannedSpeaker, ...]
    utterances: tuple[PlannedUtterance, ...]
    artifacts: tuple[PlannedArtifact, ...]
    warnings: tuple[ImportWarning, ...]
    narrator_external_id: str | None

    def counts(self) -> dict[str, int]:
        return {
            "chapters": len(self.chapters),
            "speakers": len(self.speakers),
            "utterances": len(self.utterances),
            "artifacts": len(self.artifacts),
            "warnings": len(self.warnings),
        }


# ---------------------------------------------------------------------------
# Planning
# ---------------------------------------------------------------------------


def _media_type(audio_format: Mapping[str, Any] | None) -> tuple[str, str | None]:
    """Return the media type for a format, and a warning code when it is a guess."""

    container = None
    if isinstance(audio_format, Mapping):
        container = audio_format.get("container")
    if isinstance(container, str):
        media = CONTAINER_MEDIA_TYPES.get(container.strip().lower())
        if media is not None:
            return media, None
    return UNKNOWN_MEDIA_TYPE, "unknown_container"


def _book_directory(bundle: Bundle, book: Mapping[str, Any]) -> str | None:
    scope = bundle.envelope.get("scope")
    if isinstance(scope, Mapping):
        names = scope.get("books_full_detail")
        if isinstance(names, list) and names and isinstance(names[0], str):
            return names[0]
    for key in ("path_relative_to", "name"):
        value = book.get(key)
        if isinstance(value, str) and value and value != "book_directory":
            return value
    return None


def plan_import(
    bundle: Bundle,
    *,
    organization_id: UUID,
    project_name: str | None = None,
    project_slug: str | None = None,
    node_id: str | None = None,
) -> ImportPlan:
    """Map a validated bundle onto the rows the controller would write.

    Raises :class:`BundleError` when the bundle is not valid. Callers that want
    the individual problems rather than an exception call :func:`validate_bundle`
    first, which is what the CLI does so it can print all of them.
    """

    problems = validate_bundle(bundle)
    if problems:
        raise BundleError("; ".join(problems))

    book = bundle.of_kind("book")[0]
    source = bundle.envelope.get("source")
    source = source if isinstance(source, Mapping) else {}
    resolved_node = node_id or source.get("node_id") or "unknown"
    languages = book.get("language_designations")
    languages_tuple = tuple(str(value) for value in languages) if isinstance(languages, list) else ()
    numbering = book.get("numbering")
    numbering = dict(numbering) if isinstance(numbering, Mapping) else {}

    external_id = str(book.get("id"))
    resolved_name = project_name or str(book.get("name") or external_id)
    resolved_slug = project_slug or _slugify(resolved_name)

    project = PlannedProject(
        organization_id=organization_id,
        external_id=external_id,
        name=resolved_name,
        slug=resolved_slug,
        title=book.get("title") if isinstance(book.get("title"), str) else None,
        source_status=book.get("status") if isinstance(book.get("status"), str) else None,
        languages=languages_tuple,
        numbering=numbering,
        system=str(source.get("system") or DEFAULT_SYSTEM),
        node_id=str(resolved_node),
        product_version=source.get("product_version") if isinstance(source.get("product_version"), str) else None,
        book_directory=_book_directory(bundle, book),
        content_hash=bundle.manifest_sha256,
    )

    warnings: list[ImportWarning] = []

    chapters = tuple(
        _plan_chapter(record) for record in sorted(bundle.of_kind("chapter"), key=_chapter_sort_key)
    )

    speakers_raw = bundle.of_kind("speaker")
    narrator_ids = [s.get("id") for s in speakers_raw if s.get("is_narrator")]
    canonical_narrator = _canonical_narrator(speakers_raw)

    if len(narrator_ids) > 1:
        warnings.append(
            ImportWarning(
                "narrator_rows_merged",
                str(canonical_narrator),
                f"{len(narrator_ids)} narrator speaker rows ({', '.join(str(i) for i in narrator_ids)}) "
                f"resolve to one narrator role; canonical is {canonical_narrator!r}. "
                "Source spellings are kept as aliases.",
            )
        )

    speakers = tuple(
        _plan_speaker(record, canonical_narrator if record.get("is_narrator") else None)
        for record in sorted(speakers_raw, key=lambda r: str(r.get("id")))
    )

    utterances = tuple(
        _plan_utterance(record, canonical_narrator, narrator_ids)
        for record in sorted(bundle.of_kind("utterance"), key=_utterance_sort_key)
    )

    artifacts: list[PlannedArtifact] = []
    for render in sorted(bundle.of_kind("render"), key=lambda r: str(r.get("id"))):
        artifact, warning = _plan_artifact(render, project, bundle)
        artifacts.append(artifact)
        if warning is not None:
            warnings.append(warning)

    warnings.extend(_bundle_warnings(bundle, chapters, speakers))

    return ImportPlan(
        project=project,
        chapters=chapters,
        speakers=speakers,
        utterances=utterances,
        artifacts=tuple(artifacts),
        warnings=tuple(warnings),
        narrator_external_id=str(canonical_narrator) if canonical_narrator else None,
    )


def _slugify(value: str) -> str:
    import re

    slug = re.sub(r"[^a-z0-9]+", "-", value.strip().lower()).strip("-")
    return slug or "project"


def _chapter_sort_key(record: Mapping[str, Any]) -> tuple[int, str]:
    index = record.get("chapter_index")
    return (index if isinstance(index, int) else 0, str(record.get("id")))


def _utterance_sort_key(record: Mapping[str, Any]) -> tuple[str, int]:
    index = record.get("utterance_index")
    return (str(record.get("chapter_id")), index if isinstance(index, int) else 0)


def _canonical_narrator(speakers: tuple[Mapping[str, Any], ...]) -> str | None:
    """Pick the one narrator row that becomes the role.

    Preference order is the same one the mapping document states: a row whose
    voice was reviewed and approved, then the row carrying the most narration,
    then the lowest id so the choice is deterministic across runs.
    """

    narrators = [s for s in speakers if s.get("is_narrator")]
    if not narrators:
        return None

    def rank(record: Mapping[str, Any]) -> tuple[int, int, str]:
        approved = 1 if record.get("voice_review_status") == "approved" else 0
        count = record.get("utterance_count_including_implicit_narrator")
        return (-approved, -(count if isinstance(count, int) else 0), str(record.get("id")))

    return str(min(narrators, key=rank).get("id"))


def _plan_chapter(record: Mapping[str, Any]) -> PlannedChapter:
    index = record.get("chapter_index")
    reading_order = index if isinstance(index, int) else 0
    row_id = record.get("database_row_id")
    numbering = record.get("numbering_basis")
    return PlannedChapter(
        external_id=str(record.get("id")),
        reading_order=reading_order,
        matter="matter" if reading_order < 0 else "novel",
        title=record.get("title") if isinstance(record.get("title"), str) else None,
        source_hash=str(record.get("source_text_sha256")),
        source_row_id=row_id if isinstance(row_id, int) else 0,
        numbering_basis=dict(numbering) if isinstance(numbering, Mapping) else {},
        utterance_count=record.get("utterance_count") if isinstance(record.get("utterance_count"), int) else 0,
        source_status=record.get("status") if isinstance(record.get("status"), str) else None,
    )


def _plan_speaker(record: Mapping[str, Any], canonical_narrator: str | None) -> PlannedSpeaker:
    speaker_id = str(record.get("id"))
    is_narrator = bool(record.get("is_narrator"))
    aliases = record.get("aliases")
    return PlannedSpeaker(
        external_id=speaker_id,
        display_name=str(record.get("primary_name")),
        role="narrator" if is_narrator else "character",
        aliases=tuple(str(a) for a in aliases) if isinstance(aliases, list) else (),
        collision_group=record.get("casefold_collision_group")
        if isinstance(record.get("casefold_collision_group"), str)
        else None,
        has_profile=bool(record.get("has_profile")),
        is_group=bool(record.get("is_group")),
        unidentified=bool(record.get("unidentified")),
        edge_voice=record.get("assigned_edge_voice")
        if isinstance(record.get("assigned_edge_voice"), str)
        else None,
        tts_engine=record.get("tts_engine_used") if isinstance(record.get("tts_engine_used"), str) else None,
        voice_review_status=record.get("voice_review_status")
        if isinstance(record.get("voice_review_status"), str)
        else None,
        merged_into=canonical_narrator if is_narrator and canonical_narrator and canonical_narrator != speaker_id else None,
    )


def _plan_utterance(
    record: Mapping[str, Any],
    canonical_narrator: str | None,
    narrator_ids: list[Any],
) -> PlannedUtterance:
    speaker_id = record.get("speaker_id")
    # The merge, applied at the only place it changes an answer: an utterance
    # whose speaker is any narrator spelling points at the one narrator role.
    if canonical_narrator and speaker_id in narrator_ids:
        speaker_id = canonical_narrator
    row_id = record.get("database_row_id")
    index = record.get("utterance_index")
    return PlannedUtterance(
        external_id=str(record.get("id")),
        chapter_external_id=str(record.get("chapter_id")),
        index=index if isinstance(index, int) else 0,
        kind=str(record.get("utterance_type")),
        text=str(record.get("text") or ""),
        source_hash=str(record.get("text_sha256")),
        speaker_external_id=str(speaker_id),
        speaker_name_raw=record.get("speaker_name_raw")
        if isinstance(record.get("speaker_name_raw"), str)
        else None,
        speaker_name_resolved=str(record.get("speaker_name_resolved") or ""),
        attribution=str(record.get("speaker_resolution") or ""),
        needs_review=bool(record.get("needs_review")),
        render_external_id=record.get("render_id") if isinstance(record.get("render_id"), str) else None,
        source_row_id=row_id if isinstance(row_id, int) else None,
    )


def _plan_artifact(
    render: Mapping[str, Any], project: PlannedProject, bundle: Bundle
) -> tuple[PlannedArtifact, ImportWarning | None]:
    audio_format = render.get("audio_format")
    audio_format = dict(audio_format) if isinstance(audio_format, Mapping) else {}
    media_type, warning_code = _media_type(audio_format)

    role = str(render.get("render_role"))
    path = str(render.get("path"))
    byte_size = render.get("byte_size")
    byte_size = byte_size if isinstance(byte_size, int) else 0

    derivation: dict[str, Any] = {
        "source_render_id": str(render.get("id")),
        "render_role": role,
        "source_system": project.system,
        "source_node": project.node_id,
        "relative_to": render.get("relative_to"),
        "book_directory": project.book_directory,
        "audio_format": audio_format or None,
        "duration_ms": render.get("duration_ms"),
        "inputs": list(render.get("inputs") or []),
        "book_external_id": render.get("book_id"),
        "chapter_external_id": render.get("chapter_id"),
        "file_modified_at": render.get("file_modified_at"),
    }
    extension_matches = audio_format.get("extension_matches_container")
    if extension_matches is False:
        derivation["extension_mismatch"] = True

    storage_key = path
    root = project.book_directory or ""
    storage_uri = f"file://{project.node_id}/{root}/{path}" if root else f"file://{project.node_id}/{path}"

    artifact = PlannedArtifact(
        source_render_id=str(render.get("id")),
        kind="audio",
        status="final" if role in FINAL_RENDER_ROLES else "intermediate",
        content_hash=str(render.get("content_sha256")),
        byte_size=byte_size,
        media_type=media_type,
        storage_key=storage_key,
        storage_uri=storage_uri,
        integrity="unverified",
        derivation=derivation,
    )

    warning: ImportWarning | None = None
    subject = str(render.get("id"))
    if warning_code == "unknown_container":
        warning = ImportWarning(
            "unknown_container",
            subject,
            f"container {audio_format.get('container')!r} has no known media type; recorded as {UNKNOWN_MEDIA_TYPE}",
        )
    elif byte_size == 0:
        warning = ImportWarning(
            "zero_byte_render", subject, f"render is 0 bytes at {path!r}; recorded, not treated as missing"
        )
    return artifact, warning


def _bundle_warnings(
    bundle: Bundle,
    chapters: tuple[PlannedChapter, ...],
    speakers: tuple[PlannedSpeaker, ...],
) -> list[ImportWarning]:
    warnings: list[ImportWarning] = []
    book = bundle.of_kind("book")[0]

    declared = book.get("declared_total_chapters")
    measured = len(chapters)
    if isinstance(declared, int) and declared != measured:
        warnings.append(
            ImportWarning(
                "declared_vs_measured_chapters",
                str(book.get("id")),
                f"declared {declared} chapters, measured {measured}; surfaced, not reconciled",
            )
        )

    mismatches = book.get("extension_container_mismatches")
    if isinstance(mismatches, list):
        for entry in mismatches:
            if not isinstance(entry, Mapping):
                continue
            warnings.append(
                ImportWarning(
                    "extension_mismatch",
                    str(entry.get("id")),
                    f"named {entry.get('named_extension')!r} but is {entry.get('actual_container')!r} at "
                    f"{entry.get('path')!r}",
                )
            )

    for speaker in speakers:
        if speaker.role == "character" and not speaker.has_profile:
            warnings.append(
                ImportWarning("speaker_without_profile", speaker.external_id, f"{speaker.display_name!r} has no profile")
            )

    collision_groups: dict[str, list[str]] = {}
    for speaker in speakers:
        if speaker.collision_group:
            collision_groups.setdefault(speaker.collision_group, []).append(speaker.external_id)
    for group, members in sorted(collision_groups.items()):
        if len(members) > 1:
            warnings.append(
                ImportWarning(
                    "speaker_collision_group",
                    group,
                    f"spellings {', '.join(members)} share casefold group {group!r}; not merged",
                )
            )

    return warnings


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


def render_report(plan: ImportPlan, *, bundle_path: str | Path | None = None, problems: list[str] | None = None) -> str:
    """Render the dry-run report a human reads before confirming a commit."""

    lines: list[str] = []
    project = plan.project
    if problems:
        lines.append(f"VALIDATION FAILED: {len(problems)} problem(s)")
        lines.extend(f"  - {problem}" for problem in problems)
        lines.append("")
    header = f"Import plan for {project.name!r} ({project.slug})"
    if bundle_path is not None:
        header += f" from {bundle_path}"
    lines.append(header)
    lines.append(f"  source:        {project.system} / {project.node_id} ({project.product_version or 'unknown'})")
    lines.append(f"  external id:   {project.external_id}")
    lines.append(f"  content hash:  {project.content_hash}")
    lines.append(f"  languages:     {', '.join(project.languages) if project.languages else '(none declared)'}")
    lines.append(f"  narrator:      {plan.narrator_external_id or '(none)'}")
    lines.append("")
    counts = plan.counts()
    lines.append(
        "  to write: "
        + ", ".join(f"{name}={counts[name]}" for name in ("chapters", "speakers", "utterances", "artifacts"))
    )
    lines.append(f"  warnings: {counts['warnings']}")
    for warning in plan.warnings:
        lines.append(f"    [{warning.code}] {warning.subject}: {warning.message}")
    return "\n".join(lines)


# ---------------------------------------------------------------------------
# Importing
# ---------------------------------------------------------------------------


@dataclass(frozen=True, slots=True)
class ImportResult:
    project_id: UUID
    import_id: UUID
    external_id: str
    content_hash: str
    counts: Mapping[str, int]
    imported_at: datetime


@runtime_checkable
class AudioImportStore(Protocol):
    """Where a plan is written. The PostgreSQL adapter and the dev ledger share it."""

    def last_content_hash(self, external_id: str) -> str | None:
        """Return the content hash of the last import for this project, if any."""

        raise NotImplementedError(f"{type(self).__name__} does not implement last_content_hash().")

    def commit(self, plan: ImportPlan, *, imported_at: datetime) -> ImportResult:
        raise NotImplementedError(f"{type(self).__name__} does not implement commit().")


@dataclass(frozen=True, slots=True)
class ImportOutcome:
    status: str  # "dry_run" | "imported" | "no_change"
    plan: ImportPlan
    result: ImportResult | None
    report: str


def import_bundle(
    bundle: Bundle,
    store: AudioImportStore,
    *,
    organization_id: UUID,
    project_name: str | None = None,
    project_slug: str | None = None,
    node_id: str | None = None,
    dry_run: bool = True,
    confirm: str | None = None,
    now: datetime | None = None,
) -> ImportOutcome:
    """Plan a bundle and, when asked and confirmed, write it.

    The confirmation gate is here rather than in the CLI so that every caller
    passes through it: a non-dry-run commit requires ``confirm`` to be the exact
    content hash of the bundle being imported. A flag named ``--commit`` is too
    easy to leave in a script; the content hash is not.
    """

    plan = plan_import(
        bundle,
        organization_id=organization_id,
        project_name=project_name,
        project_slug=project_slug,
        node_id=node_id,
    )
    report = render_report(plan, bundle_path=bundle.path)

    if dry_run:
        return ImportOutcome(status="dry_run", plan=plan, result=None, report=report)

    if confirm != plan.project.content_hash:
        raise BundleError(
            "refusing to commit without confirmation: pass confirm=<content_hash> "
            f"({plan.project.content_hash})"
        )

    existing = store.last_content_hash(plan.project.external_id)
    if existing == plan.project.content_hash:
        return ImportOutcome(
            status="no_change",
            plan=plan,
            result=None,
            report=report + f"\n\nalready imported at {plan.project.content_hash}; nothing to do",
        )

    imported_at = now or utc_now()
    result = store.commit(plan, imported_at=imported_at)
    return ImportOutcome(status="imported", plan=plan, result=result, report=report)


@dataclass
class InMemoryAudioImportStore:
    """A store that keeps plans in memory. For tests and dry demonstrations.

    It records the most recent content hash per project, which is what makes the
    idempotency test a real test rather than a mock that always says yes.
    """

    results: list[ImportResult] = field(default_factory=list)
    plans: dict[str, ImportPlan] = field(default_factory=dict)
    hashes: dict[str, str] = field(default_factory=dict)

    def last_content_hash(self, external_id: str) -> str | None:
        return self.hashes.get(external_id)

    def commit(self, plan: ImportPlan, *, imported_at: datetime) -> ImportResult:
        result = ImportResult(
            project_id=uuid4(),
            import_id=uuid4(),
            external_id=plan.project.external_id,
            content_hash=plan.project.content_hash,
            counts=plan.counts(),
            imported_at=imported_at,
        )
        self.results.append(result)
        self.plans[plan.project.external_id] = plan
        self.hashes[plan.project.external_id] = plan.project.content_hash
        return result


__all__ = [
    "BUNDLE_FORMAT_VERSION",
    "Bundle",
    "BundleError",
    "ImportOutcome",
    "ImportPlan",
    "ImportResult",
    "ImportWarning",
    "InMemoryAudioImportStore",
    "AudioImportStore",
    "PlannedArtifact",
    "PlannedChapter",
    "PlannedProject",
    "PlannedSpeaker",
    "PlannedUtterance",
    "import_bundle",
    "plan_import",
    "read_bundle",
    "render_report",
    "validate_bundle",
]