generated from john/python-template
UI updates continue. Focus on Sources
This commit is contained in:
Jim Lancaster
@@ -0,0 +1,222 @@
|
||||
"""Sources list and detail page registration."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from uuid import UUID
|
||||
from urllib.parse import urlencode
|
||||
|
||||
from fastapi import Request
|
||||
from fastapi.responses import RedirectResponse
|
||||
from nicegui import ui
|
||||
|
||||
from transcription.db.models import JobSource
|
||||
from transcription.db.models import Source
|
||||
from transcription.services.documents import DocumentError
|
||||
from transcription.services.documents import DocumentService
|
||||
from transcription.services.jobs import JobService
|
||||
from transcription.services.transcription import TranscriptionService
|
||||
from transcription.services.transcription import TranscriptionNotFoundError
|
||||
from transcription.ui.components.app_shell import render_navigation_header
|
||||
from transcription.ui.components.document_panzoom import render_document_panzoom
|
||||
from transcription.ui.components.error_presenter import show_error
|
||||
|
||||
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:
|
||||
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-negative")
|
||||
return
|
||||
except ValueError:
|
||||
ui.label("Job not found").classes("text-h6 text-negative")
|
||||
return
|
||||
except Exception as exc: # noqa: BLE001
|
||||
show_error(exc, title="Load failed", operation="sources.list")
|
||||
return
|
||||
|
||||
if document_name is not None:
|
||||
ui.label(f"Sources for {document_name}").classes("text-h5 text-weight-medium")
|
||||
elif job_label is not None:
|
||||
ui.label(f"Sources for Job {job_label}").classes("text-h5 text-weight-medium")
|
||||
else:
|
||||
ui.label("Sources").classes("text-h5 text-weight-medium")
|
||||
|
||||
with ui.row().classes("w-full items-center gap-2"):
|
||||
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")
|
||||
|
||||
if not sources:
|
||||
ui.label("No sources added yet.").classes("text-body2 vibe-text-muted")
|
||||
return
|
||||
|
||||
with ui.column().classes("w-full gap-2"):
|
||||
for source in sources:
|
||||
with ui.card().classes("w-full") as card:
|
||||
card.on(
|
||||
"click",
|
||||
lambda _=None, source_id=source.id, filters=_build_filter_query(document_id=document_id, job_id=job_id): ui.navigate.to(
|
||||
f"/sources/{source_id}{filters}"
|
||||
),
|
||||
)
|
||||
card.classes("cursor-pointer")
|
||||
with ui.column().classes("gap-1"):
|
||||
ui.label(f"Page {source.page_number}: {source.upload_name}").classes("text-subtitle1 text-weight-medium")
|
||||
ui.label(f"Stored filename: {source.filename}").classes("text-body2")
|
||||
ui.label(f"Document id: {source.document_id}").classes("text-body2 vibe-text-muted")
|
||||
|
||||
@ui.page("/sources/{source_id}")
|
||||
async def source_detail_page(source_id: str, request: Request, session_factory: SessionFactoryDep) -> None:
|
||||
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-negative")
|
||||
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-negative")
|
||||
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)
|
||||
ui.label(f"Source {source.upload_name}").classes("text-h5 text-weight-medium")
|
||||
|
||||
with ui.row().classes("w-full items-center gap-2"):
|
||||
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")
|
||||
else:
|
||||
ui.button("Back to Sources", on_click=lambda: ui.navigate.to("/sources"), icon="arrow_back")
|
||||
|
||||
with ui.column().classes("w-full gap-3"):
|
||||
render_document_panzoom(source=source)
|
||||
|
||||
with ui.column().classes("gap-1"):
|
||||
ui.label(f"Page number: {source.page_number}").classes("text-body2")
|
||||
ui.label(f"Upload name: {source.upload_name}").classes("text-body2")
|
||||
ui.label(f"Stored filename: {source.filename}").classes("text-body2")
|
||||
ui.label(f"Document id: {source.document_id}").classes("text-body2")
|
||||
ui.label(f"Uploaded: {source.date_uploaded.isoformat()}").classes("text-body2")
|
||||
ui.label(f"Revised: {source.date_revised.isoformat() if source.date_revised else 'not set'}").classes("text-body2")
|
||||
|
||||
ui.separator()
|
||||
ui.label("Transcription text").classes("text-subtitle1 text-weight-medium")
|
||||
ui.textarea(value=_source_transcription_text(source) or "").props("outlined autogrow readonly").classes("w-full")
|
||||
|
||||
ui.label("Revision text").classes("text-subtitle1 text-weight-medium")
|
||||
revision_input = ui.textarea(label="Revision text", value=source.revised_text or "").props("outlined autogrow")
|
||||
revision_input.classes("w-full")
|
||||
|
||||
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"):
|
||||
ui.button("Save revision", on_click=save_revision, icon="save").props('unelevated color="primary"')
|
||||
|
||||
@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 _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
|
||||
Reference in New Issue
Block a user