bbchops/transcription
1
0
Fork 0
generated from john/python-template
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

Continue troubleshooting unit tests. I think it is time to let Copilot have a crack at it.

Browse Source
This commit is contained in:
Jim Lancaster
2026-08-05 18:33:44 -05:00
parent 9219adaf0c
commit be152a028e
10 changed files with 258 additions and 922 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/transcription/db/models.py
+8 -2
View File
@@ -178,8 +178,14 @@ 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", sa_relationship_kwargs={"lazy": "selectin"})
job_sources: list["JobSource"] = Relationship(back_populates="source", sa_relationship_kwargs={"lazy": "selectin"})
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"},
)
@property
def latest_job_source(self) -> Optional["JobSource"]:
src/transcription/ui/pages/sources_page.py
+150 -367
View File
@@ -1,380 +1,163 @@
"""Sources list and detail page registration."""
"""Sources UI page module for VibeScribe."""
from __future__ import annotations
from typing import TYPE_CHECKING
from urllib.parse import urlencode
from typing import Optional
from uuid import UUID
from fastapi import Request
from fastapi.responses import RedirectResponse
from nicegui import ui
from sqlalchemy.orm import selectinload
from sqlmodel import select
from transcription.db.models import Source
from transcription.services.documents import DocumentError, DocumentService
from transcription.services.jobs import JobService
from transcription.services.transcription import (
SourceDeleteBlockedError,
TranscriptionNotFoundError,
TranscriptionService,
)
from transcription.ui.components.app_shell import render_navigation_header
from transcription.db import session_scope
from transcription.db.models import Job, Source
from transcription.ui.components.cards import archival_card
from transcription.ui.components.data_display import metadata_row
from transcription.ui.components.document_panzoom import render_document_panzoom
from transcription.ui.components.error_presenter import show_error
from transcription.ui.components.primitives import destructive_button, section_header_row
from transcription.ui.components.table.sources import SourceTableRow, render_sources_table
from transcription.ui.theme import apply_archival_theme, page_header
from ...db.session import SessionFactoryDep
if TYPE_CHECKING:
from starlette.datastructures import QueryParams
def _render_header_nav(current_path: str = "/ui/sources") -> None:
"""Render inline header navigation bar."""
with ui.row().classes("w-full items-center justify-between border-b border-slate-800 pb-4 mb-4"):
ui.label("VibeScribe").classes("text-xl font-bold text-amber-500 tracking-wider")
with ui.row().classes("gap-4"):
ui.link("Documents", "/ui/documents").classes("text-slate-300 hover:text-amber-400 font-medium")
ui.link("Jobs", "/ui/jobs").classes("text-slate-300 hover:text-amber-400 font-medium")
ui.link("Sources", "/ui/sources").classes(
"text-amber-400 font-bold" if current_path == "/ui/sources" else "text-slate-300 hover:text-amber-400 font-medium"
)
ui.link("People", "/ui/people").classes("text-slate-300 hover:text-amber-400 font-medium")
ui.link("Upload", "/ui/upload").classes("text-slate-300 hover:text-amber-400 font-medium")
async def sources_page(
document_id: Optional[str] = None,
job_id: Optional[str] = None,
) -> None:
"""Render the master or context-filtered sources page."""
parsed_doc_id: Optional[UUID] = None
parsed_job_id: Optional[UUID] = None
if document_id:
try:
parsed_doc_id = UUID(document_id)
except ValueError:
pass
if job_id:
try:
parsed_job_id = UUID(job_id)
except ValueError:
pass
async with session_scope() as session:
statement = select(Source).options(
selectinload(Source.document),
selectinload(Source.job_sources),
)
if parsed_doc_id:
statement = statement.where(Source.document_id == parsed_doc_id)
result = await session.exec(statement)
sources = list(result.all())
if parsed_job_id:
sources = [
src for src in sources if any(js.job_id == parsed_job_id for js in src.job_sources)
]
header_title = "Source Asset Records"
if parsed_doc_id:
header_title = "Sources for Document"
elif parsed_job_id:
header_title = "Sources for Job"
with ui.column().classes("w-full max-w-7xl mx-auto p-6 gap-6"):
_render_header_nav(current_path="/ui/sources")
with ui.row().classes("w-full items-center justify-between mb-2"):
with ui.column().classes("gap-1"):
ui.label(header_title).classes("text-2xl font-bold text-slate-100")
ui.label("Manage digitized source pages, raw OCR transcripts, and human revisions.").classes(
"text-sm text-slate-400"
)
if not sources:
with archival_card(title="No Source Assets Found"):
ui.label("No source images or pages match the current filter criteria.").classes(
"text-sm text-slate-400 mb-4"
)
ui.button("Upload New Documents", on_click=lambda: ui.navigate.to("/ui/upload")).props("color=amber-6")
return
columns = [
{"name": "upload_name", "label": "Upload Name", "field": "upload_name", "align": "left", "sortable": True},
{"name": "document_name", "label": "Document Context", "field": "document_name", "align": "left", "sortable": True},
{"name": "page_number", "label": "Page #", "field": "page_number", "align": "center", "sortable": True},
{"name": "status", "label": "Latest Status", "field": "status", "align": "center", "sortable": True},
{"name": "revised", "label": "Human Revised", "field": "revised", "align": "center", "sortable": True},
{"name": "actions", "label": "Actions", "field": "actions", "align": "right"},
]
rows = [
{
"id": str(source.id),
"upload_name": source.upload_name,
"document_name": source.document_name or "Unlinked Document",
"page_number": source.page_number,
"status": source.latest_status.value if source.latest_status else "Unprocessed",
"revised": "Yes" if source.revised_text else "No",
"error_detail": source.latest_error_detail,
}
for source in sources
]
with archival_card().classes("p-0 overflow-hidden"):
table = ui.table(columns=columns, rows=rows, row_key="id").classes("w-full bg-transparent text-slate-200")
table.add_slot(
"body-cell-actions",
r"""
<q-td :props="props">
<q-btn flat round dense icon="visibility" color="amber-6" @click="() => $parent.$emit('view_source', props.row.id)" />
<q-btn flat round dense icon="delete" color="red-5" @click="() => $parent.$emit('delete_source', props.row.id)" />
</q-td>
""",
)
table.on("view_source", lambda e: ui.navigate.to(f"/ui/sources/{e.args}"))
table.on("delete_source", lambda e: ui.navigate.to(f"/ui/sources/{e.args}/delete"))
async def source_detail_page(source_id: str) -> None:
"""Render individual source detail preview and revision workspace."""
with ui.column().classes("w-full max-w-7xl mx-auto p-6 gap-6"):
_render_header_nav(current_path="/ui/sources")
with archival_card(title=f"Source Record: {source_id}"):
ui.label("Source Preview & Revision Workspace").classes("text-slate-300")
async def source_delete_page(source_id: str) -> None:
"""Render source deletion confirmation workspace."""
async with session_scope() as session:
source = await session.get(Source, UUID(source_id))
is_linked = False
if source:
statement = select(Job).join(Job.job_sources).where(Job.id == Job.job_id) # simplified check or check job_sources relationship
# or check source.job_sources
if source.job_sources:
is_linked = True
with ui.column().classes("w-full max-w-7xl mx-auto p-6 gap-6"):
_render_header_nav(current_path="/ui/sources")
with archival_card(title="Delete Source Confirmation"):
if is_linked:
ui.label("Cannot delete source linked to active jobs.").classes("text-red-400")
else:
ui.label(f"Are you sure you want to delete source {source_id}?").classes("text-slate-300")
def register_page() -> None:
"""Register source list and detail routes."""
@ui.page("/sources")
async def sources_page(request: Request, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
sources_service = TranscriptionService(session_factory=session_factory)
jobs_service = JobService(session_factory=session_factory)
documents_service = DocumentService(session_factory=session_factory)
render_navigation_header(current_path="/sources")
document_id = _parse_uuid(request.query_params.get("document_id"))
job_id = _parse_uuid(request.query_params.get("job_id"))
document_name = None
job_label = None
back_path = None
sources: list[Source] = []
try:
if document_id is not None:
document = await documents_service.read_document_detail(document_id=document_id)
document_name = document.name
back_path = f"/documents/{document.id}"
sources = sorted(document.sources, key=lambda item: (item.page_number, item.upload_name.casefold()))
elif job_id is not None:
job = await jobs_service.read_job(job_id=job_id)
job_label = str(job.id)
back_path = f"/jobs/{job.id}"
job_sources = await sources_service.list_job_sources(job_id=job.id)
sources = sorted(
[js.source for js in job_sources if js.source is not None],
key=lambda item: (item.page_number, item.upload_name.casefold()),
)
else:
sources = sorted(
await sources_service.list_sources(),
key=lambda item: (item.document_id, item.page_number),
)
except DocumentError:
ui.label("Document not found").classes("text-h6 text-red-800 p-4")
return
except ValueError:
ui.label("Job not found").classes("text-h6 text-red-800 p-4")
return
except Exception as exc: # noqa: BLE001
show_error(exc, title="Load failed", operation="sources.list")
return
with ui.column().classes("w-full max-w-7xl mx-auto p-4 gap-4"):
with section_header_row():
header_title = _get_list_header_title(document_name, job_label)
page_header(header_title)
if back_path is not None:
back_label = "Back to Document" if document_id is not None else "Back to Job"
ui.button(
back_label,
on_click=lambda route=back_path: ui.navigate.to(route),
icon="arrow_back",
).classes("ui-btn-primary text-xs")
# Clean row construction leveraging SQLModel @property definitions
rows = [
SourceTableRow(
id=source.id,
page_number=source.page_number,
upload_name=source.upload_name,
filename=source.filename,
document_id=source.document_id,
document_name=source.document_name or document_name,
job_source_status=source.latest_status.value if source.latest_status else None,
job_source_error_detail=source.latest_error_detail,
)
for source in sources
]
render_sources_table(rows)
@ui.page("/sources/{source_id}")
async def source_detail_page(source_id: str, request: Request, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
sources_service = TranscriptionService(session_factory=session_factory)
render_navigation_header(current_path="/sources")
parsed_source_id = _parse_uuid(source_id)
if parsed_source_id is None:
ui.label("Invalid source id").classes("text-h6 text-red-800 p-4")
return
try:
source = await sources_service.read_source_detail(source_id=parsed_source_id)
except TranscriptionNotFoundError:
ui.label("Source not found").classes("text-h6 text-red-800 p-4")
return
except Exception as exc: # noqa: BLE001
show_error(exc, title="Load failed", operation="sources.read")
return
back_path = _back_path_from_query(request.query_params)
with ui.column().classes("w-full max-w-[1800px] mx-auto p-4 gap-4"):
with section_header_row():
page_header(f"Source Page {source.page_number}: {source.upload_name}", subtitle=f"Source ID: {source.id}")
_render_source_header_actions(source, back_path, request.query_params)
with ui.grid().classes("w-full grid-cols-12 gap-4"):
with ui.column().classes("col-span-12 lg:col-span-7 gap-4"):
with archival_card(title="Source Inspection Viewer", extra_classes="p-2"):
render_document_panzoom(source=source)
with ui.column().classes("col-span-12 lg:col-span-5 gap-4"):
_render_source_metadata(source)
_render_job_outcomes(source)
_render_raw_transcription(source)
_render_curated_transcription(source, sources_service, request)
@ui.page("/sources/{source_id}/delete")
async def source_delete_page(source_id: str, request: Request, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
sources_service = TranscriptionService(session_factory=session_factory)
render_navigation_header(current_path="/sources")
parsed_source_id = _parse_uuid(source_id)
if parsed_source_id is None:
ui.label("Invalid source id").classes("text-h6 text-red-800 p-4")
return
try:
source = await sources_service.read_source_detail(source_id=parsed_source_id)
except TranscriptionNotFoundError:
ui.label("Source not found").classes("text-h6 text-red-800 p-4")
return
except Exception as exc: # noqa: BLE001
show_error(exc, title="Load failed", operation="sources.delete.load")
return
back_path = _back_path_from_query(request.query_params) or "/sources"
next_sources_path = f"/sources{_back_query(request.query_params)}"
linked_count = len(source.job_sources)
with ui.column().classes("w-full max-w-xl mx-auto p-4 gap-4"):
page_header("Delete Source Record")
with archival_card(extra_classes="gap-2"):
metadata_row("Source ID:", str(source.id))
metadata_row("Upload Name:", source.upload_name)
metadata_row("Linked Jobs:", str(linked_count))
if linked_count > 0:
ui.label("Delete is only available for unlinked sources.").classes("text-xs text-red-800 font-bold mt-2")
ui.label("This source is linked to one or more jobs and cannot be deleted from this view.").classes(
"text-xs ui-text-muted italic"
)
else:
ui.label("This action permanently deletes the source record.").classes("text-xs text-red-800 font-medium")
async def submit_delete() -> None:
try:
await sources_service.delete_unlinked_source(source_id=source.id)
except SourceDeleteBlockedError as exc:
ui.notify(exc.message, type="warning")
return
except TranscriptionNotFoundError:
ui.notify("Source not found.", type="warning")
ui.navigate.to(next_sources_path)
return
except Exception as exc: # noqa: BLE001
show_error(exc, title="Delete source failed", operation="sources.delete")
return
ui.notify("Source deleted", type="positive")
ui.navigate.to(next_sources_path)
with ui.row().classes("w-full items-center gap-2 mt-2"):
destructive_button(
"Delete source permanently",
on_click=submit_delete,
icon="delete_forever",
variant="solid",
)
ui.button("Cancel", on_click=lambda route=back_path: ui.navigate.to(route), icon="arrow_back").props("flat")
@ui.page("/documents/{document_id}/sources")
async def document_sources_page(document_id: str) -> RedirectResponse:
return RedirectResponse(url=f"/ui/sources?document_id={document_id}")
@ui.page("/jobs/{job_id}/sources")
async def job_sources_page(job_id: str) -> RedirectResponse:
return RedirectResponse(url=f"/ui/sources?job_id={job_id}")
# --- Component Extraction Helpers ---
def _render_source_header_actions(source: Source, back_path: str | None, query_params: QueryParams) -> None:
"""Render top actions for the source detail page."""
with ui.row().classes("items-center gap-2"):
if back_path is not None:
back_label = (
"Back to Document"
if "document_id" in query_params
else "Back to Job"
if "job_id" in query_params
else "Back to Sources"
)
ui.button(
back_label,
on_click=lambda route=back_path: ui.navigate.to(route),
icon="arrow_back",
).classes("ui-btn-primary text-xs")
else:
ui.button("Back to Sources", on_click=lambda: ui.navigate.to("/sources"), icon="arrow_back").props(
"flat text-xs"
)
destructive_button(
"Delete Source",
on_click=lambda: ui.navigate.to(f"/sources/{source.id}/delete{_back_query(query_params)}"),
icon="delete",
extra_classes="text-xs",
)
def _render_source_metadata(source: Source) -> None:
"""Render standard metadata fields for a source."""
with archival_card(title="Source Metadata"):
metadata_row("Page Number:", str(source.page_number))
metadata_row("Upload Name:", source.upload_name)
metadata_row("Stored Filename:", source.filename)
metadata_row("Document ID:", str(source.document_id))
metadata_row("Date Uploaded:", source.date_uploaded.isoformat())
metadata_row(
"Date Revised:",
source.date_revised.isoformat() if source.date_revised else "Not revised",
)
def _render_job_outcomes(source: Source) -> None:
"""Render related execution outcome cards."""
with archival_card(title="Job Source Outcomes"):
if not source.job_sources:
ui.label("No job-source execution records found for this source.").classes("text-xs ui-text-muted")
return
for job_source in sorted(source.job_sources, key=lambda item: item.executed_at, reverse=True):
with ui.column().classes("w-full gap-1 p-2 ui-row-surface rounded"):
metadata_row("Job ID:", str(job_source.job_id))
metadata_row("Status:", job_source.status.value)
metadata_row("Executed At:", job_source.executed_at.isoformat())
metadata_row("Error Detail:", job_source.error_detail or "None")
def _render_raw_transcription(source: Source) -> None:
"""Render raw machine transcription output."""
with archival_card(title="Automated Raw Transcription"):
ui.textarea(value=_source_transcription_text(source) or "").props("outlined autogrow readonly bg-white").classes(
"w-full text-xs font-mono"
)
def _render_curated_transcription(source: Source, sources_service: TranscriptionService, request: Request) -> None:
"""Render human revision editing panel."""
with archival_card(title="Curated Human Transcription"):
revision_input = (
ui.textarea(value=source.revised_text or "").props("outlined autogrow bg-white").classes("w-full text-xs")
)
async def save_revision() -> None:
candidate = (revision_input.value or "").strip()
if not candidate:
ui.notify("Revision text is required.", type="warning")
return
try:
await sources_service.upsert_revision_for_source(source_id=source.id, text=candidate)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Save failed", operation="sources.save_revision")
return
ui.notify("Revision saved", type="positive")
ui.navigate.to(request.url.path + _back_query(request.query_params))
with ui.row().classes("w-full items-center gap-2 mt-2"):
ui.button("Save Revision", on_click=save_revision, icon="save").classes("ui-btn-primary text-xs")
# --- Utility Functions ---
def _get_list_header_title(document_name: str | None, job_label: str | None) -> str:
if document_name is not None:
return f"Sources: {document_name}"
if job_label is not None:
return f"Sources for Job {job_label}"
return "Archival Source Media"
def _parse_uuid(value: str | None) -> UUID | None:
if not value:
return None
try:
return UUID(value)
except ValueError:
return None
def _build_filter_query(*, document_id: UUID | None, job_id: UUID | None) -> str:
params: dict[str, str] = {}
if document_id is not None:
params["document_id"] = str(document_id)
if job_id is not None:
params["job_id"] = str(job_id)
return f"?{urlencode(params)}" if params else ""
def _source_detail_path(*, source_id: UUID, document_id: UUID | None, job_id: UUID | None) -> str:
return f"/sources/{source_id}{_build_filter_query(document_id=document_id, job_id=job_id)}"
def _back_query(query_params: QueryParams) -> str:
params = {}
for key in ("document_id", "job_id"):
if query_params.get(key):
params[key] = query_params.get(key)
return f"?{urlencode(params)}" if params else ""
def _back_path_from_query(query_params: QueryParams) -> str | None:
document_id = query_params.get("document_id")
if document_id:
return f"/documents/{document_id}"
job_id = query_params.get("job_id")
if job_id:
return f"/jobs/{job_id}"
return None
def _source_transcription_text(source: Source) -> str | None:
for job_source in sorted(source.job_sources, key=lambda item: item.executed_at, reverse=True):
if job_source.raw_transcription:
return job_source.raw_transcription
for job_source in sorted(source.job_sources, key=lambda item: item.executed_at, reverse=True):
if job_source.error_detail:
return job_source.error_detail
return None
"""Register all source-related UI routes with the application router."""
ui.page("/ui/sources")(sources_page)
ui.page("/ui/sources/{source_id}")(source_detail_page)
ui.page("/ui/sources/{source_id}/delete")(source_delete_page)