From 61cc8a200ba7f26688093daab66f240ac8b40df9 Mon Sep 17 00:00:00 2001 From: Jim Lancaster <40281233+zoltan57@users.noreply.github.com> Date: Sat, 1 Aug 2026 16:17:38 -0500 Subject: [PATCH] V2 step 2 complete --- src/transcription/db/models.py | 51 ++++--- src/transcription/db/operations.py | 3 +- src/transcription/services/jobs.py | 11 +- src/transcription/services/store.py | 13 +- src/transcription/services/transcription.py | 132 ++++++++++-------- src/transcription/services/workflows.py | 4 +- src/transcription/ui/components/job_detail.py | 5 +- src/transcription/ui/components/transcript.py | 47 +++++-- src/transcription/ui/pages/jobs_page.py | 33 ++--- tests/conftest.py | 1 + tests/integration/test_pipeline_flow.py | 15 +- tests/services/test_job_service.py | 19 ++- tests/services/test_transcription_service.py | 30 ++-- tests/ui/conftest.py | 23 +-- 14 files changed, 227 insertions(+), 160 deletions(-) diff --git a/src/transcription/db/models.py b/src/transcription/db/models.py index d8855ab..128a9a5 100644 --- a/src/transcription/db/models.py +++ b/src/transcription/db/models.py @@ -13,6 +13,7 @@ from sqlalchemy import Column from sqlalchemy import JSON from sqlalchemy import UniqueConstraint from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm.exc import DetachedInstanceError from sqlalchemy.types import TypeDecorator from sqlmodel import Field from sqlmodel import Relationship @@ -33,6 +34,7 @@ class JSONBCompat(TypeDecorator): class JobStatus(StrEnum): QUEUED = "queued" PROCESSING = "processing" + TRANSCRIBED = "transcribed" COMPLETED = "completed" PARTIAL_SUCCESS = "partial_success" FAILED = "failed" @@ -63,9 +65,9 @@ class Document(SQLModel, table=True): created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) - jobs: list["Job"] = Relationship(back_populates="document") - sources: list["Source"] = Relationship(back_populates="document") - document_people: list["DocumentPerson"] = Relationship(back_populates="document") + jobs: list["Job"] = Relationship(back_populates="document", sa_relationship_kwargs={"lazy": "selectin"}) + sources: list["Source"] = Relationship(back_populates="document", sa_relationship_kwargs={"lazy": "selectin"}) + document_people: list["DocumentPerson"] = Relationship(back_populates="document", sa_relationship_kwargs={"lazy": "selectin"}) class Person(SQLModel, table=True): @@ -90,7 +92,7 @@ class Person(SQLModel, table=True): created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) - document_people: list["DocumentPerson"] = Relationship(back_populates="person") + document_people: list["DocumentPerson"] = Relationship(back_populates="person", sa_relationship_kwargs={"lazy": "selectin"}) class DocumentPerson(SQLModel, table=True): @@ -108,8 +110,8 @@ class DocumentPerson(SQLModel, table=True): UniqueConstraint("document_id", "person_id", "role", name="uq_document_person_role"), ) - document: Optional["Document"] = Relationship(back_populates="document_people") - person: Optional["Person"] = Relationship(back_populates="document_people") + document: Optional["Document"] = Relationship(back_populates="document_people", sa_relationship_kwargs={"lazy": "selectin"}) + person: Optional["Person"] = Relationship(back_populates="document_people", sa_relationship_kwargs={"lazy": "selectin"}) class Job(SQLModel, table=True): @@ -125,15 +127,29 @@ class Job(SQLModel, table=True): model: str | None = None prompt_name: str | None = None - document: Optional["Document"] = Relationship(back_populates="jobs") - job_sources: list["JobSource"] = Relationship(back_populates="job") + document: Optional["Document"] = Relationship(back_populates="jobs", sa_relationship_kwargs={"lazy": "selectin"}) + job_sources: list["JobSource"] = Relationship(back_populates="job", sa_relationship_kwargs={"lazy": "selectin"}) @property def filename(self) -> str: """Return the filename of the associated source, when available.""" if not self.job_sources: return "unknown" - return self.job_sources[0].source.filename if self.job_sources[0].source is not None else "unknown" + + for job_source in self.job_sources: + source = job_source.__dict__.get("source") + if source is None: + try: + source = job_source.source + except DetachedInstanceError: + source = None + except Exception: # noqa: BLE001 + source = None + + if source is not None: + return source.filename + + return "unknown" class Source(SQLModel, table=True): @@ -141,7 +157,7 @@ class Source(SQLModel, table=True): id: UUID = Field(default_factory=uuid4, primary_key=True) document_id: UUID = Field(foreign_key="document.id") - page_number: int = Field(ge=1) + page_number: int = Field(default=1, ge=1) upload_name: str filename: str file_path: str @@ -150,8 +166,8 @@ class Source(SQLModel, table=True): date_uploaded: datetime = Field(default_factory=lambda: datetime.now(UTC)) date_revised: datetime | None = None - document: Optional["Document"] = Relationship(back_populates="sources") - job_sources: list["JobSource"] = Relationship(back_populates="source") + document: Optional["Document"] = Relationship(back_populates="sources", sa_relationship_kwargs={"lazy": "selectin"}) + job_sources: list["JobSource"] = Relationship(back_populates="source", sa_relationship_kwargs={"lazy": "selectin"}) class JobSource(SQLModel, table=True): @@ -169,14 +185,7 @@ class JobSource(SQLModel, table=True): error_detail: str | None = None executed_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) - job: Optional["Job"] = Relationship(back_populates="job_sources") - source: Optional["Source"] = Relationship(back_populates="job_sources") + job: Optional["Job"] = Relationship(back_populates="job_sources", sa_relationship_kwargs={"lazy": "selectin"}) + source: Optional["Source"] = Relationship(back_populates="job_sources", sa_relationship_kwargs={"lazy": "selectin"}) -class Revision(SQLModel): - """Temporary compatibility shim for older service imports during the V2 migration.""" - - id: UUID = Field(default_factory=uuid4, primary_key=True) - source_id: UUID = Field(default_factory=uuid4) - text: str | None = None - date_created: datetime = Field(default_factory=lambda: datetime.now(UTC)) diff --git a/src/transcription/db/operations.py b/src/transcription/db/operations.py index 7663745..c938423 100644 --- a/src/transcription/db/operations.py +++ b/src/transcription/db/operations.py @@ -15,12 +15,13 @@ logger = logging.getLogger(__name__) async def create_all(*, engine: AsyncEngine | None = None) -> None: - """Create all tables on the selected engine.""" + """Create all tables on the selected engine from scratch.""" # Import models so SQLModel metadata is fully registered before bootstrap. from transcription.db import models as _models # noqa: F401 active_engine = engine or resolve_engine() async with active_engine.begin() as connection: + await connection.run_sync(SQLModel.metadata.drop_all) await connection.run_sync(SQLModel.metadata.create_all) logger.debug("Database schema bootstrap complete for database_url=%s", active_engine.url) diff --git a/src/transcription/services/jobs.py b/src/transcription/services/jobs.py index a0eabff..71b330a 100644 --- a/src/transcription/services/jobs.py +++ b/src/transcription/services/jobs.py @@ -8,6 +8,7 @@ from sqlmodel import select from sqlmodel.ext.asyncio.session import AsyncSession from ..db.models import Job +from ..db.models import JobSource from ..db.models import JobStatus from ..db.models import Source from .base import ServiceBase @@ -38,7 +39,7 @@ class JobService(ServiceBase): select(Job) .options( selectinload(Job.document), # pyright: ignore[reportArgumentType] - selectinload(Job.sources).selectinload(Source.revision), # pyright: ignore[reportArgumentType] + selectinload(Job.job_sources).selectinload(JobSource.source), # pyright: ignore[reportArgumentType] ) .where(Job.id == job_id) .execution_options(populate_existing=True) @@ -74,12 +75,12 @@ class JobService(ServiceBase): async with self._session_scope(session) as _session: query = select(Job).options( selectinload(Job.document), # pyright: ignore[reportArgumentType] - selectinload(Job.sources), # pyright: ignore[reportArgumentType] + selectinload(Job.job_sources).selectinload(JobSource.source), # pyright: ignore[reportArgumentType] ) if status is not None: query = query.where(Job.status == status) if filename is not None: - query = query.where(Job.sources.any(Source.filename == filename)) + query = query.where(Job.job_sources.any(JobSource.source.has(Source.filename == filename))) result = await _session.exec(query) return result.all() @@ -94,7 +95,7 @@ class JobService(ServiceBase): async with self._session_scope(session) as _session: query = select(Job).options( selectinload(Job.document), # pyright: ignore[reportArgumentType] - selectinload(Job.sources), # pyright: ignore[reportArgumentType] + selectinload(Job.job_sources).selectinload(JobSource.source), # pyright: ignore[reportArgumentType] ) result = await _session.exec(query) return result.all() @@ -151,7 +152,7 @@ class JobService(ServiceBase): select(Job) .options( selectinload(Job.document), # pyright: ignore[reportArgumentType] - selectinload(Job.sources).selectinload(Source.revision), # pyright: ignore[reportArgumentType] + selectinload(Job.job_sources).selectinload(JobSource.source), # pyright: ignore[reportArgumentType] ) .where(Job.status == JobStatus.QUEUED) .order_by(Job.date_created) # pyright: ignore[reportArgumentType] diff --git a/src/transcription/services/store.py b/src/transcription/services/store.py index 21e4fc0..83eb506 100644 --- a/src/transcription/services/store.py +++ b/src/transcription/services/store.py @@ -13,6 +13,8 @@ from transcription.errors import ErrorCategory from ..db.models import Document from ..db.models import Job +from ..db.models import JobSource +from ..db.models import JobSourceStatus from ..db.models import Source from .documents import UploadJobResult @@ -81,12 +83,21 @@ async def _create_upload_records( source = Source( document_id=document.id, - job_id=job.id, + page_number=1, upload_name=Path(original_filename).name, filename=stored_path.name, file_path=str(stored_path), ) session.add(source) + await session.flush() + + session.add( + JobSource( + job_id=job.id, + source_id=source.id, + status=JobSourceStatus.PENDING, + ) + ) await session.commit() await session.refresh(document) diff --git a/src/transcription/services/transcription.py b/src/transcription/services/transcription.py index f26b5e3..90d1474 100644 --- a/src/transcription/services/transcription.py +++ b/src/transcription/services/transcription.py @@ -19,7 +19,8 @@ from sqlmodel.ext.asyncio.session import AsyncSession from transcription.config import Settings from transcription.config import get_settings from transcription.db.models import Job -from transcription.db.models import Revision +from transcription.db.models import JobSource +from transcription.db.models import JobSourceStatus from transcription.db.models import Source from transcription.errors import AppError from transcription.errors import ErrorCategory @@ -51,7 +52,7 @@ class TranscriptionNotFoundError(TranscriptionError): class TranscriptionService(ServiceBase): - """Service class for job transcription output and optional source revisions.""" + """Service class for job transcription output and page-level source revisions.""" provider: TranscriptionProvider @@ -59,52 +60,47 @@ class TranscriptionService(ServiceBase): super().__init__(session_factory=session_factory) self.provider = get_transcription_provider(settings=self.settings) - async def create_revision(self, revision: Revision, *, session: AsyncSession | None = None) -> Revision: - """Create a new revision in the database.""" + async def create_job_source( + self, + job_source: JobSource, + *, + session: AsyncSession | None = None, + ) -> JobSource: + """Create a new job_source execution record in the database.""" async with self._session_scope(session) as _session: - _session.add(revision) - await self._finalize(session=_session, caller_session=session, refresh=(revision,)) - return revision + _session.add(job_source) + await self._finalize(session=_session, caller_session=session, refresh=(job_source,)) + return job_source - async def read_revision(self, revision_id: UUID, *, session: AsyncSession | None = None) -> Revision: - """Read an existing revision from the database.""" + async def read_job_source(self, job_source_id: UUID, *, session: AsyncSession | None = None) -> JobSource: + """Read an existing job_source record.""" async with self._session_scope(session) as _session: - revision = await _session.get( - Revision, - revision_id, - options=(selectinload(Revision.source),), # pyright: ignore[reportArgumentType] + job_source = await _session.get( + JobSource, + job_source_id, + options=(selectinload(JobSource.source),), # pyright: ignore[reportArgumentType] ) - if revision is None: + if job_source is None: raise TranscriptionNotFoundError( - f"Revision with id {revision_id} not found", + f"JobSource with id {job_source_id} not found", category=ErrorCategory.NOT_FOUND, - suggestion="Verify the revision id and retry.", + suggestion="Verify the job source id and retry.", ) - return revision + return job_source - async def update_revision(self, revision: Revision, *, session: AsyncSession | None = None) -> Revision: - """Update an existing revision in the database.""" + async def update_job_source(self, job_source: JobSource, *, session: AsyncSession | None = None) -> JobSource: + """Update an existing job_source record.""" async with self._session_scope(session) as _session: - merged = await _session.merge(revision) + merged = await _session.merge(job_source) await self._finalize(session=_session, caller_session=session, refresh=(merged,)) return merged - async def delete_revision(self, revision: Revision, *, session: AsyncSession | None = None) -> None: - """Delete a revision from the database.""" + async def delete_job_source(self, job_source: JobSource, *, session: AsyncSession | None = None) -> None: + """Delete a job_source record.""" async with self._session_scope(session) as _session: - await _session.delete(revision) + await _session.delete(job_source) await self._finalize(session=_session, caller_session=session) - # Temporary compatibility methods for callers still using transcript naming. - - async def read_transcript(self, transcript_id: UUID, *, session: AsyncSession | None = None) -> Revision: - """Backward-compatible alias for read_revision.""" - return await self.read_revision(transcript_id, session=session) - - async def delete_transcript(self, transcript: Revision, *, session: AsyncSession | None = None) -> None: - """Backward-compatible alias for delete_revision.""" - await self.delete_revision(transcript, session=session) - async def transcribe_document( self, image_path: str | Path, @@ -151,13 +147,37 @@ class TranscriptionService(ServiceBase): suggestion="Verify the job id and retry.", ) - job.text = text - job.error_detail = error_detail job.provider = provider or job.provider or self.settings.provider.value job.model = model or job.model or _resolve_transcript_model(provider=self.provider, settings=self.settings) job.prompt_name = prompt_name or job.prompt_name or DEFAULT_PROMPT_FILE job.date_updated = datetime.now(UTC) + source = await _session.exec( + select(Source) + .where(Source.document_id == job.document_id) + .order_by(Source.page_number) # pyright: ignore[reportArgumentType] + ) + source_row = source.first() + if source_row is not None: + existing_job_source = await _session.exec( + select(JobSource).where(JobSource.job_id == job.id).where(JobSource.source_id == source_row.id) + ) + job_source = existing_job_source.first() + if job_source is None: + job_source = JobSource( + job_id=job.id, + source_id=source_row.id, + status=JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED, + raw_transcription=text, + error_detail=error_detail, + ) + _session.add(job_source) + else: + job_source.raw_transcription = text + job_source.error_detail = error_detail + job_source.status = JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED + job_source.executed_at = datetime.now(UTC) + await self._finalize(session=_session, caller_session=session, refresh=(job,)) return job @@ -167,8 +187,8 @@ class TranscriptionService(ServiceBase): source_id: UUID, text: str, session: AsyncSession | None = None, - ) -> Revision: - """Create or replace the single optional revision for a source.""" + ) -> Source: + """Persist a human revision on a source page.""" async with self._session_scope(session) as _session: source = await _session.get(Source, source_id) if source is None: @@ -178,45 +198,35 @@ class TranscriptionService(ServiceBase): suggestion="Verify the source id and retry.", ) - query = select(Revision).where(Revision.source_id == source_id) - existing = (await _session.exec(query)).one_or_none() - - if existing is None: - revision = Revision(source_id=source_id, text=text) - _session.add(revision) - await self._finalize(session=_session, caller_session=session, refresh=(revision,)) - return revision - - existing.text = text - merged = await _session.merge(existing) - await self._finalize(session=_session, caller_session=session, refresh=(merged,)) - return merged + source.revised_text = text + source.date_revised = datetime.now(UTC) + await self._finalize(session=_session, caller_session=session, refresh=(source,)) + return source async def read_revision_by_source( self, source_id: UUID, *, session: AsyncSession | None = None, - ) -> Revision | None: - """Read the single optional revision for a source.""" + ) -> Source | None: + """Read the source record for a given page, including any revision text.""" async with self._session_scope(session) as _session: - query = select(Revision).where(Revision.source_id == source_id) - result = await _session.exec(query) - return result.one_or_none() + return await _session.get(Source, source_id) async def list_revisions_by_job( self, job_id: UUID, *, session: AsyncSession | None = None, - ) -> Sequence[Revision]: - """List revisions connected to all sources for a job.""" + ) -> Sequence[Source]: + """List source pages for a job that carry revision text.""" async with self._session_scope(session) as _session: query = ( - select(Revision) - .join(Source, Source.id == Revision.source_id) - .where(Source.job_id == job_id) - .order_by(Revision.date_created) # pyright: ignore[reportArgumentType] + select(Source) + .join(JobSource, JobSource.source_id == Source.id) + .where(JobSource.job_id == job_id) + .where(Source.revised_text.is_not(None)) + .order_by(Source.date_revised) # pyright: ignore[reportArgumentType] ) result = await _session.exec(query) return result.all() diff --git a/src/transcription/services/workflows.py b/src/transcription/services/workflows.py index 611ecd0..d85b45b 100644 --- a/src/transcription/services/workflows.py +++ b/src/transcription/services/workflows.py @@ -291,9 +291,9 @@ async def _finalize_failed( def _resolve_primary_source(job: Job) -> Source | None: - if not job.sources: + if not job.job_sources: return None - return job.sources[0] + return next((job_source.source for job_source in job.job_sources if job_source.source is not None), None) def _validate_transcription_quality(*, result: TranscriptionResult, settings: Settings) -> None: diff --git a/src/transcription/ui/components/job_detail.py b/src/transcription/ui/components/job_detail.py index 8b35618..3301454 100644 --- a/src/transcription/ui/components/job_detail.py +++ b/src/transcription/ui/components/job_detail.py @@ -7,7 +7,6 @@ import logging from nicegui import ui from transcription.db.models import Job -from transcription.db.models import Revision from transcription.db.models import Source from transcription.ui.components.document_panzoom import render_document_panzoom from transcription.ui.components.transcript import render_original_transcription_card @@ -48,7 +47,7 @@ def _render_source_section(source: Source) -> None: render_document_panzoom(source=source) -def _render_revision_section(revision: Revision | None) -> None: +def _render_revision_section(revision: Source | None) -> None: with ui.card().classes("w-full q-pa-md vibe-card"): ui.label("Revision").classes("text-subtitle1 text-weight-medium") ui.separator().classes("q-my-sm") @@ -60,7 +59,7 @@ def _render_revision_section(revision: Revision | None) -> None: render_revision_row(revision=revision, initially_expanded=True) -def render_job_detail(*, job: Job, source: Source | None, revision: Revision | None) -> None: +def render_job_detail(*, job: Job, source: Source | None, revision: Source | None) -> None: """Render all sections for the job detail page.""" logger.debug("Rendering job detail for job ID %s", job.id) status_text = job.status.value diff --git a/src/transcription/ui/components/transcript.py b/src/transcription/ui/components/transcript.py index c7933b0..6cd3cf9 100644 --- a/src/transcription/ui/components/transcript.py +++ b/src/transcription/ui/components/transcript.py @@ -10,14 +10,16 @@ from typing import Any from nicegui import ui from transcription.db.models import Job -from transcription.db.models import Revision +from transcription.db.models import JobSource +from transcription.db.models import Source -type RevisionAction = Callable[[Revision], Awaitable[None] | None] +type RevisionAction = Callable[[Source], Awaitable[None] | None] def render_original_transcription_card(*, job: Job, classes: str = "w-full") -> Any: """Render the immutable original job transcription output.""" - status_label = "Failed" if job.error_detail else "Transcribed" + latest_error_detail = _latest_job_error_detail(job) + status_label = "Failed" if latest_error_detail else "Transcribed" header = f"Original Transcription | {status_label}" provider = job.provider or "unknown" model = job.model or "unknown" @@ -30,28 +32,33 @@ def render_original_transcription_card(*, job: Job, classes: str = "w-full") -> _metadata_row(label="Prompt", value=job.prompt_name or "unknown") _metadata_row(label="Updated", value=_format_created_at(job.date_updated)) - if job.text: - with ui.card().classes("w-full q-pa-sm"): - ui.markdown(job.text) + latest_transcription = _latest_job_transcription(job) - if job.error_detail: + if latest_transcription: + with ui.card().classes("w-full q-pa-sm"): + ui.markdown(latest_transcription) + + if latest_error_detail: with ui.card().classes("w-full vibe-card--error q-pa-sm"): ui.label("Failure detail").classes("text-caption text-uppercase") - ui.label(job.error_detail).classes("text-body2") + ui.label(latest_error_detail).classes("text-body2") return card def render_revision_row( *, - revision: Revision, + revision: Source | None, initially_expanded: bool = False, classes: str = "w-full", on_delete: RevisionAction | None = None, ) -> Any: """Render a collapsible row for the single optional source revision.""" + if revision is None: + return None + header = "Revision | User-authored" - caption = _format_created_at(revision.date_created) + caption = _format_created_at(revision.date_revised or revision.date_uploaded) expansion = ui.expansion(value=initially_expanded, group="group").classes(f"{classes} rounded-borders vibe-card") @@ -84,15 +91,29 @@ def render_revision_row( ui.button(icon="delete", on_click=delete_current_transcript).props( 'flat round dense color="negative"' ) - _metadata_row(label="Created", value=_format_created_at(revision.date_created)) + _metadata_row(label="Created", value=_format_created_at(revision.date_revised or revision.date_uploaded)) - if revision.text: + if revision.revised_text: with ui.card().classes("w-full q-pa-sm"): - ui.markdown(revision.text) + ui.markdown(revision.revised_text) return expansion +def _latest_job_transcription(job: Job) -> str | None: + for job_source in sorted(job.job_sources, key=lambda item: item.executed_at, reverse=True): + if job_source.raw_transcription: + return job_source.raw_transcription + return None + + +def _latest_job_error_detail(job: Job) -> str | None: + for job_source in sorted(job.job_sources, key=lambda item: item.executed_at, reverse=True): + if job_source.error_detail: + return job_source.error_detail + return None + + def _format_created_at(value: datetime) -> str: """Return a compact UTC-like timestamp for row captions.""" return value.strftime("%Y-%m-%d %H:%M:%S %Z") diff --git a/src/transcription/ui/pages/jobs_page.py b/src/transcription/ui/pages/jobs_page.py index 63920ee..795e130 100644 --- a/src/transcription/ui/pages/jobs_page.py +++ b/src/transcription/ui/pages/jobs_page.py @@ -87,17 +87,6 @@ def register_page() -> None: # noqa: PLR0915 render_original_transcription_card(job=job) - async def delete_revision_by_id(revision_id: UUID) -> None: - try: - revision = await transcription_service.read_revision(revision_id=revision_id) - await transcription_service.delete_revision(revision) - except Exception as exc: # noqa: BLE001 - show_error(exc, title="Delete failed", operation="jobs.delete_revision") - return - - ui.notify("Deleted revision", type="positive") - await render_revision_panel.refresh() - @ui.refreshable async def render_revision_panel() -> None: refreshed_job = await jobs_service.read_job(job_id=parsed_job_id) @@ -106,10 +95,8 @@ def register_page() -> None: # noqa: PLR0915 ui.label("No source is available for revision editing.").classes("text-body2 vibe-text-muted") return - current_revision = refreshed_source.revision - default_revision_text = ( - current_revision.text if current_revision is not None else (refreshed_job.text or "") - ) + current_revision_text = refreshed_source.revised_text + default_revision_text = current_revision_text or "" ui.label("Revision Editor").classes("text-subtitle1 text-weight-medium") editor = ui.textarea(label="Revision text", value=default_revision_text).props("autogrow outlined") @@ -135,25 +122,29 @@ def register_page() -> None: # noqa: PLR0915 with ui.row().classes("w-full justify-end"): ui.button( - "Create revision" if current_revision is None else "Update revision", + "Create revision" if current_revision_text is None else "Update revision", on_click=save_revision, icon="save", ).props('unelevated color="primary"') - if current_revision is None: + if current_revision_text is None: ui.label("No revision exists for this source.").classes("text-body2 vibe-text-muted") return render_revision_row( - revision=current_revision, + revision=refreshed_source, initially_expanded=True, - on_delete=lambda _revision, rid=current_revision.id: delete_revision_by_id(rid), ) await render_revision_panel() def _resolve_primary_source(job: Job) -> Source | None: - if not job.sources: + if not job.job_sources: return None - return job.sources[0] + + for job_source in job.job_sources: + if job_source.source is not None: + return job_source.source + + return None diff --git a/tests/conftest.py b/tests/conftest.py index d2edb0b..6f8db3c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -31,6 +31,7 @@ def session(): connect_args={"check_same_thread": False}, poolclass=StaticPool, ) + SQLModel.metadata.drop_all(engine) SQLModel.metadata.create_all(engine) with Session(engine) as sync_session: yield sync_session diff --git a/tests/integration/test_pipeline_flow.py b/tests/integration/test_pipeline_flow.py index 7303d30..111bb42 100644 --- a/tests/integration/test_pipeline_flow.py +++ b/tests/integration/test_pipeline_flow.py @@ -64,8 +64,8 @@ class TestPipelineSuccessFlow: assert processed is True assert job is not None assert job.status == JobStatus.TRANSCRIBED - assert job.text == "Pipeline transcript" - assert job.error_detail is None + assert any(job_source.raw_transcription == "Pipeline transcript" for job_source in job.job_sources) + assert all(job_source.error_detail is None for job_source in job.job_sources) @pytest.mark.integration @@ -104,8 +104,9 @@ class TestPipelineFailureFlow: assert processed is True assert job is not None assert job.status == JobStatus.FAILED - assert job.text is None - assert job.error_detail is not None - assert "pipeline provider failure" in job.error_detail - assert "[internal_unexpected_error]" in job.error_detail - assert "error_id=" in job.error_detail + assert all(job_source.raw_transcription is None for job_source in job.job_sources) + assert any(job_source.error_detail is not None for job_source in job.job_sources) + error_detail = next(job_source.error_detail for job_source in job.job_sources if job_source.error_detail is not None) + assert "pipeline provider failure" in error_detail + assert "[internal_unexpected_error]" in error_detail + assert "error_id=" in error_detail diff --git a/tests/services/test_job_service.py b/tests/services/test_job_service.py index 5acd0e4..4a9df72 100644 --- a/tests/services/test_job_service.py +++ b/tests/services/test_job_service.py @@ -4,6 +4,8 @@ import pytest from transcription.db.models import Document from transcription.db.models import Job +from transcription.db.models import JobSource +from transcription.db.models import JobSourceStatus from transcription.db.models import JobStatus from transcription.db.models import Source from transcription.services.documents import DocumentService @@ -66,13 +68,20 @@ class TestJobService: await job_service.create_job(job=job) async with job_service._session_scope() as session: + source = Source( + document_id=document.id, + page_number=1, + upload_name="letter.jpg", + filename="stored-letter.jpg", + file_path="/uploads/stored-letter.jpg", + ) + session.add(source) + await session.flush() session.add( - Source( - document_id=document.id, + JobSource( job_id=job.id, - upload_name="letter.jpg", - filename="stored-letter.jpg", - file_path="/uploads/stored-letter.jpg", + source_id=source.id, + status=JobSourceStatus.PENDING, ) ) await session.commit() diff --git a/tests/services/test_transcription_service.py b/tests/services/test_transcription_service.py index 2ffd85b..781f19b 100644 --- a/tests/services/test_transcription_service.py +++ b/tests/services/test_transcription_service.py @@ -1,4 +1,4 @@ -"""Tests for revision behavior in TranscriptionService.""" +"""Tests for source revision behavior in TranscriptionService.""" from uuid import uuid4 @@ -6,6 +6,8 @@ import pytest from transcription.db.models import Document from transcription.db.models import Job +from transcription.db.models import JobSource +from transcription.db.models import JobSourceStatus from transcription.db.models import JobStatus from transcription.db.models import Source from transcription.services.documents import DocumentService @@ -15,7 +17,7 @@ from transcription.services.transcription import TranscriptionService @pytest.mark.integration class TestTranscriptionServiceRevisionUpsert: - """Verify optional single-revision create/update semantics.""" + """Verify page-level source revision semantics.""" @pytest.mark.asyncio async def test_upsert_revision_creates_new_revision(self, default_session_factory): @@ -26,29 +28,31 @@ class TestTranscriptionServiceRevisionUpsert: document = Document(id=uuid4(), name="revision-create") await documents.create_document(document=document) - job = Job(document_id=document.id, status=JobStatus.TRANSCRIBED, text="Original text") + job = Job(document_id=document.id, status=JobStatus.TRANSCRIBED) await jobs.create_job(job=job) source = Source( document_id=document.id, - job_id=job.id, + page_number=1, upload_name="source.jpg", filename="source.jpg", file_path="uploads/source.jpg", ) async with transcriptions._session_scope() as session: session.add(source) + await session.flush() + session.add(JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING)) await session.commit() await session.refresh(source) revision = await transcriptions.upsert_revision_for_source(source_id=source.id, text="User revision") fetched = await transcriptions.read_revision_by_source(source.id) - assert revision.source_id == source.id - assert revision.text == "User revision" + assert revision.id == source.id + assert revision.revised_text == "User revision" assert fetched is not None - assert fetched.id == revision.id - assert fetched.text == "User revision" + assert fetched.id == source.id + assert fetched.revised_text == "User revision" @pytest.mark.asyncio async def test_upsert_revision_updates_existing_single_revision(self, default_session_factory): @@ -59,18 +63,20 @@ class TestTranscriptionServiceRevisionUpsert: document = Document(id=uuid4(), name="revision-update") await documents.create_document(document=document) - job = Job(document_id=document.id, status=JobStatus.TRANSCRIBED, text="Original text") + job = Job(document_id=document.id, status=JobStatus.TRANSCRIBED) await jobs.create_job(job=job) source = Source( document_id=document.id, - job_id=job.id, + page_number=1, upload_name="source.jpg", filename="source.jpg", file_path="uploads/source.jpg", ) async with transcriptions._session_scope() as session: session.add(source) + await session.flush() + session.add(JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING)) await session.commit() await session.refresh(source) @@ -79,7 +85,7 @@ class TestTranscriptionServiceRevisionUpsert: revisions = await transcriptions.list_revisions_by_job(job.id) assert first.id == second.id - assert second.text == "Revision v2" + assert second.revised_text == "Revision v2" assert len(revisions) == 1 assert revisions[0].id == first.id - assert revisions[0].text == "Revision v2" + assert revisions[0].revised_text == "Revision v2" diff --git a/tests/ui/conftest.py b/tests/ui/conftest.py index 71bdf8e..c02dbc8 100644 --- a/tests/ui/conftest.py +++ b/tests/ui/conftest.py @@ -5,6 +5,8 @@ from __future__ import annotations import asyncio from collections.abc import Callable from collections.abc import Generator +from datetime import UTC +from datetime import datetime from pathlib import Path from uuid import UUID @@ -21,8 +23,9 @@ from transcription.db import initialize_database_runtime from transcription.db import session_scope from transcription.db.models import Document from transcription.db.models import Job +from transcription.db.models import JobSource +from transcription.db.models import JobSourceStatus from transcription.db.models import JobStatus -from transcription.db.models import Revision from transcription.db.models import Source RevisionSeed = str @@ -55,7 +58,6 @@ def clear_ui_database(app_client: tuple[FastAPI, TestClient]) -> None: async def _clear() -> None: async with session_scope() as session: - await session.exec(delete(Revision)) await session.exec(delete(Source)) await session.exec(delete(Job)) await session.exec(delete(Document)) @@ -94,8 +96,6 @@ def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., UUID]: document_id=document.id, status=status, retry_count=0, - text=transcription_text, - error_detail=error_detail, provider="openrouter", model="google/gemini-2.5-flash", prompt_name="transcribe_document.md", @@ -105,7 +105,6 @@ def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., UUID]: source = Source( document_id=document.id, - job_id=job.id, upload_name=filename, filename=filename, file_path=str(stored_path), @@ -113,14 +112,22 @@ def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., UUID]: session.add(source) await session.flush() - if revision_text is not None: + if transcription_text is not None or error_detail is not None: session.add( - Revision( + JobSource( + job_id=job.id, source_id=source.id, - text=revision_text, + status=JobSourceStatus.TRANSCRIBED if transcription_text is not None else JobSourceStatus.FAILED, + raw_transcription=transcription_text, + error_detail=error_detail, ) ) + if revision_text is not None: + source.revised_text = revision_text + source.date_revised = datetime.now(UTC) + session.add(source) + await session.commit() return job.id