893 lines
33 KiB
Python
893 lines
33 KiB
Python
"""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",
|
|
]
|