split up docs pages

This commit is contained in:
John Lancaster
2026-08-03 22:25:07 -05:00
parent 759d4c2434
commit edb6967888
13 changed files with 668 additions and 605 deletions
+2 -3
View File
@@ -12,10 +12,9 @@ def create_cli_app() -> FastAPI:
def main() -> None:
settings = parse_cli_settings()
application = "transcription.__main__:create_cli_app" if settings.reload else create_app(settings=settings)
uvicorn.run(
application,
factory=settings.reload,
"transcription.__main__:create_cli_app",
factory=True,
host=settings.host,
port=settings.port,
log_level=settings.log_level,
+3
View File
@@ -13,6 +13,7 @@ from fastapi import FastAPI
from fastapi import status
from fastapi.responses import RedirectResponse
from fastapi.staticfiles import StaticFiles
from nicegui import ui
from .api.errors import register_error_handlers
from .api.health import router as health_router
@@ -25,6 +26,7 @@ from .db import initialize_database_runtime
from .services import ServiceBundle
from .services.jobs import JobService
from .ui import register_pages
from .ui.theme import THEME_COLORS
from .worker import worker_consumer_lifespan
logger = logging.getLogger(__name__)
@@ -99,4 +101,5 @@ def create_app(settings: Settings | None = None) -> FastAPI:
register_error_handlers(app)
register_pages(app)
app.include_router(health_router)
ui.colors(**THEME_COLORS)
return app
+10
View File
@@ -3,6 +3,7 @@
from dataclasses import dataclass
from dataclasses import field
from ..db.session import SessionFactory
from .documents import DocumentService
from .jobs import JobService
from .transcription import TranscriptionService
@@ -17,3 +18,12 @@ class ServiceBundle:
documents: DocumentService = field(default_factory=DocumentService)
jobs: JobService = field(default_factory=JobService)
transcriptions: TranscriptionService = field(default_factory=TranscriptionService)
@classmethod
def from_session_factory(cls, session_factory: SessionFactory) -> "ServiceBundle":
"""Create a ServiceBundle from a session factory."""
return cls(
documents=DocumentService(session_factory=session_factory),
jobs=JobService(session_factory=session_factory),
transcriptions=TranscriptionService(session_factory=session_factory),
)
+2 -2
View File
@@ -3,7 +3,7 @@
from fastapi import FastAPI
from nicegui import ui
from transcription.ui.pages.documents_page import register_page as register_documents_page
from transcription.ui.pages.documents import register_pages as register_documents_pages
from transcription.ui.pages.jobs_page import register_page as register_jobs_page
from transcription.ui.pages.people_page import register_page as register_people_page
from transcription.ui.pages.sources_page import register_page as register_sources_page
@@ -26,7 +26,7 @@ def register_pages(app: FastAPI) -> None:
"""Register all NiceGUI pages and mount them onto the FastAPI app."""
_register_global_styles(app)
register_upload_page()
register_documents_page()
register_documents_pages()
register_people_page()
register_sources_page()
register_jobs_page()
@@ -0,0 +1,62 @@
"""Documents list and detail page registration."""
from __future__ import annotations
from nicegui import ui
from transcription.db.models import Document
from transcription.db.models import DocumentPerson
from .cards import archival_card
from .data_display import archival_badge
from .data_display import metadata_row
from .primitives import render_empty_state
def render_archival_metadata(document: Document, author_link: DocumentPerson | None = None) -> None:
with ui.column().classes("col-span-12 lg:col-span-4 gap-4"):
with archival_card(title="Archival Metadata"):
metadata_row("Author:", author_link.person.full_name if author_link and author_link.person else "Not set")
metadata_row("Exact Date:", document.document_date.isoformat() if document.document_date else "Not set")
metadata_row("Approx. Date:", document.document_date_raw or "Not set")
metadata_row("Location Created:", document.location_created or "Not set")
metadata_row("Archive Identifier:", document.archive_identifier or "Not set")
with ui.column().classes("w-full mt-2"):
ui.label("Archival Notes:").classes("ui-text-muted text-xs mb-1")
ui.label(document.notes or "No notes added.").classes("p-2 ui-note-box text-xs")
with archival_card(title="System Logistics"):
ui.label(f"Created: {document.created_at.isoformat()}").classes("text-[11px] ui-text-muted")
ui.label(f"Updated: {document.updated_at.isoformat()}").classes("text-[11px] ui-text-muted")
def render_doc_people_details(document: Document) -> None:
with archival_card(title="Related People"):
if not document.document_people:
render_empty_state("No linked people yet.", italic=True)
else:
with ui.column().classes("w-full gap-2"):
for link in document.document_people:
person_label = link.person.full_name if link.person is not None else "Unknown person"
with ui.row().classes("w-full justify-between items-center ui-row-surface p-2"):
ui.label(person_label).classes("text-xs font-semibold ui-text-primary")
archival_badge(link.role.value)
def render_doc_job_details(document: Document) -> None:
with archival_card(title="Pipeline Jobs"):
with ui.row().classes("w-full justify-between items-center mb-2"):
ui.label(f"{len(document.jobs)} Active Jobs").classes("text-xs ui-link-primary font-bold")
with ui.row().classes("w-full gap-2 mt-2"):
ui.button(
"View Jobs",
on_click=lambda: ui.navigate.to(f"/documents/{document.id}/jobs"),
icon="work_history",
).props("flat dense text-xs").classes("ui-link-primary")
ui.button(
"+ Add Job",
on_click=lambda: ui.navigate.to(f"/jobs/new?document_id={document.id}"),
icon="add",
).classes("ui-btn-primary text-xs")
+17
View File
@@ -0,0 +1,17 @@
from typing import Annotated
from fastapi import Depends
from transcription.db.session import SessionFactory
from transcription.db.session import resolve_session_factory
from ..services import ServiceBundle
type SessionFactoryDep = Annotated[SessionFactory, Depends(resolve_session_factory)]
def _get_service_bundle(session_factory: SessionFactoryDep) -> ServiceBundle:
return ServiceBundle.from_session_factory(session_factory)
type ServicesDep = Annotated[ServiceBundle, Depends(_get_service_bundle)]
@@ -0,0 +1,37 @@
from __future__ import annotations
from fastapi import Request
from nicegui import ui
from ...dependency import ServicesDep
from .delete_document import render_delete_document_page
from .document_detail import render_document_detail_page
from .document_overview import render_document_overview_page
from .edit_document import render_document_edit_page
from .new_document import render_new_document_page
__all__ = ["register_pages"]
def register_pages() -> None:
"""Register documents list and detail routes."""
@ui.page("/documents")
async def documents_page(services: ServicesDep) -> None:
await render_document_overview_page(services=services)
@ui.page("/documents/new")
async def document_create_page(request: Request, services: ServicesDep) -> None:
await render_new_document_page(request, services=services)
@ui.page("/documents/{document_id}")
async def document_detail_page(document_id: str, services: ServicesDep) -> None:
await render_document_detail_page(document_id, services=services)
@ui.page("/documents/{document_id}/edit")
async def document_edit_page(document_id: str, services: ServicesDep) -> None:
await render_document_edit_page(document_id, services=services)
@ui.page("/documents/{document_id}/delete")
async def document_delete_page(document_id: str, services: ServicesDep) -> None:
await render_delete_document_page(document_id, services=services)
@@ -0,0 +1,101 @@
from __future__ import annotations
from uuid import UUID
from nicegui import ui
from transcription.errors import ErrorCategory
from transcription.services.documents import DocumentDeleteBlockedError
from transcription.services.documents import DocumentError
from transcription.ui.components.app_shell import render_navigation_header
from transcription.ui.components.cards import archival_card
from transcription.ui.components.error_presenter import show_error
from transcription.ui.components.primitives import destructive_button
from transcription.ui.theme import page_header
from ...dependency import ServicesDep
async def render_delete_document_page(document_id: str, services: ServicesDep) -> None:
render_navigation_header(current_path="/documents")
try:
parsed_document_id = UUID(document_id)
except ValueError:
ui.label("Invalid document id").classes("text-h6 text-red-800 p-4")
return
try:
document = await services.documents.read_document_detail(document_id=parsed_document_id)
except DocumentError:
ui.label("Document not found").classes("text-h6 text-red-800 p-4")
return
except Exception as exc: # noqa: BLE001
show_error(exc, title="Load failed", operation="documents.delete.read")
return
with ui.column().classes("w-full max-w-xl mx-auto p-4 gap-4"):
page_header("Delete Document")
with archival_card(extra_classes="gap-2"):
ui.label(f"Document: {document.name}").classes("text-sm font-semibold ui-text-primary")
has_sources = bool(document.sources)
has_jobs = bool(document.jobs)
if has_sources or has_jobs:
ui.label("Delete is blocked because related records exist.").classes(
"text-xs text-red-800 font-bold mt-2"
)
categories: list[str] = []
if has_sources:
categories.append("Sources")
if has_jobs:
categories.append("Jobs")
ui.label(f"Dependencies present: {', '.join(categories)}").classes("text-xs ui-text-muted")
ui.label("Remove related records first, then retry deletion.").classes("text-xs ui-text-muted italic")
with ui.row().classes("w-full items-center gap-2 mt-4"):
ui.button(
"Back to Document",
on_click=lambda: ui.navigate.to(f"/documents/{document.id}"),
icon="arrow_back",
).classes("ui-btn-primary text-xs")
ui.button("Go to Jobs", on_click=lambda: ui.navigate.to("/jobs"), icon="work_history").props(
"flat text-xs"
)
return
ui.label("This action permanently deletes the document.").classes("text-xs text-red-800 font-medium")
async def submit_delete() -> None:
try:
await services.documents.delete_document(document)
except DocumentDeleteBlockedError as exc:
ui.notify(exc.message, type="warning")
ui.navigate.to(f"/documents/{document.id}/delete")
return
except DocumentError as exc:
if exc.category == ErrorCategory.NOT_FOUND:
ui.notify("Document not found.", type="warning")
ui.navigate.to("/documents")
return
show_error(exc, title="Delete failed", operation="documents.delete")
return
except Exception as exc: # noqa: BLE001
show_error(exc, title="Delete failed", operation="documents.delete")
return
ui.notify("Document deleted", type="positive")
ui.navigate.to("/documents")
with ui.row().classes("w-full items-center gap-2 mt-2"):
destructive_button(
"Delete document permanently",
on_click=submit_delete,
icon="delete_forever",
variant="solid",
)
ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/documents/{document.id}"), icon="arrow_back").props(
"flat"
)
@@ -0,0 +1,90 @@
"""Documents list and detail page registration."""
from __future__ import annotations
from uuid import UUID
from nicegui import ui
from transcription.db.models import DocumentPersonRole
from transcription.services.documents import DocumentError
from transcription.ui.components.app_shell import render_navigation_header
from transcription.ui.components.error_presenter import show_error
from transcription.ui.components.primitives import destructive_button
from transcription.ui.components.primitives import section_header_row
from transcription.ui.components.viewers import dark_room_viewer
from transcription.ui.dependency import ServicesDep
from transcription.ui.theme import page_header
from ...components import document_details as details
async def render_document_detail_page(document_id: str, services: ServicesDep) -> None:
render_navigation_header(current_path="/documents")
try:
parsed_document_id = UUID(document_id)
except ValueError:
ui.label("Invalid document id").classes("text-h6 text-red-800 p-4")
return
try:
document = await services.documents.read_document_detail(document_id=parsed_document_id)
except DocumentError:
ui.label("Document not found").classes("text-h6 text-red-800 p-4")
return
except Exception as exc: # noqa: BLE001
show_error(exc, title="Load failed", operation="documents.read")
return
author_link = next(
(
item
for item in document.document_people
if item.role == DocumentPersonRole.AUTHOR and item.person is not None
),
None,
)
# Main Bento Grid Wrapper
with ui.column().classes("w-full max-w-[1800px] mx-auto p-4 gap-4"):
# Header Bar
with section_header_row():
page_header(document.name, subtitle=f"Type: {document.document_type or 'Unspecified'} | ID: {document.id}")
with ui.row().classes("items-center gap-2"):
ui.button(
"Edit Document",
on_click=lambda: ui.navigate.to(f"/documents/{document.id}/edit"),
icon="edit",
).classes("ui-btn-primary text-xs")
destructive_button(
"Delete",
on_click=lambda: ui.navigate.to(f"/documents/{document.id}/delete"),
icon="delete",
extra_classes="text-xs",
)
# High-Density Bento Grid Layout
with ui.grid().classes("w-full grid-cols-12 gap-4"):
# ZONE 1: Source Image Viewer (Cols 1-5)
with ui.column().classes("col-span-12 lg:col-span-5"):
source_path = document.sources[0].file_path if document.sources else None
dark_room_viewer(source_path, count_label=f"{len(document.sources)} Source(s) Linked")
with ui.row().classes("w-full justify-between items-center mt-2"):
ui.button(
"View All Sources",
on_click=lambda: ui.navigate.to(f"/sources?document_id={document.id}"),
icon="description",
).props("flat dense text-xs").classes("ui-link-primary")
ui.button(
"+ Add Source",
on_click=lambda: ui.navigate.to(f"/jobs/new?document_id={document.id}"),
icon="add",
).classes("ui-btn-primary text-xs")
details.render_archival_metadata(document=document, author_link=author_link)
with ui.column().classes("col-span-12 lg:col-span-3 gap-4"):
details.render_doc_people_details(document=document)
details.render_doc_job_details(document=document)
@@ -0,0 +1,48 @@
from __future__ import annotations
from nicegui import ui
from transcription.ui.components.app_shell import render_navigation_header
from transcription.ui.components.error_presenter import show_error
from transcription.ui.components.primitives import section_header_row
from transcription.ui.components.table.documents import DocumentTableRow
from transcription.ui.components.table.documents import render_documents_table
from transcription.ui.theme import page_header
from ...dependency import ServicesDep
async def render_document_overview_page(services: ServicesDep) -> None:
render_navigation_header(current_path="/documents")
with ui.column().classes("w-full max-w-7xl mx-auto p-4 gap-4"):
with section_header_row():
page_header("Archival Documents")
ui.button(
"Create new document",
on_click=lambda: ui.navigate.to("/documents/new"),
icon="note_add",
).classes("ui-btn-primary")
try:
documents = sorted(
await services.documents.list_documents(),
key=lambda item: item.created_at,
reverse=True,
)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Load failed", operation="documents.list")
return
# Format documents into read-model rows for the table renderer
rows = [
DocumentTableRow(
id=doc.id,
name=doc.name,
document_type=doc.document_type or "",
archive_identifier=doc.archive_identifier or "",
created_at=doc.created_at.strftime("%b %d, %Y"),
)
for doc in documents
]
render_documents_table(rows)
@@ -0,0 +1,175 @@
from datetime import date
from uuid import UUID
from nicegui import ui
from transcription.db.models import Document
from transcription.db.models import DocumentPerson
from transcription.db.models import DocumentPersonRole
from transcription.services.documents import DocumentError
from ...components.app_shell import render_navigation_header
from ...components.cards import archival_card
from ...components.error_presenter import show_error
from ...dependency import ServicesDep
from ...theme import page_header
CREATE_NEW_PERSON_OPTION = "__create_new_person__"
async def render_document_edit_page(document_id: str, services: ServicesDep) -> None:
render_navigation_header(current_path="/documents")
try:
parsed_document_id = UUID(document_id)
except ValueError:
ui.label("Invalid document id").classes("text-h6 text-red-800 p-4")
return
try:
document = await services.documents.read_document_detail(document_id=parsed_document_id)
except DocumentError:
ui.label("Document not found").classes("text-h6 text-red-800 p-4")
return
except Exception as exc: # noqa: BLE001
show_error(exc, title="Load failed", operation="documents.edit.read")
return
with ui.column().classes("w-full max-w-4xl mx-auto p-4 gap-4"):
page_header("Edit Document Record", subtitle="Document name and document type are required.")
with archival_card(extra_classes="gap-3"):
name_input = (
ui.input(label="Document name", value=document.name).props("outlined bg-white").classes("w-full")
)
document_type_input = (
ui.input(label="Document type", value=document.document_type or "")
.props("outlined bg-white")
.classes("w-full")
)
with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-2"):
date_input = ui.input(
label="Exact date (YYYY-MM-DD)",
value=document.document_date.isoformat() if document.document_date else "",
).props('outlined bg-white type="date"')
date_raw_input = ui.input(label="Approximate date", value=document.document_date_raw or "").props(
"outlined bg-white"
)
location_input = (
ui.input(label="Document location", value=document.location_created or "")
.props("outlined bg-white")
.classes("w-full")
)
archive_input = (
ui.input(label="Archive identifier", value=document.archive_identifier or "")
.props("outlined bg-white")
.classes("w-full")
)
notes_input = (
ui.textarea(label="Notes", value=document.notes or "")
.props("outlined bg-white autogrow")
.classes("w-full")
)
people = sorted(await services.documents.list_people(), key=lambda item: item.full_name.casefold())
author_options = {"": "No author", CREATE_NEW_PERSON_OPTION: "Create new item"} | {
str(person.id): person.full_name for person in people
}
existing_author = next(
(item for item in document.document_people if item.role == DocumentPersonRole.AUTHOR),
None,
)
author_value = str(existing_author.person_id) if existing_author is not None else ""
def on_author_change(event) -> None:
selected = str(event.value or "").strip()
if selected == CREATE_NEW_PERSON_OPTION:
ui.navigate.to("/people/new")
author_select = (
ui.select(
author_options,
label="Author (Person)",
value=author_value,
on_change=on_author_change,
)
.props("outlined bg-white")
.classes("w-full")
)
ui.link("Create new person", "/people/new").classes("text-xs ui-link-primary font-medium")
async def submit_edit() -> None:
candidate_name = (name_input.value or "").strip()
candidate_type = (document_type_input.value or "").strip()
if not candidate_name:
ui.notify("Document name is required.", type="warning")
return
if not candidate_type:
ui.notify("Document type is required.", type="warning")
return
parsed_date: date | None = None
candidate_date_text = (date_input.value or "").strip()
if candidate_date_text:
try:
parsed_date = date.fromisoformat(candidate_date_text)
except ValueError:
ui.notify("Exact date must use YYYY-MM-DD.", type="warning")
return
candidate = Document(
id=document.id,
name=candidate_name,
document_type=candidate_type,
document_date=parsed_date,
document_date_raw=(date_raw_input.value or "").strip() or None,
location_created=(location_input.value or "").strip() or None,
notes=(notes_input.value or "").strip() or None,
archive_identifier=(archive_input.value or "").strip() or None,
created_at=document.created_at,
updated_at=document.updated_at,
)
try:
await services.documents.update_document(candidate)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Save failed", operation="documents.edit.save")
return
selected_author = (author_select.value or "").strip()
if selected_author == CREATE_NEW_PERSON_OPTION:
ui.navigate.to("/people/new")
return
existing_author_links = [
link for link in document.document_people if link.role == DocumentPersonRole.AUTHOR
]
try:
if not selected_author:
for link in existing_author_links:
await services.documents.delete_document_person(link)
else:
selected_author_id = UUID(selected_author)
if not any(link.person_id == selected_author_id for link in existing_author_links):
for link in existing_author_links:
await services.documents.delete_document_person(link)
await services.documents.create_document_person(
DocumentPerson(
document_id=document.id,
person_id=selected_author_id,
role=DocumentPersonRole.AUTHOR,
)
)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Author update failed", operation="documents.edit.link_author")
return
ui.notify("Document updated", type="positive")
ui.navigate.to(f"/documents/{document.id}")
with ui.row().classes("w-full items-center gap-2 mt-2"):
ui.button("Save changes", on_click=submit_edit, icon="save").classes("ui-btn-primary")
ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/documents/{document.id}"), icon="arrow_back").props(
"flat"
)
@@ -0,0 +1,121 @@
from __future__ import annotations
from datetime import date
from uuid import UUID
from fastapi import Request
from nicegui import ui
from transcription.db.models import Document
from transcription.db.models import DocumentPerson
from transcription.db.models import DocumentPersonRole
from transcription.ui.components.app_shell import render_navigation_header
from transcription.ui.components.cards import archival_card
from transcription.ui.components.error_presenter import show_error
from transcription.ui.theme import page_header
from ...dependency import ServicesDep
CREATE_NEW_PERSON_OPTION = "__create_new_person__"
async def render_new_document_page(request: Request, services: ServicesDep) -> None:
render_navigation_header(current_path="/documents")
with ui.column().classes("w-full max-w-4xl mx-auto p-4 gap-4"):
page_header("Create Document", subtitle="Document name is required.")
with archival_card(extra_classes="gap-3"):
name_input = ui.input(label="Document name").props("outlined bg-white").classes("w-full")
document_type_input = ui.input(label="Document type").props("outlined bg-white").classes("w-full")
with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-2"):
date_input = ui.input(label="Exact date (YYYY-MM-DD)").props('outlined bg-white type="date"')
date_raw_input = ui.input(label="Approximate date").props("outlined bg-white")
location_input = ui.input(label="Document location").props("outlined bg-white").classes("w-full")
archive_input = ui.input(label="Archive identifier").props("outlined bg-white").classes("w-full")
notes_input = ui.textarea(label="Notes").props("outlined bg-white autogrow").classes("w-full")
people = sorted(await services.documents.list_people(), key=lambda item: item.full_name.casefold())
author_options = {"": "No author", CREATE_NEW_PERSON_OPTION: "Create new item"} | {
str(person.id): person.full_name for person in people
}
def on_author_change(event) -> None:
selected = str(event.value or "").strip()
if selected == CREATE_NEW_PERSON_OPTION:
ui.navigate.to("/people/new")
author_select = (
ui.select(author_options, label="Author (Person)", value="", on_change=on_author_change)
.props("outlined bg-white")
.classes("w-full")
)
ui.link("Create new person", "/people/new").classes("text-xs ui-link-primary font-medium")
return_to = request.query_params.get("return_to")
async def submit_create() -> None:
candidate_name = (name_input.value or "").strip()
if not candidate_name:
ui.notify("Document name is required.", type="warning")
return
parsed_date: date | None = None
candidate_date_text = (date_input.value or "").strip()
if candidate_date_text:
try:
parsed_date = date.fromisoformat(candidate_date_text)
except ValueError:
ui.notify("Exact date must use YYYY-MM-DD.", type="warning")
return
candidate = Document(
name=candidate_name,
document_type=(document_type_input.value or "").strip() or None,
document_date=parsed_date,
document_date_raw=(date_raw_input.value or "").strip() or None,
location_created=(location_input.value or "").strip() or None,
notes=(notes_input.value or "").strip() or None,
archive_identifier=(archive_input.value or "").strip() or None,
)
try:
created = await services.documents.create_document(candidate)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Create failed", operation="documents.create")
return
selected_author = (author_select.value or "").strip()
if selected_author == CREATE_NEW_PERSON_OPTION:
ui.navigate.to("/people/new")
return
if selected_author:
try:
parsed_person_id = UUID(selected_author)
except ValueError:
ui.notify("Selected author is invalid.", type="warning")
return
try:
await services.documents.create_document_person(
DocumentPerson(
document_id=created.id,
person_id=parsed_person_id,
role=DocumentPersonRole.AUTHOR,
)
)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Author link failed", operation="documents.create.link_author")
return
ui.notify("Document created", type="positive")
if return_to == "jobs_new":
ui.navigate.to(f"/jobs/new?document_id={created.id}")
return
ui.navigate.to(f"/documents/{created.id}")
with ui.row().classes("w-full items-center gap-2 mt-2"):
ui.button("Save document", on_click=submit_create, icon="save").classes("ui-btn-primary")
ui.button("Cancel", on_click=lambda: ui.navigate.to("/documents"), icon="arrow_back").props("flat")
@@ -1,600 +0,0 @@
"""Documents list and detail page registration."""
from __future__ import annotations
from datetime import date
from uuid import UUID
from fastapi import Request
from fastapi.responses import RedirectResponse
from nicegui import ui
from transcription.db.models import Document
from transcription.db.models import DocumentPerson
from transcription.db.models import DocumentPersonRole
from transcription.errors import ErrorCategory
from transcription.services.documents import DocumentDeleteBlockedError
from transcription.services.documents import DocumentError
from transcription.services.documents import DocumentService
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 archival_badge
from transcription.ui.components.data_display import metadata_row
from transcription.ui.components.error_presenter import show_error
from transcription.ui.components.primitives import destructive_button
from transcription.ui.components.primitives import render_empty_state
from transcription.ui.components.primitives import section_header_row
from transcription.ui.components.table.documents import DocumentTableRow
from transcription.ui.components.table.documents import render_documents_table
from transcription.ui.components.viewers import dark_room_viewer
from transcription.ui.theme import apply_archival_theme
from transcription.ui.theme import page_header
from ...db.session import SessionFactoryDep
CREATE_NEW_PERSON_OPTION = "__create_new_person__"
def register_page() -> None:
"""Register documents list and detail routes."""
@ui.page("/documents/new")
async def document_create_page(request: Request, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
document_service = DocumentService(session_factory=session_factory)
render_navigation_header(current_path="/documents")
with ui.column().classes("w-full max-w-4xl mx-auto p-4 gap-4"):
page_header("Create Document", subtitle="Document name is required.")
with archival_card(extra_classes="gap-3"):
name_input = ui.input(label="Document name").props("outlined bg-white").classes("w-full")
document_type_input = ui.input(label="Document type").props("outlined bg-white").classes("w-full")
with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-2"):
date_input = ui.input(label="Exact date (YYYY-MM-DD)").props('outlined bg-white type="date"')
date_raw_input = ui.input(label="Approximate date").props("outlined bg-white")
location_input = ui.input(label="Document location").props("outlined bg-white").classes("w-full")
archive_input = ui.input(label="Archive identifier").props("outlined bg-white").classes("w-full")
notes_input = ui.textarea(label="Notes").props("outlined bg-white autogrow").classes("w-full")
people = sorted(await document_service.list_people(), key=lambda item: item.full_name.casefold())
author_options = (
{"": "No author", CREATE_NEW_PERSON_OPTION: "Create new item"}
| {str(person.id): person.full_name for person in people}
)
def on_author_change(event) -> None:
selected = str(event.value or "").strip()
if selected == CREATE_NEW_PERSON_OPTION:
ui.navigate.to("/people/new")
author_select = (
ui.select(author_options, label="Author (Person)", value="", on_change=on_author_change)
.props("outlined bg-white")
.classes("w-full")
)
ui.link("Create new person", "/people/new").classes("text-xs ui-link-primary font-medium")
return_to = request.query_params.get("return_to")
async def submit_create() -> None:
candidate_name = (name_input.value or "").strip()
if not candidate_name:
ui.notify("Document name is required.", type="warning")
return
parsed_date: date | None = None
candidate_date_text = (date_input.value or "").strip()
if candidate_date_text:
try:
parsed_date = date.fromisoformat(candidate_date_text)
except ValueError:
ui.notify("Exact date must use YYYY-MM-DD.", type="warning")
return
candidate = Document(
name=candidate_name,
document_type=(document_type_input.value or "").strip() or None,
document_date=parsed_date,
document_date_raw=(date_raw_input.value or "").strip() or None,
location_created=(location_input.value or "").strip() or None,
notes=(notes_input.value or "").strip() or None,
archive_identifier=(archive_input.value or "").strip() or None,
)
try:
created = await document_service.create_document(candidate)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Create failed", operation="documents.create")
return
selected_author = (author_select.value or "").strip()
if selected_author == CREATE_NEW_PERSON_OPTION:
ui.navigate.to("/people/new")
return
if selected_author:
try:
parsed_person_id = UUID(selected_author)
except ValueError:
ui.notify("Selected author is invalid.", type="warning")
return
try:
await document_service.create_document_person(
DocumentPerson(
document_id=created.id,
person_id=parsed_person_id,
role=DocumentPersonRole.AUTHOR,
)
)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Author link failed", operation="documents.create.link_author")
return
ui.notify("Document created", type="positive")
if return_to == "jobs_new":
ui.navigate.to(f"/jobs/new?document_id={created.id}")
return
ui.navigate.to(f"/documents/{created.id}")
with ui.row().classes("w-full items-center gap-2 mt-2"):
ui.button("Save document", on_click=submit_create, icon="save").classes("ui-btn-primary")
ui.button("Cancel", on_click=lambda: ui.navigate.to("/documents"), icon="arrow_back").props("flat")
@ui.page("/documents")
async def documents_page(session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
document_service = DocumentService(session_factory=session_factory)
render_navigation_header(current_path="/documents")
with ui.column().classes("w-full max-w-7xl mx-auto p-4 gap-4"):
with section_header_row():
page_header("Archival Documents")
ui.button(
"Create new document",
on_click=lambda: ui.navigate.to("/documents/new"),
icon="note_add",
).classes("ui-btn-primary")
try:
documents = sorted(
await document_service.list_documents(),
key=lambda item: item.created_at,
reverse=True,
)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Load failed", operation="documents.list")
return
# Format documents into read-model rows for the table renderer
rows = [
DocumentTableRow(
id=doc.id,
name=doc.name,
document_type=doc.document_type or "",
archive_identifier=doc.archive_identifier or "",
created_at=doc.created_at.strftime("%b %d, %Y"),
)
for doc in documents
]
render_documents_table(rows)
@ui.page("/documents/{document_id}")
async def document_detail_page(document_id: str, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
document_service = DocumentService(session_factory=session_factory)
render_navigation_header(current_path="/documents")
try:
parsed_document_id = UUID(document_id)
except ValueError:
ui.label("Invalid document id").classes("text-h6 text-red-800 p-4")
return
try:
document = await document_service.read_document_detail(document_id=parsed_document_id)
except DocumentError:
ui.label("Document not found").classes("text-h6 text-red-800 p-4")
return
except Exception as exc: # noqa: BLE001
show_error(exc, title="Load failed", operation="documents.read")
return
author_link = next(
(
item
for item in document.document_people
if item.role == DocumentPersonRole.AUTHOR and item.person is not None
),
None,
)
# Main Bento Grid Wrapper
with ui.column().classes("w-full max-w-[1800px] mx-auto p-4 gap-4"):
# Header Bar
with section_header_row():
page_header(document.name, subtitle=f"Type: {document.document_type or 'Unspecified'} | ID: {document.id}")
with ui.row().classes("items-center gap-2"):
ui.button(
"Edit Document",
on_click=lambda: ui.navigate.to(f"/documents/{document.id}/edit"),
icon="edit",
).classes("ui-btn-primary text-xs")
destructive_button(
"Delete",
on_click=lambda: ui.navigate.to(f"/documents/{document.id}/delete"),
icon="delete",
extra_classes="text-xs",
)
# High-Density Bento Grid Layout
with ui.grid().classes("w-full grid-cols-12 gap-4"):
# ZONE 1: Source Image Viewer (Cols 1-5)
with ui.column().classes("col-span-12 lg:col-span-5"):
source_path = document.sources[0].file_path if document.sources else None
dark_room_viewer(source_path, count_label=f"{len(document.sources)} Source(s) Linked")
with ui.row().classes("w-full justify-between items-center mt-2"):
ui.button(
"View All Sources",
on_click=lambda: ui.navigate.to(f"/sources?document_id={document.id}"),
icon="description",
).props("flat dense text-xs").classes("ui-link-primary")
ui.button(
"+ Add Source",
on_click=lambda: ui.navigate.to(f"/jobs/new?document_id={document.id}"),
icon="add",
).classes("ui-btn-primary text-xs")
# ZONE 2: Metadata & Archival Attributes (Cols 6-8)
with ui.column().classes("col-span-12 lg:col-span-4 gap-4"):
with archival_card(title="Archival Metadata"):
metadata_row("Author:", author_link.person.full_name if author_link and author_link.person else "Not set")
metadata_row("Exact Date:", document.document_date.isoformat() if document.document_date else "Not set")
metadata_row("Approx. Date:", document.document_date_raw or "Not set")
metadata_row("Location Created:", document.location_created or "Not set")
metadata_row("Archive Identifier:", document.archive_identifier or "Not set")
with ui.column().classes("w-full mt-2"):
ui.label("Archival Notes:").classes("ui-text-muted text-xs mb-1")
ui.label(document.notes or "No notes added.").classes(
"p-2 ui-note-box text-xs"
)
with archival_card(title="System Logistics"):
ui.label(f"Created: {document.created_at.isoformat()}").classes("text-[11px] ui-text-muted")
ui.label(f"Updated: {document.updated_at.isoformat()}").classes("text-[11px] ui-text-muted")
# ZONE 3: Related Entities & Pipeline Jobs (Cols 9-12)
with ui.column().classes("col-span-12 lg:col-span-3 gap-4"):
with archival_card(title="Related People"):
if not document.document_people:
render_empty_state("No linked people yet.", italic=True)
else:
with ui.column().classes("w-full gap-2"):
for link in document.document_people:
person_label = link.person.full_name if link.person is not None else "Unknown person"
with ui.row().classes(
"w-full justify-between items-center ui-row-surface p-2"
):
ui.label(person_label).classes("text-xs font-semibold ui-text-primary")
archival_badge(link.role.value)
with archival_card(title="Pipeline Jobs"):
with ui.row().classes("w-full justify-between items-center mb-2"):
ui.label(f"{len(document.jobs)} Active Jobs").classes("text-xs ui-link-primary font-bold")
with ui.row().classes("w-full gap-2 mt-2"):
ui.button(
"View Jobs",
on_click=lambda: ui.navigate.to(f"/documents/{document.id}/jobs"),
icon="work_history",
).props("flat dense text-xs").classes("ui-link-primary")
ui.button(
"+ Add Job",
on_click=lambda: ui.navigate.to(f"/jobs/new?document_id={document.id}"),
icon="add",
).classes("ui-btn-primary text-xs")
@ui.page("/documents/{document_id}/jobs")
async def document_jobs_page(document_id: str, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
document_service = DocumentService(session_factory=session_factory)
render_navigation_header(current_path="/documents")
try:
parsed_document_id = UUID(document_id)
except ValueError:
ui.label("Invalid document id").classes("text-h6 text-red-800 p-4")
return
try:
document = await document_service.read_document_detail(document_id=parsed_document_id)
except DocumentError:
ui.label("Document not found").classes("text-h6 text-red-800 p-4")
return
except Exception as exc: # noqa: BLE001
show_error(exc, title="Load failed", operation="documents.jobs")
return
with ui.column().classes("w-full max-w-4xl mx-auto p-4 gap-4"):
with section_header_row():
page_header(f"Jobs for {document.name}")
with ui.row().classes("gap-2"):
ui.button(
"Back to Document",
on_click=lambda: ui.navigate.to(f"/documents/{document.id}"),
icon="arrow_back",
).props("flat")
ui.button(
"Create Job",
on_click=lambda: ui.navigate.to(f"/jobs/new?document_id={document.id}"),
icon="add",
).classes("ui-btn-primary")
if not document.jobs:
with archival_card(extra_classes="p-6 text-center"):
render_empty_state("No transcription processing jobs created yet.")
return
for job in sorted(document.jobs, key=lambda item: item.date_created, reverse=True):
with archival_card(extra_classes="p-3"):
with ui.row().classes("w-full items-center justify-between"):
with ui.row().classes("items-center gap-2"):
archival_badge(job.status.value)
ui.label(f"Job ID: {job.id}").classes("text-xs font-mono ui-text-primary")
ui.button(
"Open Job",
on_click=lambda _=None, job_id=job.id: ui.navigate.to(f"/jobs/{job_id}"),
icon="open_in_new",
).props("flat dense").classes("text-xs ui-link-primary")
@ui.page("/documents/{document_id}/sources")
async def document_sources_page(document_id: str, session_factory: SessionFactoryDep) -> RedirectResponse:
_ = session_factory
return RedirectResponse(url=f"/ui/sources?document_id={document_id}")
@ui.page("/documents/{document_id}/edit")
async def document_edit_page(document_id: str, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
document_service = DocumentService(session_factory=session_factory)
render_navigation_header(current_path="/documents")
try:
parsed_document_id = UUID(document_id)
except ValueError:
ui.label("Invalid document id").classes("text-h6 text-red-800 p-4")
return
try:
document = await document_service.read_document_detail(document_id=parsed_document_id)
except DocumentError:
ui.label("Document not found").classes("text-h6 text-red-800 p-4")
return
except Exception as exc: # noqa: BLE001
show_error(exc, title="Load failed", operation="documents.edit.read")
return
with ui.column().classes("w-full max-w-4xl mx-auto p-4 gap-4"):
page_header("Edit Document Record", subtitle="Document name and document type are required.")
with archival_card(extra_classes="gap-3"):
name_input = ui.input(label="Document name", value=document.name).props("outlined bg-white").classes("w-full")
document_type_input = (
ui.input(label="Document type", value=document.document_type or "")
.props("outlined bg-white")
.classes("w-full")
)
with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-2"):
date_input = ui.input(
label="Exact date (YYYY-MM-DD)",
value=document.document_date.isoformat() if document.document_date else "",
).props('outlined bg-white type="date"')
date_raw_input = (
ui.input(label="Approximate date", value=document.document_date_raw or "")
.props("outlined bg-white")
)
location_input = (
ui.input(label="Document location", value=document.location_created or "")
.props("outlined bg-white")
.classes("w-full")
)
archive_input = (
ui.input(label="Archive identifier", value=document.archive_identifier or "")
.props("outlined bg-white")
.classes("w-full")
)
notes_input = (
ui.textarea(label="Notes", value=document.notes or "").props("outlined bg-white autogrow").classes("w-full")
)
people = sorted(await document_service.list_people(), key=lambda item: item.full_name.casefold())
author_options = (
{"": "No author", CREATE_NEW_PERSON_OPTION: "Create new item"}
| {str(person.id): person.full_name for person in people}
)
existing_author = next(
(item for item in document.document_people if item.role == DocumentPersonRole.AUTHOR),
None,
)
author_value = str(existing_author.person_id) if existing_author is not None else ""
def on_author_change(event) -> None:
selected = str(event.value or "").strip()
if selected == CREATE_NEW_PERSON_OPTION:
ui.navigate.to("/people/new")
author_select = (
ui.select(
author_options,
label="Author (Person)",
value=author_value,
on_change=on_author_change,
)
.props("outlined bg-white")
.classes("w-full")
)
ui.link("Create new person", "/people/new").classes("text-xs ui-link-primary font-medium")
async def submit_edit() -> None:
candidate_name = (name_input.value or "").strip()
candidate_type = (document_type_input.value or "").strip()
if not candidate_name:
ui.notify("Document name is required.", type="warning")
return
if not candidate_type:
ui.notify("Document type is required.", type="warning")
return
parsed_date: date | None = None
candidate_date_text = (date_input.value or "").strip()
if candidate_date_text:
try:
parsed_date = date.fromisoformat(candidate_date_text)
except ValueError:
ui.notify("Exact date must use YYYY-MM-DD.", type="warning")
return
candidate = Document(
id=document.id,
name=candidate_name,
document_type=candidate_type,
document_date=parsed_date,
document_date_raw=(date_raw_input.value or "").strip() or None,
location_created=(location_input.value or "").strip() or None,
notes=(notes_input.value or "").strip() or None,
archive_identifier=(archive_input.value or "").strip() or None,
created_at=document.created_at,
updated_at=document.updated_at,
)
try:
await document_service.update_document(candidate)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Save failed", operation="documents.edit.save")
return
selected_author = (author_select.value or "").strip()
if selected_author == CREATE_NEW_PERSON_OPTION:
ui.navigate.to("/people/new")
return
existing_author_links = [
link for link in document.document_people if link.role == DocumentPersonRole.AUTHOR
]
try:
if not selected_author:
for link in existing_author_links:
await document_service.delete_document_person(link)
else:
selected_author_id = UUID(selected_author)
if not any(link.person_id == selected_author_id for link in existing_author_links):
for link in existing_author_links:
await document_service.delete_document_person(link)
await document_service.create_document_person(
DocumentPerson(
document_id=document.id,
person_id=selected_author_id,
role=DocumentPersonRole.AUTHOR,
)
)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Author update failed", operation="documents.edit.link_author")
return
ui.notify("Document updated", type="positive")
ui.navigate.to(f"/documents/{document.id}")
with ui.row().classes("w-full items-center gap-2 mt-2"):
ui.button("Save changes", on_click=submit_edit, icon="save").classes("ui-btn-primary")
ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/documents/{document.id}"), icon="arrow_back").props(
"flat"
)
@ui.page("/documents/{document_id}/delete")
async def document_delete_page(document_id: str, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
document_service = DocumentService(session_factory=session_factory)
render_navigation_header(current_path="/documents")
try:
parsed_document_id = UUID(document_id)
except ValueError:
ui.label("Invalid document id").classes("text-h6 text-red-800 p-4")
return
try:
document = await document_service.read_document_detail(document_id=parsed_document_id)
except DocumentError:
ui.label("Document not found").classes("text-h6 text-red-800 p-4")
return
except Exception as exc: # noqa: BLE001
show_error(exc, title="Load failed", operation="documents.delete.read")
return
with ui.column().classes("w-full max-w-xl mx-auto p-4 gap-4"):
page_header("Delete Document")
with archival_card(extra_classes="gap-2"):
ui.label(f"Document: {document.name}").classes("text-sm font-semibold ui-text-primary")
has_sources = bool(document.sources)
has_jobs = bool(document.jobs)
if has_sources or has_jobs:
ui.label("Delete is blocked because related records exist.").classes("text-xs text-red-800 font-bold mt-2")
categories: list[str] = []
if has_sources:
categories.append("Sources")
if has_jobs:
categories.append("Jobs")
ui.label(f"Dependencies present: {', '.join(categories)}").classes("text-xs ui-text-muted")
ui.label("Remove related records first, then retry deletion.").classes("text-xs ui-text-muted italic")
with ui.row().classes("w-full items-center gap-2 mt-4"):
ui.button(
"Back to Document",
on_click=lambda: ui.navigate.to(f"/documents/{document.id}"),
icon="arrow_back",
).classes("ui-btn-primary text-xs")
ui.button("Go to Jobs", on_click=lambda: ui.navigate.to("/jobs"), icon="work_history").props(
"flat text-xs"
)
return
ui.label("This action permanently deletes the document.").classes("text-xs text-red-800 font-medium")
async def submit_delete() -> None:
try:
await document_service.delete_document(document)
except DocumentDeleteBlockedError as exc:
ui.notify(exc.message, type="warning")
ui.navigate.to(f"/documents/{document.id}/delete")
return
except DocumentError as exc:
if exc.category == ErrorCategory.NOT_FOUND:
ui.notify("Document not found.", type="warning")
ui.navigate.to("/documents")
return
show_error(exc, title="Delete failed", operation="documents.delete")
return
except Exception as exc: # noqa: BLE001
show_error(exc, title="Delete failed", operation="documents.delete")
return
ui.notify("Document deleted", type="positive")
ui.navigate.to("/documents")
with ui.row().classes("w-full items-center gap-2 mt-2"):
destructive_button(
"Delete document permanently",
on_click=submit_delete,
icon="delete_forever",
variant="solid",
)
ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/documents/{document.id}"), icon="arrow_back").props(
"flat"
)