365 lines
15 KiB
Python
365 lines
15 KiB
Python
"""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",
|
|
]
|