generated from john/python-template
254 lines
11 KiB
Python
254 lines
11 KiB
Python
"""Sources list and detail page registration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from urllib.parse import urlencode
|
|
from uuid import UUID
|
|
|
|
from fastapi import Request
|
|
from fastapi.responses import RedirectResponse
|
|
from nicegui import ui
|
|
|
|
from transcription.db.models import JobSource, Source
|
|
from transcription.services.documents import DocumentError, DocumentService
|
|
from transcription.services.jobs import JobService
|
|
from transcription.services.transcription import (
|
|
TranscriptionNotFoundError,
|
|
TranscriptionService,
|
|
)
|
|
from transcription.ui.components.app_shell import render_navigation_header
|
|
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.table.sources import SourceTableRow, render_sources_table
|
|
from transcription.ui.components.typography import page_header
|
|
from transcription.ui.theme import apply_archival_theme
|
|
|
|
from ...db.session import SessionFactoryDep
|
|
|
|
|
|
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_text = request.query_params.get("document_id")
|
|
job_id_text = request.query_params.get("job_id")
|
|
|
|
document_id = _parse_uuid(document_id_text)
|
|
job_id = _parse_uuid(job_id_text)
|
|
|
|
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 = list(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 = [job_source.source for job_source in job_sources if job_source.source is not None]
|
|
sources.sort(key=lambda item: (item.page_number, item.upload_name.casefold()))
|
|
else:
|
|
sources = list(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 ui.row().classes("w-full items-center justify-between pb-2 border-b border-[#6B6A65]/20"):
|
|
if document_name is not None:
|
|
header_title = f"Sources: {document_name}"
|
|
elif job_label is not None:
|
|
header_title = f"Sources for Job {job_label}"
|
|
else:
|
|
header_title = "Archival Source Media"
|
|
|
|
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(
|
|
"bg-[#2D5A4C] text-white text-xs"
|
|
)
|
|
|
|
# Format source records into read-model rows for the table renderer
|
|
rows = [
|
|
SourceTableRow(
|
|
id=source.id,
|
|
page_number=source.page_number,
|
|
upload_name=source.upload_name,
|
|
filename=source.filename,
|
|
document_id=source.document_id,
|
|
)
|
|
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")
|
|
|
|
try:
|
|
parsed_source_id = UUID(source_id)
|
|
except ValueError:
|
|
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 ui.row().classes("w-full justify-between items-center pb-2 border-b border-[#6B6A65]/30"):
|
|
page_header(f"Source Page {source.page_number}: {source.upload_name}", subtitle=f"Source ID: {source.id}")
|
|
|
|
if back_path is not None:
|
|
back_label = (
|
|
"Back to Document"
|
|
if "document_id" in request.query_params
|
|
else "Back to Job"
|
|
if "job_id" in request.query_params
|
|
else "Back to Sources"
|
|
)
|
|
ui.button(back_label, on_click=lambda route=back_path: ui.navigate.to(route), icon="arrow_back").classes(
|
|
"bg-[#2D5A4C] text-white text-xs"
|
|
)
|
|
else:
|
|
ui.button("Back to Sources", on_click=lambda: ui.navigate.to("/sources"), icon="arrow_back").props(
|
|
"flat text-xs"
|
|
)
|
|
|
|
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"):
|
|
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",
|
|
)
|
|
|
|
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"
|
|
)
|
|
|
|
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("bg-[#2D5A4C] text-white text-xs")
|
|
|
|
@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}")
|
|
|
|
|
|
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) -> 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) -> 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 |