audio: Phase 2 — Postgres AudioImportStore adapter + 0009 ledger; use PG store on commit with org from node; 0008 header updated; tests
This commit is contained in:
umbra audio
892
umbratrans_web/audio_import.py
Normal file
892
umbratrans_web/audio_import.py
Normal file
@@ -0,0 +1,892 @@
|
||||
"""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",
|
||||
]
|
||||
364
umbratrans_web/audio_import_postgres.py
Normal file
364
umbratrans_web/audio_import_postgres.py
Normal file
@@ -0,0 +1,364 @@
|
||||
"""The PostgreSQL store for audio-bundle imports.
|
||||
|
||||
This is the durable half of :mod:`umbratrans_web.audio_import`. That module maps
|
||||
a bundle to an :class:`~umbratrans_web.audio_import.ImportPlan`; this one writes
|
||||
the plan to the tables migration 0008 (and 0009) created, inside one transaction,
|
||||
so that a commit either lands in full or not at all.
|
||||
|
||||
Two identities are derived rather than minted at random:
|
||||
|
||||
* ``project_id`` is ``uuid5`` of ``(organization_id, system, external_id)``. The
|
||||
same source book imported again therefore keeps the same project id, which is
|
||||
what lets a later translation or export point at it stably. A random id per
|
||||
import would orphan everything the previous import produced.
|
||||
|
||||
* ``import_id`` is ``uuid5`` of ``(project_id, content_hash)`` and is written to
|
||||
``public.translation_imports``. It names a real row; the audio side's warning
|
||||
was exactly that an import must not return an identifier that corresponds to
|
||||
nothing. It is reproducible, so replaying an import that was rolled back
|
||||
reproduces the same id.
|
||||
|
||||
A changed bundle (a different ``content_hash``) is a new import, not an edit. The
|
||||
source tables are append-only on UPDATE, so a re-import removes the rows it is
|
||||
replacing (DELETE is permitted, and is what makes re-import possible) and writes
|
||||
the new snapshot. The import ledger is append-only on UPDATE too and is never
|
||||
cleared, so the run history survives the rows it describes.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import datetime
|
||||
from typing import Any, Sequence
|
||||
from uuid import UUID, uuid5
|
||||
|
||||
import psycopg
|
||||
from psycopg.rows import dict_row
|
||||
from psycopg_pool import ConnectionPool
|
||||
|
||||
from .audio_import import ImportPlan, ImportResult
|
||||
|
||||
#: Namespace for deterministic audio-import identities. Fixed forever: changing it
|
||||
#: rewrites the identity of every project already imported.
|
||||
AUDIO_IMPORT_NAMESPACE = UUID("ddca30b0-0e0b-4dcd-afba-8ec400532193")
|
||||
|
||||
|
||||
def audio_project_id(organization_id: UUID, system: str, external_id: str) -> UUID:
|
||||
"""The stable project id for a source book, independent of when it came in."""
|
||||
|
||||
return uuid5(AUDIO_IMPORT_NAMESPACE, f"{organization_id}:{system}:{external_id}")
|
||||
|
||||
|
||||
def audio_import_id(project_id: UUID, content_hash: str) -> UUID:
|
||||
"""The stable id for one import run: this content, for this project."""
|
||||
|
||||
return uuid5(AUDIO_IMPORT_NAMESPACE, f"{project_id}:{content_hash}")
|
||||
|
||||
|
||||
def _json(value: Any) -> str:
|
||||
"""Serialize to the compact, UTF-8 form jsonb round-trips without surprises."""
|
||||
|
||||
return json.dumps(value, separators=(",", ":"), ensure_ascii=False, default=str)
|
||||
|
||||
|
||||
class PostgresAudioImportStore:
|
||||
"""Write an :class:`ImportPlan` to the audio source tables in one transaction.
|
||||
|
||||
Implements :class:`~umbratrans_web.audio_import.AudioImportStore`. The pool is
|
||||
owned by this object; call :meth:`close` on shutdown.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
dsn: str,
|
||||
*,
|
||||
min_size: int = 1,
|
||||
max_size: int = 4,
|
||||
application_name: str = "umbratrans-audio-import",
|
||||
) -> None:
|
||||
self._pool = ConnectionPool(
|
||||
dsn,
|
||||
min_size=min_size,
|
||||
max_size=max_size,
|
||||
open=True,
|
||||
kwargs={
|
||||
"row_factory": dict_row,
|
||||
"autocommit": False,
|
||||
"application_name": application_name,
|
||||
},
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_settings(cls, settings: object) -> PostgresAudioImportStore:
|
||||
dsn = getattr(settings, "database_dsn", None)
|
||||
if not dsn:
|
||||
raise RuntimeError(
|
||||
"the audio import store requires a database; UMBRATRANS_DATABASE_DSN is unset"
|
||||
)
|
||||
return cls(dsn)
|
||||
|
||||
def close(self) -> None:
|
||||
self._pool.close()
|
||||
|
||||
# -- AudioImportStore --------------------------------------------------
|
||||
|
||||
def last_content_hash(self, external_id: str) -> str | None:
|
||||
with self._pool.connection() as connection:
|
||||
row = connection.execute(
|
||||
"""
|
||||
SELECT content_hash
|
||||
FROM public.project_source_refs
|
||||
WHERE external_id = %s
|
||||
ORDER BY imported_at DESC
|
||||
LIMIT 1
|
||||
""",
|
||||
(external_id,),
|
||||
).fetchone()
|
||||
return row["content_hash"] if row else None
|
||||
|
||||
def commit(self, plan: ImportPlan, *, imported_at: datetime) -> ImportResult:
|
||||
project = plan.project
|
||||
project_id = audio_project_id(
|
||||
project.organization_id, project.system, project.external_id
|
||||
)
|
||||
import_id = audio_import_id(project_id, project.content_hash)
|
||||
|
||||
with self._pool.connection() as connection:
|
||||
with connection.transaction():
|
||||
cursor = connection.cursor()
|
||||
# Artifacts live in public.artifacts and are content-addressed, so
|
||||
# they are inserted with ON CONFLICT DO NOTHING and never deleted;
|
||||
# a changed file becomes a second artifact rather than a rewrite.
|
||||
before_artifacts = connection.execute(
|
||||
"SELECT count(*) AS n FROM public.artifacts WHERE project_id = %s",
|
||||
(project_id,),
|
||||
).fetchone()["n"]
|
||||
|
||||
# The four source tables describe one snapshot, so a re-import
|
||||
# replaces them wholesale. DELETE is permitted on these tables
|
||||
# precisely so this is possible; there is no UPDATE anywhere.
|
||||
for table in (
|
||||
"translation_utterances",
|
||||
"translation_chapters",
|
||||
"translation_speakers",
|
||||
"project_source_refs",
|
||||
):
|
||||
connection.execute(f"DELETE FROM public.{table} WHERE project_id = %s", (project_id,))
|
||||
|
||||
connection.execute(
|
||||
"""
|
||||
INSERT INTO public.project_source_refs (
|
||||
project_id, organization_id, system, node_id, external_id,
|
||||
source_title, source_status, languages, numbering,
|
||||
content_hash, product_version, source_meta, imported_at
|
||||
) VALUES (
|
||||
%s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s::jsonb,
|
||||
%s, %s, %s::jsonb, %s
|
||||
)
|
||||
""",
|
||||
(
|
||||
project_id,
|
||||
project.organization_id,
|
||||
project.system,
|
||||
project.node_id,
|
||||
project.external_id,
|
||||
project.title,
|
||||
project.source_status,
|
||||
_json(list(project.languages)),
|
||||
_json(project.numbering),
|
||||
project.content_hash,
|
||||
project.product_version,
|
||||
_json(
|
||||
{
|
||||
"name": project.name,
|
||||
"slug": project.slug,
|
||||
"book_directory": project.book_directory,
|
||||
}
|
||||
),
|
||||
imported_at,
|
||||
),
|
||||
)
|
||||
|
||||
if plan.speakers:
|
||||
cursor.executemany(
|
||||
"""
|
||||
INSERT INTO public.translation_speakers (
|
||||
project_id, organization_id, external_id, display_name,
|
||||
role, aliases, collision_group, has_profile, is_group,
|
||||
unidentified, edge_voice, tts_engine, voice_review_status,
|
||||
merged_into_external_id, created_at
|
||||
) VALUES (
|
||||
%s, %s, %s, %s, %s, %s::jsonb, %s, %s, %s,
|
||||
%s, %s, %s, %s, %s, %s
|
||||
)
|
||||
""",
|
||||
[
|
||||
(
|
||||
project_id,
|
||||
project.organization_id,
|
||||
speaker.external_id,
|
||||
speaker.display_name,
|
||||
speaker.role,
|
||||
_json(list(speaker.aliases)),
|
||||
speaker.collision_group,
|
||||
speaker.has_profile,
|
||||
speaker.is_group,
|
||||
speaker.unidentified,
|
||||
speaker.edge_voice,
|
||||
speaker.tts_engine,
|
||||
speaker.voice_review_status,
|
||||
speaker.merged_into,
|
||||
imported_at,
|
||||
)
|
||||
for speaker in plan.speakers
|
||||
],
|
||||
)
|
||||
|
||||
if plan.chapters:
|
||||
cursor.executemany(
|
||||
"""
|
||||
INSERT INTO public.translation_chapters (
|
||||
project_id, organization_id, external_id, reading_order,
|
||||
matter, title, source_hash, source_row_id, numbering_basis,
|
||||
utterance_count, source_status, created_at
|
||||
) VALUES (
|
||||
%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s, %s, %s
|
||||
)
|
||||
""",
|
||||
[
|
||||
(
|
||||
project_id,
|
||||
project.organization_id,
|
||||
chapter.external_id,
|
||||
chapter.reading_order,
|
||||
chapter.matter,
|
||||
chapter.title,
|
||||
chapter.source_hash,
|
||||
chapter.source_row_id,
|
||||
_json(chapter.numbering_basis),
|
||||
chapter.utterance_count,
|
||||
chapter.source_status,
|
||||
imported_at,
|
||||
)
|
||||
for chapter in plan.chapters
|
||||
],
|
||||
)
|
||||
|
||||
if plan.utterances:
|
||||
cursor.executemany(
|
||||
"""
|
||||
INSERT INTO public.translation_utterances (
|
||||
project_id, organization_id, external_id,
|
||||
chapter_external_id, utterance_index, kind, text,
|
||||
source_hash, speaker_external_id, speaker_name_raw,
|
||||
speaker_name_resolved, attribution, needs_review,
|
||||
render_external_id, source_row_id, created_at
|
||||
) VALUES (
|
||||
%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
|
||||
)
|
||||
""",
|
||||
[
|
||||
(
|
||||
project_id,
|
||||
project.organization_id,
|
||||
utterance.external_id,
|
||||
utterance.chapter_external_id,
|
||||
utterance.index,
|
||||
utterance.kind,
|
||||
utterance.text,
|
||||
utterance.source_hash,
|
||||
utterance.speaker_external_id,
|
||||
utterance.speaker_name_raw,
|
||||
utterance.speaker_name_resolved,
|
||||
utterance.attribution,
|
||||
utterance.needs_review,
|
||||
utterance.render_external_id,
|
||||
utterance.source_row_id,
|
||||
imported_at,
|
||||
)
|
||||
for utterance in plan.utterances
|
||||
],
|
||||
)
|
||||
|
||||
if plan.artifacts:
|
||||
cursor.executemany(
|
||||
"""
|
||||
INSERT INTO public.artifacts (
|
||||
project_id, organization_id, lang, kind, status,
|
||||
content_hash, byte_size, media_type, storage_key,
|
||||
storage_uri, integrity, derivation, created_at
|
||||
) VALUES (
|
||||
%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s
|
||||
)
|
||||
ON CONFLICT DO NOTHING
|
||||
""",
|
||||
[
|
||||
(
|
||||
project_id,
|
||||
project.organization_id,
|
||||
None,
|
||||
artifact.kind,
|
||||
artifact.status,
|
||||
artifact.content_hash,
|
||||
artifact.byte_size,
|
||||
artifact.media_type,
|
||||
artifact.storage_key,
|
||||
artifact.storage_uri,
|
||||
artifact.integrity,
|
||||
_json(dict(artifact.derivation)),
|
||||
imported_at,
|
||||
)
|
||||
for artifact in plan.artifacts
|
||||
],
|
||||
)
|
||||
|
||||
after_artifacts = connection.execute(
|
||||
"SELECT count(*) AS n FROM public.artifacts WHERE project_id = %s",
|
||||
(project_id,),
|
||||
).fetchone()["n"]
|
||||
|
||||
written_artifacts = int(after_artifacts) - int(before_artifacts)
|
||||
counts = {
|
||||
"chapters": len(plan.chapters),
|
||||
"speakers": len(plan.speakers),
|
||||
"utterances": len(plan.utterances),
|
||||
"artifacts": written_artifacts,
|
||||
"warnings": len(plan.warnings),
|
||||
}
|
||||
|
||||
connection.execute(
|
||||
"""
|
||||
INSERT INTO public.translation_imports (
|
||||
import_id, project_id, organization_id, system, node_id,
|
||||
external_id, content_hash, counts, imported_at
|
||||
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s)
|
||||
""",
|
||||
(
|
||||
import_id,
|
||||
project_id,
|
||||
project.organization_id,
|
||||
project.system,
|
||||
project.node_id,
|
||||
project.external_id,
|
||||
project.content_hash,
|
||||
_json(counts),
|
||||
imported_at,
|
||||
),
|
||||
)
|
||||
|
||||
return ImportResult(
|
||||
project_id=project_id,
|
||||
import_id=import_id,
|
||||
external_id=project.external_id,
|
||||
content_hash=project.content_hash,
|
||||
counts=counts,
|
||||
imported_at=imported_at,
|
||||
)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"AUDIO_IMPORT_NAMESPACE",
|
||||
"PostgresAudioImportStore",
|
||||
"audio_import_id",
|
||||
"audio_project_id",
|
||||
]
|
||||
305
umbratrans_web/migrations/0008_audio_source_import.sql
Normal file
305
umbratrans_web/migrations/0008_audio_source_import.sql
Normal file
@@ -0,0 +1,305 @@
|
||||
-- 0008: the audio source import — the controller's copy of what the audio node
|
||||
-- analyzed, and the tables the file-based importer writes.
|
||||
--
|
||||
-- Status: APPLIED (applied by the audio-controller to `umbratrans` as
|
||||
-- described in the migration procedure; verified against scratch and then
|
||||
-- applied). Proposed alongside `umbratrans_web/audio_import.py` and
|
||||
-- `docs/audio-system-mapping-v1.md`.
|
||||
--
|
||||
-- Additive and idempotent: four new tables, one guard function, their policies.
|
||||
-- Nothing existing is altered, and no existing table gains or loses a column.
|
||||
--
|
||||
-- Why these tables exist at all. The audio node owns the audiobooks and exports
|
||||
-- a metadata bundle; the controller is the authoritative repository for the
|
||||
-- translation pipeline and must hold a copy of that analysis rather than reach
|
||||
-- across to the audio database whenever it needs a chapter or a speaker. The
|
||||
-- copy is a snapshot identified by `content_hash`, and the fields are the ones
|
||||
-- `audio_import.py` maps — the mapping document is the contract, this file is
|
||||
-- where it lands.
|
||||
--
|
||||
-- Three decisions here are load-bearing and are commented where they are made:
|
||||
--
|
||||
-- 1. `project_id` carries no foreign key. `projects` lives in the per-tenant
|
||||
-- schema (0005) and is created by `app.provision_tenant_schema`, so there is
|
||||
-- no stable relation name to reference. `public.artifacts` reached the same
|
||||
-- conclusion in 0003 and this follows it: `organization_id` is the FK that
|
||||
-- matters for tenancy, and `project_id` is an opaque key scoped by it.
|
||||
-- 2. Identity is the source row id, never the order. `source_row_id` and
|
||||
-- `external_id` come from the audio database; `reading_order` is a snapshot
|
||||
-- of one export and is deliberately not unique or indexed as an identity.
|
||||
-- 3. The four tables are append-only on UPDATE. A source fact whose text is
|
||||
-- rewritten underneath its `source_hash` is a row that no longer describes
|
||||
-- anything; a changed source is a new import, not an edit. DELETE is
|
||||
-- permitted, because re-import and retention both need it and the same
|
||||
-- reasoning as `artifacts` applies.
|
||||
--
|
||||
-- `integrity` for an artifact is `unverified` on import by design: the importer
|
||||
-- records a render's path and checksum as a claim. Reading the bytes back is a
|
||||
-- different operation with a different name.
|
||||
|
||||
BEGIN;
|
||||
|
||||
-- Fail fast rather than half-create if the prerequisites are missing. Both are
|
||||
-- phase-1 objects; if either is absent this is the wrong database.
|
||||
DO $preflight$
|
||||
BEGIN
|
||||
IF to_regclass('public.organizations') IS NULL THEN
|
||||
RAISE EXCEPTION 'public.organizations is missing; 0008 belongs on the control-plane schema';
|
||||
END IF;
|
||||
IF NOT EXISTS (SELECT 1 FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace
|
||||
WHERE n.nspname = 'app' AND p.proname = 'is_organization_member') THEN
|
||||
RAISE EXCEPTION 'app.is_organization_member(uuid) is missing; the RLS policies this file writes would be inert';
|
||||
END IF;
|
||||
END
|
||||
$preflight$;
|
||||
|
||||
-- ---------------------------------------------------------------------
|
||||
-- 1. The source reference: one row per project, one source system
|
||||
-- ---------------------------------------------------------------------
|
||||
|
||||
-- The book-level record: what this project came from, and the content hash of
|
||||
-- the whole export. The hash is what makes an import idempotent — the same bytes
|
||||
-- are the same import, and a different hash is a different one.
|
||||
CREATE TABLE IF NOT EXISTS public.project_source_refs (
|
||||
project_id uuid PRIMARY KEY,
|
||||
organization_id uuid NOT NULL REFERENCES public.organizations(id),
|
||||
-- The exporting system and the node that ran it. Text, not a FK to
|
||||
-- compute_nodes: the source system is a fact about the data (it is
|
||||
-- `umbraaudio` on `umbraaudio-m4`), and a node row can be removed (0006)
|
||||
-- without the provenance of an already-imported book becoming unknown.
|
||||
system text NOT NULL CHECK (length(btrim(system)) BETWEEN 1 AND 64),
|
||||
node_id text NOT NULL,
|
||||
external_id text NOT NULL,
|
||||
source_title text,
|
||||
source_status text,
|
||||
-- Languages the audio side designated, if any. Empty is accepted as an
|
||||
-- answer: target languages are a translation-side decision.
|
||||
languages jsonb NOT NULL DEFAULT '[]'::jsonb,
|
||||
-- The numbering schemes and the hazard that accompanies them, stored
|
||||
-- verbatim because the note ("three numbers identify one chapter and none
|
||||
-- derives from another") is the useful part.
|
||||
numbering jsonb NOT NULL DEFAULT '{}'::jsonb,
|
||||
content_hash text NOT NULL CHECK (content_hash ~ '^[0-9a-f]{64}$'),
|
||||
product_version text,
|
||||
source_meta jsonb NOT NULL DEFAULT '{}'::jsonb,
|
||||
imported_at timestamptz NOT NULL DEFAULT now(),
|
||||
-- One source system may only describe one project once. A second system (a
|
||||
-- future cover pipeline, say) may add its own row about the same project, so
|
||||
-- the uniqueness is on the pair and not on either alone.
|
||||
CONSTRAINT project_source_refs_system_external_key UNIQUE (system, external_id)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS project_source_refs_org_idx ON public.project_source_refs (organization_id);
|
||||
|
||||
-- ---------------------------------------------------------------------
|
||||
-- 2. Speakers
|
||||
-- ---------------------------------------------------------------------
|
||||
|
||||
-- One row per speaker *as the source recorded it*. The narrator is the one
|
||||
-- merge the importer performs, and it is performed on the identity the pipeline
|
||||
-- uses, not here: both narrator source rows are kept, and `merged_into` records
|
||||
-- which spelling they resolve to. Nothing else is merged — a casefold collision
|
||||
-- group is stored and surfaced to a human, never collapsed.
|
||||
CREATE TABLE IF NOT EXISTS public.translation_speakers (
|
||||
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
project_id uuid NOT NULL,
|
||||
organization_id uuid NOT NULL REFERENCES public.organizations(id),
|
||||
external_id text NOT NULL,
|
||||
display_name text NOT NULL,
|
||||
role text NOT NULL CHECK (role IN ('narrator', 'character')),
|
||||
aliases jsonb NOT NULL DEFAULT '[]'::jsonb,
|
||||
collision_group text,
|
||||
has_profile boolean NOT NULL DEFAULT false,
|
||||
is_group boolean NOT NULL DEFAULT false,
|
||||
unidentified boolean NOT NULL DEFAULT false,
|
||||
edge_voice text,
|
||||
tts_engine text,
|
||||
voice_review_status text,
|
||||
-- The canonical narrator this row resolves to, when it is not itself the
|
||||
-- canonical one. NULL for characters and for the canonical narrator.
|
||||
merged_into_external_id text,
|
||||
source_meta jsonb NOT NULL DEFAULT '{}'::jsonb,
|
||||
created_at timestamptz NOT NULL DEFAULT now(),
|
||||
CONSTRAINT translation_speakers_identity_key UNIQUE (project_id, external_id)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS translation_speakers_org_idx ON public.translation_speakers (organization_id);
|
||||
CREATE INDEX IF NOT EXISTS translation_speakers_project_idx ON public.translation_speakers (project_id);
|
||||
|
||||
-- ---------------------------------------------------------------------
|
||||
-- 3. Chapters
|
||||
-- ---------------------------------------------------------------------
|
||||
|
||||
CREATE TABLE IF NOT EXISTS public.translation_chapters (
|
||||
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
project_id uuid NOT NULL,
|
||||
organization_id uuid NOT NULL REFERENCES public.organizations(id),
|
||||
external_id text NOT NULL,
|
||||
-- A snapshot of the export's order. Useful, not an identity.
|
||||
reading_order integer NOT NULL,
|
||||
matter text NOT NULL DEFAULT 'novel' CHECK (matter IN ('novel', 'matter')),
|
||||
title text,
|
||||
source_hash text NOT NULL CHECK (source_hash ~ '^[0-9a-f]{64}$'),
|
||||
source_row_id bigint NOT NULL,
|
||||
numbering_basis jsonb NOT NULL DEFAULT '{}'::jsonb,
|
||||
utterance_count integer NOT NULL DEFAULT 0 CHECK (utterance_count >= 0),
|
||||
source_status text,
|
||||
source_meta jsonb NOT NULL DEFAULT '{}'::jsonb,
|
||||
created_at timestamptz NOT NULL DEFAULT now(),
|
||||
CONSTRAINT translation_chapters_identity_key UNIQUE (project_id, external_id)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS translation_chapters_org_idx ON public.translation_chapters (organization_id);
|
||||
CREATE INDEX IF NOT EXISTS translation_chapters_project_idx ON public.translation_chapters (project_id, reading_order);
|
||||
|
||||
-- ---------------------------------------------------------------------
|
||||
-- 4. Utterances
|
||||
-- ---------------------------------------------------------------------
|
||||
|
||||
-- `chapter_external_id` is text and not a FK to `translation_chapters`, for the
|
||||
-- same reason `project_id` has no FK: these rows arrive together in one import
|
||||
-- and the importer has already resolved that every reference resolves. A FK
|
||||
-- would add a constraint the import order has to satisfy without adding a
|
||||
-- guarantee the importer does not already hold.
|
||||
CREATE TABLE IF NOT EXISTS public.translation_utterances (
|
||||
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
project_id uuid NOT NULL,
|
||||
organization_id uuid NOT NULL REFERENCES public.organizations(id),
|
||||
external_id text NOT NULL,
|
||||
chapter_external_id text NOT NULL,
|
||||
utterance_index integer NOT NULL,
|
||||
kind text NOT NULL CHECK (kind IN ('narration', 'dialogue', 'scene_break')),
|
||||
text text NOT NULL,
|
||||
source_hash text NOT NULL CHECK (source_hash ~ '^[0-9a-f]{64}$'),
|
||||
speaker_external_id text NOT NULL,
|
||||
-- NULL is the implicit narrator, and that is the whole meaning of NULL here.
|
||||
-- An explicit attribution always carries the name the source wrote.
|
||||
speaker_name_raw text,
|
||||
speaker_name_resolved text NOT NULL DEFAULT '',
|
||||
attribution text NOT NULL CHECK (attribution IN ('implicit_narrator', 'explicit')),
|
||||
needs_review boolean NOT NULL DEFAULT false,
|
||||
render_external_id text,
|
||||
source_row_id bigint,
|
||||
source_meta jsonb NOT NULL DEFAULT '{}'::jsonb,
|
||||
created_at timestamptz NOT NULL DEFAULT now(),
|
||||
CONSTRAINT translation_utterances_identity_key UNIQUE (project_id, external_id)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS translation_utterances_org_idx ON public.translation_utterances (organization_id);
|
||||
CREATE INDEX IF NOT EXISTS translation_utterances_chapter_idx
|
||||
ON public.translation_utterances (project_id, chapter_external_id, utterance_index);
|
||||
CREATE INDEX IF NOT EXISTS translation_utterances_speaker_idx
|
||||
ON public.translation_utterances (project_id, speaker_external_id);
|
||||
|
||||
-- ---------------------------------------------------------------------
|
||||
-- 5. Append-only on UPDATE
|
||||
-- ---------------------------------------------------------------------
|
||||
|
||||
-- One guard for the four tables. UPDATE is refused because a source fact is a
|
||||
-- statement about a document at a moment, and rewriting it in place breaks the
|
||||
-- link between the row and its `source_hash`. DELETE is permitted, exactly as it
|
||||
-- is for `artifacts`: re-importing a changed bundle and reclaiming storage both
|
||||
-- need it, and the guarantee that matters (no silent edit) is not weakened by
|
||||
-- allowing a row to be removed and written again.
|
||||
DO $guard$
|
||||
BEGIN
|
||||
EXECUTE $func$
|
||||
CREATE OR REPLACE FUNCTION app.prevent_translation_source_mutation()
|
||||
RETURNS trigger LANGUAGE plpgsql AS $body$
|
||||
BEGIN
|
||||
RAISE EXCEPTION
|
||||
'translation source rows are a snapshot and are append-only; a changed source is a new import, not an edit';
|
||||
END;
|
||||
$body$;
|
||||
$func$;
|
||||
|
||||
EXECUTE 'DROP TRIGGER IF EXISTS project_source_refs_append_only ON public.project_source_refs';
|
||||
EXECUTE 'CREATE TRIGGER project_source_refs_append_only BEFORE UPDATE ON public.project_source_refs FOR EACH ROW EXECUTE FUNCTION app.prevent_translation_source_mutation()';
|
||||
EXECUTE 'DROP TRIGGER IF EXISTS translation_speakers_append_only ON public.translation_speakers';
|
||||
EXECUTE 'CREATE TRIGGER translation_speakers_append_only BEFORE UPDATE ON public.translation_speakers FOR EACH ROW EXECUTE FUNCTION app.prevent_translation_source_mutation()';
|
||||
EXECUTE 'DROP TRIGGER IF EXISTS translation_chapters_append_only ON public.translation_chapters';
|
||||
EXECUTE 'CREATE TRIGGER translation_chapters_append_only BEFORE UPDATE ON public.translation_chapters FOR EACH ROW EXECUTE FUNCTION app.prevent_translation_source_mutation()';
|
||||
EXECUTE 'DROP TRIGGER IF EXISTS translation_utterances_append_only ON public.translation_utterances';
|
||||
EXECUTE 'CREATE TRIGGER translation_utterances_append_only BEFORE UPDATE ON public.translation_utterances FOR EACH ROW EXECUTE FUNCTION app.prevent_translation_source_mutation()';
|
||||
|
||||
EXECUTE 'REVOKE ALL ON FUNCTION app.prevent_translation_source_mutation() FROM PUBLIC';
|
||||
IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'umbracontrol_api') THEN
|
||||
EXECUTE 'GRANT EXECUTE ON FUNCTION app.prevent_translation_source_mutation() TO umbracontrol_api';
|
||||
END IF;
|
||||
IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'umbracontrol_worker') THEN
|
||||
EXECUTE 'GRANT EXECUTE ON FUNCTION app.prevent_translation_source_mutation() TO umbracontrol_worker';
|
||||
END IF;
|
||||
END
|
||||
$guard$;
|
||||
|
||||
-- ---------------------------------------------------------------------
|
||||
-- 6. Row-level security
|
||||
-- ---------------------------------------------------------------------
|
||||
|
||||
-- Enabled but not forced, matching every other table in this schema: the
|
||||
-- application owns these tables, so the policy is defence in depth for a
|
||||
-- non-owner role and the adapter's own scoping is today's control. `FORCE` and a
|
||||
-- non-owner application role arrive together at gate G2 and nowhere sooner.
|
||||
--
|
||||
-- `CREATE POLICY` has no `IF NOT EXISTS`, so a re-run would fail on the second
|
||||
-- apply. Each is dropped first, which is what makes this file safe to apply
|
||||
-- twice.
|
||||
DO $rls$
|
||||
DECLARE
|
||||
target text;
|
||||
BEGIN
|
||||
FOREACH target IN ARRAY ARRAY[
|
||||
'project_source_refs', 'translation_speakers', 'translation_chapters', 'translation_utterances'
|
||||
] LOOP
|
||||
EXECUTE format('ALTER TABLE public.%I ENABLE ROW LEVEL SECURITY', target);
|
||||
EXECUTE format('DROP POLICY IF EXISTS %I ON public.%I', target || '_tenant', target);
|
||||
END LOOP;
|
||||
|
||||
EXECUTE 'CREATE POLICY project_source_refs_tenant ON public.project_source_refs USING (app.is_platform_admin() OR app.is_organization_member(organization_id))';
|
||||
EXECUTE 'CREATE POLICY translation_speakers_tenant ON public.translation_speakers USING (app.is_platform_admin() OR app.is_organization_member(organization_id))';
|
||||
EXECUTE 'CREATE POLICY translation_chapters_tenant ON public.translation_chapters USING (app.is_platform_admin() OR app.is_organization_member(organization_id))';
|
||||
EXECUTE 'CREATE POLICY translation_utterances_tenant ON public.translation_utterances USING (app.is_platform_admin() OR app.is_organization_member(organization_id))';
|
||||
END
|
||||
$rls$;
|
||||
|
||||
-- ---------------------------------------------------------------------
|
||||
-- 7. Documentation for the schema itself
|
||||
-- ---------------------------------------------------------------------
|
||||
|
||||
COMMENT ON TABLE public.project_source_refs IS
|
||||
'One row per project per source system: what the project was imported from and the content hash of that import.';
|
||||
COMMENT ON COLUMN public.project_source_refs.content_hash IS
|
||||
'sha256 of the exported manifest bytes. The same bytes are the same import; a different hash is a different import.';
|
||||
COMMENT ON TABLE public.translation_speakers IS
|
||||
'Speakers as the source recorded them. The narrator merge is recorded in merged_into_external_id; no other row is merged.';
|
||||
COMMENT ON TABLE public.translation_chapters IS
|
||||
'Chapters. Identity is external_id (the source row id); reading_order is a snapshot and not an identity.';
|
||||
COMMENT ON TABLE public.translation_utterances IS
|
||||
'Utterances. speaker_name_raw NULL means the implicit narrator, which is the entire meaning of NULL in that column.';
|
||||
COMMENT ON FUNCTION app.prevent_translation_source_mutation() IS
|
||||
'Refuses UPDATE on the translation source tables. A changed source is a new import, not an edit.';
|
||||
|
||||
COMMIT;
|
||||
|
||||
-- ---------------------------------------------------------------------
|
||||
-- Rollback, in reverse dependency order. Not applied automatically.
|
||||
--
|
||||
-- DROP POLICY IF EXISTS translation_utterances_tenant ON public.translation_utterances;
|
||||
-- DROP POLICY IF EXISTS translation_chapters_tenant ON public.translation_chapters;
|
||||
-- DROP POLICY IF EXISTS translation_speakers_tenant ON public.translation_speakers;
|
||||
-- DROP POLICY IF EXISTS project_source_refs_tenant ON public.project_source_refs;
|
||||
-- DROP TRIGGER IF EXISTS translation_utterances_append_only ON public.translation_utterances;
|
||||
-- DROP TRIGGER IF EXISTS translation_chapters_append_only ON public.translation_chapters;
|
||||
-- DROP TRIGGER IF EXISTS translation_speakers_append_only ON public.translation_speakers;
|
||||
-- DROP TRIGGER IF EXISTS project_source_refs_append_only ON public.project_source_refs;
|
||||
-- DROP FUNCTION IF EXISTS app.prevent_translation_source_mutation();
|
||||
-- DROP TABLE IF EXISTS public.translation_utterances;
|
||||
-- DROP TABLE IF EXISTS public.translation_chapters;
|
||||
-- DROP TABLE IF EXISTS public.translation_speakers;
|
||||
-- DROP TABLE IF EXISTS public.project_source_refs;
|
||||
--
|
||||
-- The rollback is destructive and drops the imported data with the tables. It is
|
||||
-- safe only while nothing the translation pipeline produced points at a
|
||||
-- translation row, which is true today because the pipeline has not run. Once a
|
||||
-- translated chapter exists, the rollback must be replaced by a deprecation, not
|
||||
-- run as written.
|
||||
96
umbratrans_web/migrations/0009_translation_import_runs.sql
Normal file
96
umbratrans_web/migrations/0009_translation_import_runs.sql
Normal file
@@ -0,0 +1,96 @@
|
||||
-- 0009: the import run ledger — one durable row per audio source import.
|
||||
--
|
||||
-- Status: NOT YET APPLIED. Proposed alongside the PostgreSQL audio-import store
|
||||
-- (`umbratrans_web/audio_import_postgres.py`). Apply with 0008 already present;
|
||||
-- this file depends only on `public.organizations` and the append-only guard
|
||||
-- function 0008 defines, and it is additive and idempotent.
|
||||
--
|
||||
-- Why a fifth table. 0008 stores *what was imported* — the source reference, the
|
||||
-- speakers, chapters and utterances. It has no row for *the act of importing*, so
|
||||
-- the importer had no durable identity to return for an import and would have
|
||||
-- had to mint one from nothing. `project_source_refs.project_id` is one row per
|
||||
-- project and identifies the project, not the run; two imports of the same book
|
||||
-- at different content hashes are two runs and one project.
|
||||
--
|
||||
-- The audio side named this failure directly: an import that returns a
|
||||
-- `project_id` and an `import_id` that correspond to no row is worse than an
|
||||
-- error, because it looks like success. `translation_imports.import_id` is the
|
||||
-- row that makes `import_id` name something, and `counts` records what that run
|
||||
-- actually wrote so a later "the numbers do not add up" question has an answer.
|
||||
--
|
||||
-- The source hash is the idempotency key and stays on `project_source_refs`
|
||||
-- (UNIQUE (system, external_id)). This table is an append-only history: it is
|
||||
-- never updated, and a re-import appends a new run rather than editing the last.
|
||||
|
||||
BEGIN;
|
||||
|
||||
DO $preflight$
|
||||
BEGIN
|
||||
IF to_regclass('public.project_source_refs') IS NULL THEN
|
||||
RAISE EXCEPTION 'public.project_source_refs is missing; apply 0008 before 0009';
|
||||
END IF;
|
||||
IF NOT EXISTS (SELECT 1 FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace
|
||||
WHERE n.nspname = 'app' AND p.proname = 'prevent_translation_source_mutation') THEN
|
||||
RAISE EXCEPTION 'app.prevent_translation_source_mutation() is missing; 0008 must be applied first';
|
||||
END IF;
|
||||
END
|
||||
$preflight$;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS public.translation_imports (
|
||||
import_id uuid PRIMARY KEY,
|
||||
project_id uuid NOT NULL,
|
||||
organization_id uuid NOT NULL REFERENCES public.organizations(id),
|
||||
-- The exporting system and node, copied from the source reference at the
|
||||
-- moment of import so the run stays legible even if a node row is removed.
|
||||
system text NOT NULL CHECK (length(btrim(system)) BETWEEN 1 AND 64),
|
||||
node_id text NOT NULL,
|
||||
external_id text NOT NULL,
|
||||
-- sha256 of the manifest bytes for this run. A run with a hash already
|
||||
-- present for the project is a replay; the importer refuses it before here.
|
||||
content_hash text NOT NULL CHECK (content_hash ~ '^[0-9a-f]{64}$'),
|
||||
-- What this run actually wrote, not what it planned. Identical bytes dedupe
|
||||
-- against rows already present, so the written counts can be lower than the
|
||||
-- plan's; this is the written number and is the one an audit should read.
|
||||
counts jsonb NOT NULL DEFAULT '{}'::jsonb,
|
||||
imported_at timestamptz NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS translation_imports_org_idx
|
||||
ON public.translation_imports (organization_id);
|
||||
CREATE INDEX IF NOT EXISTS translation_imports_project_idx
|
||||
ON public.translation_imports (project_id, imported_at DESC);
|
||||
|
||||
-- Append-only on UPDATE, reusing the guard 0008 already defines. A run record is
|
||||
-- a statement about a moment; editing it in place would break the link between
|
||||
-- the row and the counts it reports. DELETE is permitted for the same reason as
|
||||
-- the other four tables: retention and re-import need it.
|
||||
DO $guard$
|
||||
BEGIN
|
||||
EXECUTE 'DROP TRIGGER IF EXISTS translation_imports_append_only ON public.translation_imports';
|
||||
EXECUTE 'CREATE TRIGGER translation_imports_append_only BEFORE UPDATE ON public.translation_imports FOR EACH ROW EXECUTE FUNCTION app.prevent_translation_source_mutation()';
|
||||
END
|
||||
$guard$;
|
||||
|
||||
DO $rls$
|
||||
BEGIN
|
||||
EXECUTE 'ALTER TABLE public.translation_imports ENABLE ROW LEVEL SECURITY';
|
||||
EXECUTE 'DROP POLICY IF EXISTS translation_imports_tenant ON public.translation_imports';
|
||||
EXECUTE 'CREATE POLICY translation_imports_tenant ON public.translation_imports USING (app.is_platform_admin() OR app.is_organization_member(organization_id))';
|
||||
END
|
||||
$rls$;
|
||||
|
||||
COMMENT ON TABLE public.translation_imports IS
|
||||
'One row per audio source import run: the durable identity an import returns, and what it actually wrote.';
|
||||
COMMENT ON COLUMN public.translation_imports.counts IS
|
||||
'Rows actually written by this run (after deduplication), keyed by chapters/speakers/utterances/artifacts/warnings.';
|
||||
COMMENT ON COLUMN public.translation_imports.content_hash IS
|
||||
'sha256 of the manifest bytes for this run. The idempotency key lives on project_source_refs; this is the run history.';
|
||||
|
||||
COMMIT;
|
||||
|
||||
-- ---------------------------------------------------------------------
|
||||
-- Rollback, in reverse dependency order. Not applied automatically.
|
||||
--
|
||||
-- DROP POLICY IF EXISTS translation_imports_tenant ON public.translation_imports;
|
||||
-- DROP TRIGGER IF EXISTS translation_imports_append_only ON public.translation_imports;
|
||||
-- DROP TABLE IF EXISTS public.translation_imports;
|
||||
@@ -27,6 +27,8 @@ building. §7.7, §13.15.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
@@ -465,6 +467,26 @@ def compute_repository(request: Request):
|
||||
return repository
|
||||
|
||||
|
||||
def audio_import_store(request: Request):
|
||||
"""The PostgreSQL audio-import store, built once per application.
|
||||
|
||||
Only a commit needs it, so it is built on the commit path rather than at
|
||||
startup: a dry run is pure validation and must succeed on a deployment whose
|
||||
database is not reachable. There is deliberately no in-memory fallback here,
|
||||
because a commit that silently names nothing is the exact failure this
|
||||
adapter exists to prevent.
|
||||
"""
|
||||
|
||||
cached = getattr(request.app.state, "audio_import_store", None)
|
||||
if cached is not None:
|
||||
return cached
|
||||
from .audio_import_postgres import PostgresAudioImportStore
|
||||
|
||||
store = PostgresAudioImportStore.from_settings(request.app.state.settings)
|
||||
request.app.state.audio_import_store = store
|
||||
return store
|
||||
|
||||
|
||||
def require_node(
|
||||
request: Request,
|
||||
authorization: str | None = Header(default=None),
|
||||
@@ -1718,6 +1740,163 @@ def decompose(
|
||||
return [_job_response(child) for child in children]
|
||||
|
||||
|
||||
class AudioIngestRequest(RequestModel):
|
||||
"""An audio dataset bundle in the accepted format.
|
||||
|
||||
The audio side may POST the full envelope and records (same structure as
|
||||
bundle.json + manifest.jsonl), allowing the server to validate and map
|
||||
without fetching files.
|
||||
"""
|
||||
|
||||
envelope: dict[str, Any]
|
||||
records: list[dict[str, Any]]
|
||||
project_name: str | None = Field(default=None, max_length=200)
|
||||
project_slug: str | None = Field(default=None, max_length=200)
|
||||
node_id: str | None = Field(default=None, max_length=200)
|
||||
dry_run: bool = True
|
||||
confirm: str | None = Field(default=None, max_length=128)
|
||||
|
||||
|
||||
class AudioIngestWarningResponse(BaseModel):
|
||||
code: str
|
||||
subject: str
|
||||
message: str
|
||||
|
||||
|
||||
class AudioIngestCountsResponse(BaseModel):
|
||||
chapters: int
|
||||
speakers: int
|
||||
utterances: int
|
||||
artifacts: int
|
||||
warnings: int
|
||||
|
||||
|
||||
class AudioIngestResultResponse(BaseModel):
|
||||
project_id: UUID | None = None
|
||||
import_id: UUID | None = None
|
||||
external_id: str | None = None
|
||||
content_hash: str | None = None
|
||||
counts: AudioIngestCountsResponse | None = None
|
||||
imported_at: str | None = None
|
||||
|
||||
|
||||
class AudioIngestResponse(BaseModel):
|
||||
status: str
|
||||
content_hash: str
|
||||
narrator_external_id: str | None = None
|
||||
counts: AudioIngestCountsResponse
|
||||
warnings: list[AudioIngestWarningResponse]
|
||||
result: AudioIngestResultResponse | None = None
|
||||
|
||||
|
||||
@agent_router.post("/audio/import", response_model=AudioIngestResponse)
|
||||
def import_audio_dataset(
|
||||
payload: AudioIngestRequest,
|
||||
request: Request,
|
||||
node: NodeContext = Depends(require_node),
|
||||
) -> AudioIngestResponse:
|
||||
"""Import an audio bundle. Dry-runs by default; commit requires confirm."""
|
||||
|
||||
node.require("dataset:ingest")
|
||||
|
||||
from .audio_import import (
|
||||
Bundle,
|
||||
BundleError,
|
||||
InMemoryAudioImportStore,
|
||||
import_bundle,
|
||||
)
|
||||
|
||||
try:
|
||||
declared = payload.envelope.get("content_hash")
|
||||
declared_value = (
|
||||
declared.get("value") if isinstance(declared, dict) else None
|
||||
)
|
||||
manifest_bytes = (
|
||||
"\n".join(
|
||||
json.dumps(
|
||||
r,
|
||||
sort_keys=True,
|
||||
separators=(",", ":"),
|
||||
ensure_ascii=False,
|
||||
)
|
||||
for r in payload.records
|
||||
)
|
||||
+ "\n"
|
||||
)
|
||||
manifest_sha256 = hashlib.sha256(manifest_bytes.encode("utf-8")).hexdigest()
|
||||
bundle = Bundle(
|
||||
path=Path("."),
|
||||
envelope=payload.envelope,
|
||||
records=tuple(payload.records),
|
||||
manifest_sha256=manifest_sha256,
|
||||
declared_content_hash=declared_value
|
||||
if isinstance(declared_value, str)
|
||||
else None,
|
||||
)
|
||||
except Exception as error:
|
||||
raise BundleError(f"invalid bundle payload: {error}")
|
||||
|
||||
# A dry run is pure validation and never touches the database. A commit goes
|
||||
# through the PostgreSQL adapter or fails; there is deliberately no in-memory
|
||||
# commit path, because a commit that writes nothing and reports success is
|
||||
# the failure this route must not reproduce.
|
||||
#
|
||||
# The organization comes from the credential, never from the payload: a node
|
||||
# belongs to exactly one organization, and letting a request name one would
|
||||
# let a node write into another tenant's tables.
|
||||
store = (
|
||||
InMemoryAudioImportStore() if payload.dry_run else audio_import_store(request)
|
||||
)
|
||||
try:
|
||||
outcome = import_bundle(
|
||||
bundle,
|
||||
store,
|
||||
organization_id=node.organization_id,
|
||||
project_name=payload.project_name,
|
||||
project_slug=payload.project_slug,
|
||||
node_id=payload.node_id or node.node_id,
|
||||
dry_run=payload.dry_run,
|
||||
confirm=payload.confirm,
|
||||
)
|
||||
except BundleError:
|
||||
raise
|
||||
except Exception as error:
|
||||
raise BundleError(str(error))
|
||||
|
||||
result = outcome.result
|
||||
counts = AudioIngestCountsResponse(
|
||||
chapters=len(outcome.plan.chapters),
|
||||
speakers=len(outcome.plan.speakers),
|
||||
utterances=len(outcome.plan.utterances),
|
||||
artifacts=len(outcome.plan.artifacts),
|
||||
warnings=len(outcome.plan.warnings),
|
||||
)
|
||||
warnings = [
|
||||
AudioIngestWarningResponse(
|
||||
code=w.code, subject=w.subject, message=w.message
|
||||
)
|
||||
for w in outcome.plan.warnings
|
||||
]
|
||||
res = AudioIngestResultResponse()
|
||||
if result is not None:
|
||||
res = AudioIngestResultResponse(
|
||||
project_id=result.project_id,
|
||||
import_id=result.import_id,
|
||||
external_id=result.external_id,
|
||||
content_hash=result.content_hash,
|
||||
counts=AudioIngestCountsResponse(**result.counts),
|
||||
imported_at=result.imported_at.isoformat(),
|
||||
)
|
||||
return AudioIngestResponse(
|
||||
status=outcome.status,
|
||||
content_hash=outcome.plan.project.content_hash,
|
||||
narrator_external_id=outcome.plan.narrator_external_id,
|
||||
counts=counts,
|
||||
warnings=warnings,
|
||||
result=res if (outcome.result is not None or outcome.status == "no_change") else None,
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["admin_router", "agent_router"]
|
||||
|
||||
|
||||
|
||||
594
umbratrans_web/tests/test_audio_import.py
Normal file
594
umbratrans_web/tests/test_audio_import.py
Normal file
@@ -0,0 +1,594 @@
|
||||
"""Tests for the file-based audio-bundle import.
|
||||
|
||||
These run against a synthetic bundle written to a temporary directory, so the
|
||||
suite never depends on a real audio export being present. The synthetic bundle
|
||||
deliberately carries every hazard the mapping document names - a negative
|
||||
chapter index, two narrator rows, an MP3 file named ``.wav``, an absent
|
||||
``chapter_id``, a zero-byte render, a chapter declared but not measured, and a
|
||||
speaker with no profile - because a mapping that is only tested on the tidy case
|
||||
is a mapping that reads the wrong field the first time it meets a real one.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
|
||||
from umbratrans_web.audio_import import (
|
||||
BundleError,
|
||||
ImportPlan,
|
||||
InMemoryAudioImportStore,
|
||||
import_bundle,
|
||||
plan_import,
|
||||
read_bundle,
|
||||
render_report,
|
||||
validate_bundle,
|
||||
)
|
||||
|
||||
ORGANIZATION_ID = uuid4()
|
||||
|
||||
|
||||
def _sha(seed: str) -> str:
|
||||
return hashlib.sha256(seed.encode()).hexdigest()
|
||||
|
||||
|
||||
def _render(rid: str, role: str, path: str, *, container: str, codec: str, ext: str, matches: bool | None,
|
||||
book_id: str = "book-1", chapter_id: str | None = "ch-1", byte_size: int = 1000,
|
||||
duration_ms: int | None = 1000, **extra: object) -> dict:
|
||||
record = {
|
||||
"id": rid,
|
||||
"kind": "render",
|
||||
"generated_at": "2026-10-14T12:00:00Z",
|
||||
"render_role": role,
|
||||
"path": path,
|
||||
"relative_to": "book_directory",
|
||||
"path_is_absolute": True,
|
||||
"audio_format": {
|
||||
"container": container,
|
||||
"codec": codec,
|
||||
"extension_matches_container": matches,
|
||||
"detected_by": "ffprobe",
|
||||
"file_extension": ext,
|
||||
"sample_rate_hz": 24000,
|
||||
"channel_count": 1,
|
||||
"bit_rate_bps": None,
|
||||
},
|
||||
"duration_ms": duration_ms,
|
||||
"byte_size": byte_size,
|
||||
"content_sha256": _sha(path),
|
||||
"book_id": book_id,
|
||||
"inputs": [],
|
||||
}
|
||||
if chapter_id is not None:
|
||||
record["chapter_id"] = chapter_id
|
||||
record.update(extra)
|
||||
return record
|
||||
|
||||
|
||||
def synthetic_records() -> list[dict]:
|
||||
return [
|
||||
{
|
||||
"id": "book-1",
|
||||
"kind": "book",
|
||||
"generated_at": "2026-10-14T12:00:00Z",
|
||||
"name": "Synthetic",
|
||||
"title": "synthetic.docx",
|
||||
"status": "analyzed",
|
||||
"chapters": ["ch-1", "ch-2"],
|
||||
"chapter_count": 2,
|
||||
"utterance_count": 4,
|
||||
"speaker_count": 3,
|
||||
"novel_chapter_count": 1,
|
||||
"front_and_back_matter_count": 1,
|
||||
"language_designations": [],
|
||||
"numbering": {"hazard": "three numbers, none derivable"},
|
||||
"path_relative_to": "book_directory",
|
||||
"declared_total_chapters": 2,
|
||||
"extension_container_mismatch_count": 1,
|
||||
"extension_container_mismatches": [
|
||||
{"id": "rnd-2", "named_extension": "wav", "actual_container": "mp3", "path": "voices/foo_ref.wav"}
|
||||
],
|
||||
"speaker_names_without_a_profile": ["Captain"],
|
||||
},
|
||||
{
|
||||
"id": "ch-1",
|
||||
"kind": "chapter",
|
||||
"generated_at": "2026-10-14T12:00:00Z",
|
||||
"book_id": "book-1",
|
||||
"chapter_index": -1,
|
||||
"database_row_id": 1,
|
||||
"title": "Copyright",
|
||||
"source_text_sha256": _sha("ch1"),
|
||||
"source_text_char_count": 10,
|
||||
"word_count": 2,
|
||||
"utterance_count": 2,
|
||||
"renders": ["rnd-1"],
|
||||
"numbering_basis": {"note": "front matter, negative index"},
|
||||
"status": "analyzed",
|
||||
},
|
||||
{
|
||||
"id": "ch-2",
|
||||
"kind": "chapter",
|
||||
"generated_at": "2026-10-14T12:00:00Z",
|
||||
"book_id": "book-1",
|
||||
"chapter_index": 0,
|
||||
"database_row_id": 2,
|
||||
"title": "Salvage",
|
||||
"source_text_sha256": _sha("ch2"),
|
||||
"source_text_char_count": 20,
|
||||
"word_count": 4,
|
||||
"utterance_count": 2,
|
||||
"renders": ["rnd-3", "rnd-4"],
|
||||
"numbering_basis": {"note": "reading order 0"},
|
||||
"status": "analyzed",
|
||||
},
|
||||
{
|
||||
"id": "sp-narrator-a",
|
||||
"kind": "speaker",
|
||||
"generated_at": "2026-10-14T12:00:00Z",
|
||||
"primary_name": "Narrator",
|
||||
"aliases": ["narrator", "the narrator"],
|
||||
"has_profile": True,
|
||||
"is_narrator": True,
|
||||
"is_group": False,
|
||||
"unidentified": False,
|
||||
"casefold_collision_group": "narrator",
|
||||
"utterance_count": 1,
|
||||
"utterance_count_casefold_grouped": 3,
|
||||
"utterance_count_including_implicit_narrator": 10,
|
||||
"voice_review_status": "approved",
|
||||
"assigned_edge_voice": "en-CA-LiamNeural:-100",
|
||||
"tts_engine_used": None,
|
||||
},
|
||||
{
|
||||
"id": "sp-narrator-b",
|
||||
"kind": "speaker",
|
||||
"generated_at": "2026-10-14T12:00:00Z",
|
||||
"primary_name": "narrator",
|
||||
"aliases": ["narrator"],
|
||||
"has_profile": True,
|
||||
"is_narrator": True,
|
||||
"is_group": False,
|
||||
"unidentified": False,
|
||||
"casefold_collision_group": "narrator",
|
||||
"utterance_count": 2,
|
||||
"utterance_count_casefold_grouped": 3,
|
||||
"utterance_count_including_implicit_narrator": 5,
|
||||
"voice_review_status": None,
|
||||
"assigned_edge_voice": None,
|
||||
"tts_engine_used": None,
|
||||
},
|
||||
{
|
||||
"id": "sp-captain",
|
||||
"kind": "speaker",
|
||||
"generated_at": "2026-10-14T12:00:00Z",
|
||||
"primary_name": "Captain",
|
||||
"aliases": [],
|
||||
"has_profile": False,
|
||||
"is_narrator": False,
|
||||
"is_group": False,
|
||||
"unidentified": False,
|
||||
"casefold_collision_group": None,
|
||||
"utterance_count": 1,
|
||||
"utterance_count_casefold_grouped": 1,
|
||||
"utterance_count_including_implicit_narrator": 1,
|
||||
"voice_review_status": None,
|
||||
"assigned_edge_voice": None,
|
||||
"tts_engine_used": "qwen",
|
||||
},
|
||||
{
|
||||
"id": "utt-1",
|
||||
"kind": "utterance",
|
||||
"generated_at": "2026-10-14T12:00:00Z",
|
||||
"book_id": "book-1",
|
||||
"chapter_id": "ch-1",
|
||||
"database_row_id": 10,
|
||||
"utterance_index": 0,
|
||||
"utterance_type": "narration",
|
||||
"text": ",,,",
|
||||
"text_sha256": _sha("t1"),
|
||||
"speaker_id": "sp-narrator-b",
|
||||
"speaker_name_raw": None,
|
||||
"speaker_name_resolved": "narrator",
|
||||
"speaker_resolution": "implicit_narrator",
|
||||
"render_id": "rnd-1",
|
||||
"needs_review": False,
|
||||
},
|
||||
{
|
||||
"id": "utt-2",
|
||||
"kind": "utterance",
|
||||
"generated_at": "2026-10-14T12:00:00Z",
|
||||
"book_id": "book-1",
|
||||
"chapter_id": "ch-2",
|
||||
"database_row_id": 11,
|
||||
"utterance_index": 0,
|
||||
"utterance_type": "dialogue",
|
||||
"text": "Hello.",
|
||||
"text_sha256": _sha("t2"),
|
||||
"speaker_id": "sp-captain",
|
||||
"speaker_name_raw": "Captain",
|
||||
"speaker_name_resolved": "Captain",
|
||||
"speaker_resolution": "explicit",
|
||||
"render_id": "rnd-1",
|
||||
"needs_review": False,
|
||||
},
|
||||
{
|
||||
"id": "utt-3",
|
||||
"kind": "utterance",
|
||||
"generated_at": "2026-10-14T12:00:00Z",
|
||||
"book_id": "book-1",
|
||||
"chapter_id": "ch-2",
|
||||
"database_row_id": 12,
|
||||
"utterance_index": 1,
|
||||
"utterance_type": "narration",
|
||||
"text": "He left.",
|
||||
"text_sha256": _sha("t3"),
|
||||
"speaker_id": "sp-narrator-a",
|
||||
"speaker_name_raw": "Narrator",
|
||||
"speaker_name_resolved": "Narrator",
|
||||
"speaker_resolution": "explicit",
|
||||
"render_id": "rnd-1",
|
||||
"needs_review": False,
|
||||
},
|
||||
{
|
||||
"id": "utt-4",
|
||||
"kind": "utterance",
|
||||
"generated_at": "2026-10-14T12:00:00Z",
|
||||
"book_id": "book-1",
|
||||
"chapter_id": "ch-2",
|
||||
"database_row_id": 13,
|
||||
"utterance_index": 2,
|
||||
"utterance_type": "scene_break",
|
||||
"text": ",,,",
|
||||
"text_sha256": _sha("t4"),
|
||||
"speaker_id": "sp-narrator-b",
|
||||
"speaker_name_raw": None,
|
||||
"speaker_name_resolved": "narrator",
|
||||
"speaker_resolution": "implicit_narrator",
|
||||
"render_id": "rnd-1",
|
||||
"needs_review": False,
|
||||
},
|
||||
_render("rnd-1", "utterance_audio", "chapter_audio/ch1/utt_0001.wav", container="wav", codec="pcm_s16le", ext="wav", matches=True),
|
||||
_render("rnd-2", "voice_reference", "voices/foo_ref.wav", container="mp3", codec="mp3", ext="wav", matches=False, chapter_id=None),
|
||||
_render("rnd-3", "distribution_track", "export/Deal_Hunter_01_Salvage.mp3", container="mp3", codec="mp3", ext="mp3", matches=True, chapter_id="ch-2", track_number_in_name="01"),
|
||||
_render("rnd-4", "utterance_audio", "chapter_audio/ch2/utt_0226.wav", container="wav", codec="pcm_s16le", ext="wav", matches=True, chapter_id="ch-2", byte_size=0),
|
||||
]
|
||||
|
||||
|
||||
def write_bundle(directory: Path, records: list[dict], *, source: dict | None = None) -> Path:
|
||||
"""Write a bundle whose declared hash matches its manifest."""
|
||||
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
manifest = "\n".join(json.dumps(record, sort_keys=True) for record in records) + "\n"
|
||||
(directory / "manifest.jsonl").write_text(manifest, encoding="utf-8")
|
||||
digest = hashlib.sha256((directory / "manifest.jsonl").read_bytes()).hexdigest()
|
||||
envelope = {
|
||||
"bundle_format_version": 1,
|
||||
"content_hash": {"algorithm": "sha256", "of": "manifest.jsonl bytes", "value": digest},
|
||||
"generated_at": "2026-10-14T12:00:00Z",
|
||||
"source": source or {"system": "umbraaudio", "node_id": "umbraaudio-m4", "product_version": "test"},
|
||||
"counts": {"books": 1, "chapters": 2, "utterances": 4, "speakers": 3, "render_records": 4, "records_total": len(records)},
|
||||
"record_kinds": {"discriminator": "kind", "values": ["book", "chapter", "speaker", "utterance", "render"]},
|
||||
"scope": {"books_full_detail": ["SyntheticBook"]},
|
||||
}
|
||||
(directory / "bundle.json").write_text(json.dumps(envelope), encoding="utf-8")
|
||||
return directory
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def bundle_dir(tmp_path: Path) -> Path:
|
||||
return write_bundle(tmp_path / "bundle", synthetic_records())
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def bundle(bundle_dir: Path):
|
||||
return read_bundle(bundle_dir)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def plan(bundle) -> ImportPlan:
|
||||
return plan_import(bundle, organization_id=ORGANIZATION_ID)
|
||||
|
||||
|
||||
# -- reading ----------------------------------------------------------------
|
||||
|
||||
|
||||
def test_read_bundle_hashes_the_manifest_bytes(bundle_dir: Path) -> None:
|
||||
bundle = read_bundle(bundle_dir)
|
||||
expected = hashlib.sha256((bundle_dir / "manifest.jsonl").read_bytes()).hexdigest()
|
||||
assert bundle.manifest_sha256 == expected
|
||||
assert bundle.content_hash_matches is True
|
||||
assert bundle.declared_content_hash == expected
|
||||
|
||||
|
||||
def test_read_bundle_rejects_a_missing_manifest(tmp_path: Path) -> None:
|
||||
(tmp_path / "bundle").mkdir()
|
||||
(tmp_path / "bundle" / "bundle.json").write_text("{}")
|
||||
with pytest.raises(BundleError):
|
||||
read_bundle(tmp_path / "bundle")
|
||||
|
||||
|
||||
def test_a_tampered_manifest_does_not_match_its_declared_hash(bundle_dir: Path) -> None:
|
||||
path = bundle_dir / "manifest.jsonl"
|
||||
lines = path.read_text().splitlines()
|
||||
record = json.loads(lines[0])
|
||||
record["name"] = record.get("name", "") + " TAMPERED"
|
||||
lines[0] = json.dumps(record, sort_keys=True)
|
||||
path.write_text("\n".join(lines) + "\n")
|
||||
|
||||
bundle = read_bundle(bundle_dir)
|
||||
assert bundle.content_hash_matches is False
|
||||
assert any("content_hash mismatch" in problem for problem in validate_bundle(bundle))
|
||||
|
||||
|
||||
# -- mapping ----------------------------------------------------------------
|
||||
|
||||
|
||||
def test_chapter_identity_is_the_id_not_the_order(plan: ImportPlan) -> None:
|
||||
by_id = {chapter.external_id: chapter for chapter in plan.chapters}
|
||||
assert by_id["ch-1"].reading_order == -1
|
||||
assert by_id["ch-2"].reading_order == 0
|
||||
assert {c.external_id for c in plan.chapters} == {"ch-1", "ch-2"}
|
||||
|
||||
|
||||
def test_negative_chapter_index_is_front_matter(plan: ImportPlan) -> None:
|
||||
by_id = {chapter.external_id: chapter for chapter in plan.chapters}
|
||||
assert by_id["ch-1"].matter == "matter"
|
||||
assert by_id["ch-2"].matter == "novel"
|
||||
|
||||
|
||||
def test_media_type_comes_from_the_container_not_the_extension(plan: ImportPlan) -> None:
|
||||
by_id = {artifact.source_render_id: artifact for artifact in plan.artifacts}
|
||||
# rnd-2 is named .wav and is MPEG audio.
|
||||
assert by_id["rnd-2"].media_type == "audio/mpeg"
|
||||
assert by_id["rnd-1"].media_type == "audio/wav"
|
||||
assert by_id["rnd-3"].media_type == "audio/mpeg"
|
||||
|
||||
|
||||
def test_artifacts_are_unverified(plan: ImportPlan) -> None:
|
||||
assert plan.artifacts
|
||||
assert all(artifact.integrity == "unverified" for artifact in plan.artifacts)
|
||||
|
||||
|
||||
def test_render_without_a_chapter_key_attaches_to_the_project(plan: ImportPlan) -> None:
|
||||
by_id = {artifact.source_render_id: artifact for artifact in plan.artifacts}
|
||||
assert by_id["rnd-2"].derivation["chapter_external_id"] is None
|
||||
assert by_id["rnd-2"].derivation["extension_mismatch"] is True
|
||||
|
||||
|
||||
def test_final_roles_are_final_and_the_rest_intermediate(plan: ImportPlan) -> None:
|
||||
by_id = {artifact.source_render_id: artifact for artifact in plan.artifacts}
|
||||
assert by_id["rnd-3"].status == "final"
|
||||
assert by_id["rnd-1"].status == "intermediate"
|
||||
|
||||
|
||||
def test_zero_byte_render_is_recorded_not_dropped(plan: ImportPlan) -> None:
|
||||
by_id = {artifact.source_render_id: artifact for artifact in plan.artifacts}
|
||||
assert by_id["rnd-4"].byte_size == 0
|
||||
assert any(warning.code == "zero_byte_render" and warning.subject == "rnd-4" for warning in plan.warnings)
|
||||
|
||||
|
||||
# -- the narrator merge -----------------------------------------------------
|
||||
|
||||
|
||||
def test_the_narrator_is_one_role_and_the_approved_row_is_canonical(plan: ImportPlan) -> None:
|
||||
assert plan.narrator_external_id == "sp-narrator-a"
|
||||
by_id = {speaker.external_id: speaker for speaker in plan.speakers}
|
||||
assert by_id["sp-narrator-a"].role == "narrator"
|
||||
assert by_id["sp-narrator-a"].merged_into is None
|
||||
assert by_id["sp-narrator-b"].role == "narrator"
|
||||
assert by_id["sp-narrator-b"].merged_into == "sp-narrator-a"
|
||||
|
||||
|
||||
def test_every_narrator_utterance_points_at_the_one_role(plan: ImportPlan) -> None:
|
||||
by_id = {utterance.external_id: utterance for utterance in plan.utterances}
|
||||
# utt-1 and utt-4 were stored against sp-narrator-b.
|
||||
assert by_id["utt-1"].speaker_external_id == "sp-narrator-a"
|
||||
assert by_id["utt-4"].speaker_external_id == "sp-narrator-a"
|
||||
# utt-3 was explicit against the canonical row and is unchanged.
|
||||
assert by_id["utt-3"].speaker_external_id == "sp-narrator-a"
|
||||
|
||||
|
||||
def test_scene_breaks_are_kept_as_structural_units(plan: ImportPlan) -> None:
|
||||
by_id = {utterance.external_id: utterance for utterance in plan.utterances}
|
||||
assert by_id["utt-4"].kind == "scene_break"
|
||||
|
||||
|
||||
# -- warnings ---------------------------------------------------------------
|
||||
|
||||
|
||||
def test_speaker_without_a_profile_is_flagged(plan: ImportPlan) -> None:
|
||||
assert any(
|
||||
warning.code == "speaker_without_profile" and warning.subject == "sp-captain" for warning in plan.warnings
|
||||
)
|
||||
|
||||
|
||||
def test_a_collision_group_is_surfaced_not_merged(plan: ImportPlan) -> None:
|
||||
collisions = [w for w in plan.warnings if w.code == "speaker_collision_group"]
|
||||
assert len(collisions) == 1
|
||||
assert "sp-narrator-a" in collisions[0].message and "sp-narrator-b" in collisions[0].message
|
||||
|
||||
|
||||
def test_extension_mismatch_is_reported_verbatim(plan: ImportPlan) -> None:
|
||||
assert any(warning.code == "extension_mismatch" and warning.subject == "rnd-2" for warning in plan.warnings)
|
||||
|
||||
|
||||
# -- validation -------------------------------------------------------------
|
||||
|
||||
|
||||
def test_implicit_narrator_with_a_written_name_is_rejected(bundle_dir: Path) -> None:
|
||||
records = synthetic_records()
|
||||
for record in records:
|
||||
if record["id"] == "utt-1":
|
||||
record["speaker_name_raw"] = "narrator"
|
||||
write_bundle(bundle_dir, records)
|
||||
problems = validate_bundle(read_bundle(bundle_dir))
|
||||
assert any("implicit_narrator" in problem for problem in problems)
|
||||
|
||||
|
||||
def test_chapter_without_numbering_basis_is_rejected(bundle_dir: Path) -> None:
|
||||
records = synthetic_records()
|
||||
for record in records:
|
||||
if record["id"] == "ch-2":
|
||||
del record["numbering_basis"]
|
||||
write_bundle(bundle_dir, records)
|
||||
problems = validate_bundle(read_bundle(bundle_dir))
|
||||
assert any("numbering_basis" in problem for problem in problems)
|
||||
|
||||
|
||||
def test_dangling_chapter_reference_is_rejected(bundle_dir: Path) -> None:
|
||||
records = synthetic_records()
|
||||
for record in records:
|
||||
if record["id"] == "utt-2":
|
||||
record["chapter_id"] = "ch-does-not-exist"
|
||||
write_bundle(bundle_dir, records)
|
||||
problems = validate_bundle(read_bundle(bundle_dir))
|
||||
assert any("does not resolve" in problem for problem in problems)
|
||||
|
||||
|
||||
# -- import and idempotency -------------------------------------------------
|
||||
|
||||
|
||||
def test_dry_run_writes_nothing(bundle) -> None:
|
||||
store = InMemoryAudioImportStore()
|
||||
outcome = import_bundle(bundle, store, organization_id=ORGANIZATION_ID, dry_run=True)
|
||||
assert outcome.status == "dry_run"
|
||||
assert outcome.result is None
|
||||
assert store.results == []
|
||||
assert store.hashes == {}
|
||||
|
||||
|
||||
def test_commit_requires_the_content_hash(bundle) -> None:
|
||||
store = InMemoryAudioImportStore()
|
||||
with pytest.raises(BundleError):
|
||||
import_bundle(bundle, store, organization_id=ORGANIZATION_ID, dry_run=False, confirm="yes")
|
||||
assert store.results == []
|
||||
|
||||
|
||||
def test_commit_writes_the_plan(bundle) -> None:
|
||||
store = InMemoryAudioImportStore()
|
||||
outcome = import_bundle(
|
||||
bundle, store, organization_id=ORGANIZATION_ID, dry_run=False, confirm=bundle.manifest_sha256
|
||||
)
|
||||
assert outcome.status == "imported"
|
||||
assert outcome.result is not None
|
||||
assert outcome.result.content_hash == bundle.manifest_sha256
|
||||
assert len(store.results) == 1
|
||||
assert store.hashes[outcome.result.external_id] == bundle.manifest_sha256
|
||||
|
||||
|
||||
def test_reimporting_the_same_content_hash_is_a_no_op(bundle) -> None:
|
||||
store = InMemoryAudioImportStore()
|
||||
import_bundle(bundle, store, organization_id=ORGANIZATION_ID, dry_run=False, confirm=bundle.manifest_sha256)
|
||||
again = import_bundle(
|
||||
bundle, store, organization_id=ORGANIZATION_ID, dry_run=False, confirm=bundle.manifest_sha256
|
||||
)
|
||||
assert again.status == "no_change"
|
||||
assert len(store.results) == 1
|
||||
|
||||
|
||||
def test_report_names_the_project_and_counts(plan: ImportPlan) -> None:
|
||||
report = render_report(plan, bundle_path="/somewhere")
|
||||
assert "Synthetic" in report
|
||||
assert "chapters=2" in report
|
||||
assert "sp-narrator-a" in report # the narrator line
|
||||
assert "/somewhere" in report
|
||||
|
||||
|
||||
# -- the command line -------------------------------------------------------
|
||||
|
||||
|
||||
def _run_cli(argv: list[str], monkeypatch: pytest.MonkeyPatch) -> tuple[int, str, str]:
|
||||
from io import StringIO
|
||||
|
||||
from umbratrans_web.cli import import_audio
|
||||
|
||||
monkeypatch.delenv("UMBRA_ENV", raising=False)
|
||||
out, err = StringIO(), StringIO()
|
||||
code = import_audio(argv, stdout=out, stderr=err)
|
||||
return code, out.getvalue(), err.getvalue()
|
||||
|
||||
|
||||
def test_cli_defaults_to_a_dry_run(bundle_dir: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
ledger = tmp_path / "ledger.json"
|
||||
code, out, _ = _run_cli(
|
||||
["--bundle", str(bundle_dir), "--organization-id", str(ORGANIZATION_ID), "--ledger", str(ledger)],
|
||||
monkeypatch,
|
||||
)
|
||||
assert code == 0
|
||||
assert "dry run" in out
|
||||
assert "chapters=2" in out
|
||||
assert not ledger.exists()
|
||||
|
||||
|
||||
def test_cli_commit_writes_the_ledger_then_is_idempotent(
|
||||
bundle_dir: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
ledger = tmp_path / "ledger.json"
|
||||
bundle = read_bundle(bundle_dir)
|
||||
base = ["--bundle", str(bundle_dir), "--organization-id", str(ORGANIZATION_ID), "--ledger", str(ledger)]
|
||||
|
||||
code, out, _ = _run_cli([*base, "--commit", "--confirm", bundle.manifest_sha256], monkeypatch)
|
||||
assert code == 0
|
||||
assert "imported:" in out
|
||||
assert ledger.is_file()
|
||||
assert ledger.stat().st_mode & 0o777 == 0o600
|
||||
|
||||
code, out, _ = _run_cli([*base, "--commit", "--confirm", bundle.manifest_sha256], monkeypatch)
|
||||
assert code == 0
|
||||
assert "nothing to do" in out
|
||||
|
||||
|
||||
def test_cli_commit_without_confirmation_is_refused(
|
||||
bundle_dir: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
ledger = tmp_path / "ledger.json"
|
||||
code, _, err = _run_cli(
|
||||
["--bundle", str(bundle_dir), "--organization-id", str(ORGANIZATION_ID), "--ledger", str(ledger), "--commit"],
|
||||
monkeypatch,
|
||||
)
|
||||
assert code == 1
|
||||
assert "confirmation" in err
|
||||
assert not ledger.exists()
|
||||
|
||||
|
||||
def test_cli_commit_needs_an_organization(bundle_dir: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
bundle = read_bundle(bundle_dir)
|
||||
code, _, err = _run_cli(
|
||||
["--bundle", str(bundle_dir), "--ledger", str(tmp_path / "ledger.json"), "--commit", "--confirm", bundle.manifest_sha256],
|
||||
monkeypatch,
|
||||
)
|
||||
assert code == 1
|
||||
assert "organization-id" in err
|
||||
|
||||
|
||||
def test_cli_reports_a_broken_bundle_and_exits_two(
|
||||
bundle_dir: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
path = bundle_dir / "manifest.jsonl"
|
||||
lines = path.read_text().splitlines()
|
||||
record = json.loads(lines[0])
|
||||
record["name"] = record.get("name", "") + " TAMPERED"
|
||||
lines[0] = json.dumps(record, sort_keys=True)
|
||||
path.write_text("\n".join(lines) + "\n")
|
||||
|
||||
code, _, err = _run_cli(["--bundle", str(bundle_dir)], monkeypatch)
|
||||
assert code == 2
|
||||
assert "VALIDATION FAILED" in err
|
||||
assert "content_hash mismatch" in err
|
||||
|
||||
|
||||
def test_cli_refuses_production(bundle_dir: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
from io import StringIO
|
||||
|
||||
from umbratrans_web.cli import import_audio
|
||||
|
||||
monkeypatch.setenv("UMBRA_ENV", "production")
|
||||
out, err = StringIO(), StringIO()
|
||||
code = import_audio(["--bundle", str(bundle_dir)], stdout=out, stderr=err)
|
||||
assert code == 3
|
||||
assert "development-only" in err.getvalue()
|
||||
447
umbratrans_web/tests/test_audio_import_postgres.py
Normal file
447
umbratrans_web/tests/test_audio_import_postgres.py
Normal file
@@ -0,0 +1,447 @@
|
||||
"""Tests for the PostgreSQL audio-import store.
|
||||
|
||||
Two layers, matching ``test_postgres_repository.py``.
|
||||
|
||||
**Mapping and contract tests run everywhere.** The identities are derived rather
|
||||
than minted, and those derivations are asserted directly: a random project id per
|
||||
import would orphan everything the previous import produced, and the audio side
|
||||
named that failure explicitly. The store is also checked against the
|
||||
``AudioImportStore`` protocol by walking the MRO, and its INSERT statements are
|
||||
checked against the columns the migrations actually create - a typo in a column
|
||||
name is otherwise only visible against a real database.
|
||||
|
||||
**Integration tests require a real, migrated scratch database and skip without
|
||||
one.** They are selected by ``UMBRATRANS_TEST_DSN`` and
|
||||
``UMBRATRANS_TEST_DATABASE_IS_DISPOSABLE``, the same gate the repository suite
|
||||
uses, and they additionally skip when migrations 0008 and 0009 have not been
|
||||
applied, so an unmigrated scratch database is a skip with an instruction rather
|
||||
than a failure that reads like a defect in the adapter.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
import pytest
|
||||
|
||||
from umbratrans_web.audio_import import (
|
||||
AudioImportStore,
|
||||
import_bundle,
|
||||
plan_import,
|
||||
read_bundle,
|
||||
)
|
||||
from umbratrans_web.audio_import_postgres import (
|
||||
AUDIO_IMPORT_NAMESPACE,
|
||||
PostgresAudioImportStore,
|
||||
audio_import_id,
|
||||
audio_project_id,
|
||||
)
|
||||
|
||||
from .test_audio_import import ORGANIZATION_ID, synthetic_records, write_bundle
|
||||
|
||||
TEST_DSN_ENV = "UMBRATRANS_TEST_DSN"
|
||||
TEST_ACK_ENV = "UMBRATRANS_TEST_DATABASE_IS_DISPOSABLE"
|
||||
MIGRATIONS = Path(__file__).resolve().parent.parent / "migrations"
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Derived identities
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_project_id_is_stable_for_the_same_source() -> None:
|
||||
first = audio_project_id(ORGANIZATION_ID, "umbraaudio", "book-1")
|
||||
second = audio_project_id(ORGANIZATION_ID, "umbraaudio", "book-1")
|
||||
assert first == second
|
||||
assert isinstance(first, UUID)
|
||||
|
||||
|
||||
def test_project_id_varies_with_every_part_of_its_key() -> None:
|
||||
base = audio_project_id(ORGANIZATION_ID, "umbraaudio", "book-1")
|
||||
assert base != audio_project_id(uuid4(), "umbraaudio", "book-1")
|
||||
assert base != audio_project_id(ORGANIZATION_ID, "other", "book-1")
|
||||
assert base != audio_project_id(ORGANIZATION_ID, "umbraaudio", "book-2")
|
||||
|
||||
|
||||
def test_import_id_is_content_addressed() -> None:
|
||||
project = audio_project_id(ORGANIZATION_ID, "umbraaudio", "book-1")
|
||||
hash_a = "a" * 64
|
||||
hash_b = "b" * 64
|
||||
assert audio_import_id(project, hash_a) == audio_import_id(project, hash_a)
|
||||
# A changed source is a new import, so the id must change with the hash.
|
||||
assert audio_import_id(project, hash_a) != audio_import_id(project, hash_b)
|
||||
# And it is scoped to the project, not globally to the hash.
|
||||
assert audio_import_id(project, hash_a) != audio_import_id(uuid4(), hash_a)
|
||||
|
||||
|
||||
def test_project_and_import_ids_live_in_the_stated_namespace() -> None:
|
||||
project = audio_project_id(ORGANIZATION_ID, "umbraaudio", "book-1")
|
||||
assert project.version == 5
|
||||
assert project.variant == AUDIO_IMPORT_NAMESPACE.variant
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Protocol conformance
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_the_adapter_provides_every_protocol_method() -> None:
|
||||
"""A method counts only if a class below the protocol defines it.
|
||||
|
||||
``hasattr`` is true for an inherited ``NotImplementedError`` stub, so it would
|
||||
pass while the adapter implemented nothing. This is the same trap the
|
||||
repository suite documents.
|
||||
"""
|
||||
|
||||
provided: set[str] = set()
|
||||
for klass in PostgresAudioImportStore.__mro__:
|
||||
if klass is AudioImportStore:
|
||||
break
|
||||
provided.update(name for name, value in vars(klass).items() if callable(value))
|
||||
declared = {name for name, value in vars(AudioImportStore).items() if callable(value)}
|
||||
assert declared - provided == set()
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# The INSERT statements against the migrations, without a database
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
_CONSTRAINT_WORDS = {
|
||||
"PRIMARY",
|
||||
"FOREIGN",
|
||||
"UNIQUE",
|
||||
"CHECK",
|
||||
"CONSTRAINT",
|
||||
"REFERENCES",
|
||||
"EXCLUDE",
|
||||
}
|
||||
|
||||
|
||||
def _created_columns() -> dict[str, set[str]]:
|
||||
"""Column names per ``public.`` table, read from the migration files.
|
||||
|
||||
Both CREATE TABLE columns and ALTER TABLE ADD COLUMN columns are collected:
|
||||
``artifacts`` gains ``storage_uri`` and ``integrity`` in 0007, and a parser
|
||||
that read only CREATE TABLE would report them as typos.
|
||||
"""
|
||||
|
||||
tables: dict[str, set[str]] = {}
|
||||
for path in sorted(MIGRATIONS.glob("*.sql")):
|
||||
text = path.read_text(encoding="utf-8")
|
||||
for match in re.finditer(
|
||||
r"CREATE TABLE IF NOT EXISTS\s+public\.(\w+)\s*\((.*?)\n\);",
|
||||
text,
|
||||
re.DOTALL,
|
||||
):
|
||||
table, body = match.group(1), match.group(2)
|
||||
columns: set[str] = set()
|
||||
for line in body.splitlines():
|
||||
stripped = line.strip()
|
||||
if not stripped:
|
||||
continue
|
||||
word = stripped.split()[0]
|
||||
if word.upper() in _CONSTRAINT_WORDS:
|
||||
continue
|
||||
columns.add(word)
|
||||
tables.setdefault(table, set()).update(columns)
|
||||
for match in re.finditer(
|
||||
r"ALTER TABLE\s+public\.(\w+)(.*?);",
|
||||
text,
|
||||
re.DOTALL | re.IGNORECASE,
|
||||
):
|
||||
table, body = match.group(1), match.group(2)
|
||||
for column in re.findall(
|
||||
r"ADD COLUMN(?:\s+IF NOT EXISTS)?\s+(\w+)", body, re.IGNORECASE
|
||||
):
|
||||
tables.setdefault(table, set()).add(column)
|
||||
return tables
|
||||
|
||||
|
||||
def _inserted_columns() -> dict[str, set[str]]:
|
||||
"""Columns named by each ``INSERT INTO public.<table> (...)`` in the adapter."""
|
||||
|
||||
module = (
|
||||
Path(__file__).resolve().parent.parent / "audio_import_postgres.py"
|
||||
).read_text(encoding="utf-8")
|
||||
inserted: dict[str, set[str]] = {}
|
||||
for match in re.finditer(
|
||||
r"INSERT INTO public\.(\w+)\s*\((.*?)\)\s*VALUES",
|
||||
module,
|
||||
re.DOTALL,
|
||||
):
|
||||
table, body = match.group(1), match.group(2)
|
||||
columns = {token.strip() for token in body.split(",") if token.strip()}
|
||||
inserted[table] = columns
|
||||
return inserted
|
||||
|
||||
|
||||
def test_every_inserted_column_exists_in_the_migrations() -> None:
|
||||
created = _created_columns()
|
||||
inserted = _inserted_columns()
|
||||
assert inserted, "no INSERT statements were found; the parser is wrong"
|
||||
for table, columns in inserted.items():
|
||||
assert table in created, f"public.{table} is not created by any migration"
|
||||
unknown = columns - created[table]
|
||||
assert not unknown, f"public.{table} has no column(s): {sorted(unknown)}"
|
||||
|
||||
|
||||
def test_the_adapter_covers_every_source_table() -> None:
|
||||
# The four snapshot tables and the import ledger. public.artifacts is the
|
||||
# fifth write but is created by 0003, not the audio-import migrations.
|
||||
assert set(_inserted_columns()) == {
|
||||
"project_source_refs",
|
||||
"translation_speakers",
|
||||
"translation_chapters",
|
||||
"translation_utterances",
|
||||
"artifacts",
|
||||
"translation_imports",
|
||||
}
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Integration: requires a real, migrated scratch database
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def _integration_dsn() -> str | None:
|
||||
return os.environ.get(TEST_DSN_ENV) or None
|
||||
|
||||
|
||||
def _acknowledged() -> bool:
|
||||
return os.environ.get(TEST_ACK_ENV, "").strip() in {"1", "true", "yes", "on"}
|
||||
|
||||
|
||||
integration = pytest.mark.skipif(
|
||||
_integration_dsn() is None or not _acknowledged(),
|
||||
reason=(
|
||||
f"{TEST_DSN_ENV} and {TEST_ACK_ENV}=1 are required. Point them at a "
|
||||
"disposable scratch database with migrations 0001-0009 applied."
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def bundle_dir(tmp_path: Path) -> Path:
|
||||
"""The synthetic bundle's directory, written fresh for each test."""
|
||||
|
||||
return write_bundle(tmp_path / "bundle", synthetic_records())
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def store_and_org():
|
||||
"""A live store and a freshly inserted organization, cleaned up afterwards."""
|
||||
|
||||
import psycopg
|
||||
|
||||
dsn = _integration_dsn()
|
||||
assert dsn is not None
|
||||
connection = psycopg.connect(dsn, autocommit=True)
|
||||
try:
|
||||
if connection.execute(
|
||||
"SELECT to_regclass('public.translation_imports')"
|
||||
).fetchone()[0] is None:
|
||||
pytest.skip(
|
||||
"migrations 0008-0009 are not applied to the scratch database; "
|
||||
"apply them before running the audio-import integration tests"
|
||||
)
|
||||
organization_id = uuid4()
|
||||
connection.execute(
|
||||
"INSERT INTO public.organizations (id, slug, name) VALUES (%s, %s, %s)",
|
||||
(organization_id, f"audio-import-{organization_id.hex[:12]}", "Audio Import Test"),
|
||||
)
|
||||
store = PostgresAudioImportStore(dsn)
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
yield store, organization_id
|
||||
|
||||
store.close()
|
||||
cleanup = psycopg.connect(dsn, autocommit=True)
|
||||
try:
|
||||
# Child-of-organization rows first; the session teardown also truncates
|
||||
# public.organizations CASCADE, but leaving rows behind between tests
|
||||
# would make one test's residue another test's surprise.
|
||||
cleanup.execute("DELETE FROM public.translation_imports WHERE organization_id = %s", (organization_id,))
|
||||
cleanup.execute("DELETE FROM public.artifacts WHERE organization_id = %s", (organization_id,))
|
||||
cleanup.execute("DELETE FROM public.translation_utterances WHERE organization_id = %s", (organization_id,))
|
||||
cleanup.execute("DELETE FROM public.translation_chapters WHERE organization_id = %s", (organization_id,))
|
||||
cleanup.execute("DELETE FROM public.translation_speakers WHERE organization_id = %s", (organization_id,))
|
||||
cleanup.execute("DELETE FROM public.project_source_refs WHERE organization_id = %s", (organization_id,))
|
||||
cleanup.execute("DELETE FROM public.organizations WHERE id = %s", (organization_id,))
|
||||
finally:
|
||||
cleanup.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def rebuilt_bundle(tmp_path: Path):
|
||||
"""A writer for a second bundle under the same book id, with changed bytes."""
|
||||
|
||||
def build(name: str, *, mutate: bool) -> Path:
|
||||
records = synthetic_records()
|
||||
if mutate:
|
||||
for record in records:
|
||||
if record.get("id") == "utt-2":
|
||||
record["text"] = "Goodbye."
|
||||
record["text_sha256"] = "c" * 64
|
||||
return write_bundle(tmp_path / name, records)
|
||||
|
||||
return build
|
||||
|
||||
|
||||
@integration
|
||||
def test_commit_writes_the_source_snapshot(store_and_org, bundle_dir: Path) -> None:
|
||||
store, organization_id = store_and_org
|
||||
bundle = read_bundle(bundle_dir)
|
||||
plan = plan_import(bundle, organization_id=organization_id)
|
||||
|
||||
imported_at = datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc)
|
||||
result = store.commit(plan, imported_at=imported_at)
|
||||
|
||||
assert result.project_id == audio_project_id(
|
||||
organization_id, plan.project.system, plan.project.external_id
|
||||
)
|
||||
assert result.import_id == audio_import_id(result.project_id, plan.project.content_hash)
|
||||
assert result.content_hash == plan.project.content_hash
|
||||
assert result.counts["chapters"] == len(plan.chapters)
|
||||
assert result.counts["speakers"] == len(plan.speakers)
|
||||
assert result.counts["utterances"] == len(plan.utterances)
|
||||
assert result.counts["artifacts"] == len(plan.artifacts)
|
||||
|
||||
assert store.last_content_hash(plan.project.external_id) == plan.project.content_hash
|
||||
|
||||
# And the rows are actually there, counted independently of the result.
|
||||
import psycopg
|
||||
from psycopg.rows import dict_row
|
||||
|
||||
connection = psycopg.connect(_integration_dsn(), row_factory=dict_row)
|
||||
try:
|
||||
for table, expected in (
|
||||
("project_source_refs", 1),
|
||||
("translation_speakers", len(plan.speakers)),
|
||||
("translation_chapters", len(plan.chapters)),
|
||||
("translation_utterances", len(plan.utterances)),
|
||||
("artifacts", len(plan.artifacts)),
|
||||
("translation_imports", 1),
|
||||
):
|
||||
row = connection.execute(
|
||||
f"SELECT count(*) AS n FROM public.{table} WHERE project_id = %s",
|
||||
(result.project_id,),
|
||||
).fetchone()
|
||||
assert row["n"] == expected, table
|
||||
ref = connection.execute(
|
||||
"SELECT external_id, content_hash, source_meta FROM public.project_source_refs WHERE project_id = %s",
|
||||
(result.project_id,),
|
||||
).fetchone()
|
||||
assert ref["external_id"] == plan.project.external_id
|
||||
assert ref["content_hash"] == plan.project.content_hash
|
||||
assert ref["source_meta"]["name"] == plan.project.name
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
|
||||
@integration
|
||||
def test_import_bundle_reports_no_change_for_identical_content(
|
||||
store_and_org, bundle_dir: Path
|
||||
) -> None:
|
||||
store, organization_id = store_and_org
|
||||
bundle = read_bundle(bundle_dir)
|
||||
plan = plan_import(bundle, organization_id=organization_id)
|
||||
|
||||
first = import_bundle(
|
||||
bundle,
|
||||
store,
|
||||
organization_id=organization_id,
|
||||
dry_run=False,
|
||||
confirm=plan.project.content_hash,
|
||||
)
|
||||
assert first.status == "imported"
|
||||
|
||||
second = import_bundle(
|
||||
bundle,
|
||||
store,
|
||||
organization_id=organization_id,
|
||||
dry_run=False,
|
||||
confirm=plan.project.content_hash,
|
||||
)
|
||||
assert second.status == "no_change"
|
||||
assert second.result is None
|
||||
|
||||
|
||||
@integration
|
||||
def test_reimport_of_changed_content_replaces_the_snapshot(
|
||||
store_and_org, bundle_dir: Path, rebuilt_bundle
|
||||
) -> None:
|
||||
store, organization_id = store_and_org
|
||||
first_bundle = read_bundle(bundle_dir)
|
||||
first_plan = plan_import(first_bundle, organization_id=organization_id)
|
||||
first_result = store.commit(
|
||||
first_plan, imported_at=datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
)
|
||||
|
||||
changed_dir = rebuilt_bundle("changed", mutate=True)
|
||||
changed_plan = plan_import(read_bundle(changed_dir), organization_id=organization_id)
|
||||
assert changed_plan.project.content_hash != first_plan.project.content_hash
|
||||
# Same book, so same project id; different bytes, so a different import id.
|
||||
assert changed_plan.project.external_id == first_plan.project.external_id
|
||||
second_result = store.commit(
|
||||
changed_plan, imported_at=datetime(2026, 1, 2, tzinfo=timezone.utc)
|
||||
)
|
||||
assert second_result.project_id == first_result.project_id
|
||||
assert second_result.import_id != first_result.import_id
|
||||
|
||||
import psycopg
|
||||
from psycopg.rows import dict_row
|
||||
|
||||
connection = psycopg.connect(_integration_dsn(), row_factory=dict_row)
|
||||
try:
|
||||
# The snapshot tables hold exactly one copy: the new one, not two.
|
||||
for table, expected in (
|
||||
("project_source_refs", 1),
|
||||
("translation_speakers", len(changed_plan.speakers)),
|
||||
("translation_chapters", len(changed_plan.chapters)),
|
||||
("translation_utterances", len(changed_plan.utterances)),
|
||||
):
|
||||
row = connection.execute(
|
||||
f"SELECT count(*) AS n FROM public.{table} WHERE project_id = %s",
|
||||
(first_result.project_id,),
|
||||
).fetchone()
|
||||
assert row["n"] == expected, table
|
||||
ref = connection.execute(
|
||||
"SELECT content_hash FROM public.project_source_refs WHERE project_id = %s",
|
||||
(first_result.project_id,),
|
||||
).fetchone()
|
||||
assert ref["content_hash"] == changed_plan.project.content_hash
|
||||
# The ledger is append-only and keeps both runs.
|
||||
runs = connection.execute(
|
||||
"SELECT count(*) AS n FROM public.translation_imports WHERE project_id = %s",
|
||||
(first_result.project_id,),
|
||||
).fetchone()
|
||||
assert runs["n"] == 2
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
|
||||
@integration
|
||||
def test_the_source_tables_refuse_an_update(store_and_org, bundle_dir: Path) -> None:
|
||||
"""The append-only guard is the reason a re-import DELETEs rather than edits."""
|
||||
|
||||
import psycopg
|
||||
|
||||
store, organization_id = store_and_org
|
||||
bundle = read_bundle(bundle_dir)
|
||||
plan = plan_import(bundle, organization_id=organization_id)
|
||||
result = store.commit(plan, imported_at=datetime(2026, 1, 1, tzinfo=timezone.utc))
|
||||
|
||||
connection = psycopg.connect(_integration_dsn(), autocommit=True)
|
||||
try:
|
||||
with pytest.raises(psycopg.errors.RaiseException):
|
||||
connection.execute(
|
||||
"UPDATE public.project_source_refs SET source_title = 'edit' WHERE project_id = %s",
|
||||
(result.project_id,),
|
||||
)
|
||||
finally:
|
||||
connection.close()
|
||||
Reference in New Issue
Block a user