generated from john/python-template
939 lines
42 KiB
Python
939 lines
42 KiB
Python
"""Documents list and detail page registration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from fastapi import Request
|
|
from fastapi.responses import RedirectResponse
|
|
from nicegui import ui
|
|
|
|
from transcription.config import Settings
|
|
from transcription.db.models import Document
|
|
from transcription.db.models import Source
|
|
from transcription.db.registries import AUTHOR_ROLE_SEMANTIC_KEY
|
|
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.services.people import PeopleService
|
|
from transcription.services.sources import SourceService
|
|
from transcription.services.workflows import create_document_with_people
|
|
from transcription.services.workflows import update_document_with_people
|
|
from transcription.ui.components.app_shell import render_navigation_header
|
|
from transcription.ui.components.cards import archival_card
|
|
from transcription.ui.components.confirm_delete import dependency_summary
|
|
from transcription.ui.components.confirm_delete import render_delete_actions
|
|
from transcription.ui.components.confirm_delete import render_delete_blocked_notice
|
|
from transcription.ui.components.data_display import archival_badge
|
|
from transcription.ui.components.data_display import metadata_link_row
|
|
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 run_ui_action
|
|
from transcription.ui.components.error_presenter import show_error
|
|
from transcription.ui.components.formatters import compact_date
|
|
from transcription.ui.components.formatters import google_maps_search_url
|
|
from transcription.ui.components.formatters import parse_iso_date
|
|
from transcription.ui.components.formatters import parse_uuid
|
|
from transcription.ui.components.guards import parsed_record_id
|
|
from transcription.ui.components.guards import render_record_not_found
|
|
from transcription.ui.components.linked_people import LinkedPeopleEditor
|
|
from transcription.ui.components.linked_people import StagedLinkedPerson
|
|
from transcription.ui.components.media_urls import resolve_media_url
|
|
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.runtime import resolve_runtime_settings
|
|
from transcription.ui.theme import page_header
|
|
|
|
from ...db.session import SessionFactoryDep
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class DocumentFormFields:
|
|
"""Bound input widgets for the Document create and edit forms."""
|
|
|
|
name: ui.input
|
|
document_type: ui.select
|
|
type_options: dict[str, str]
|
|
document_date: ui.input
|
|
document_date_raw: ui.input
|
|
location: ui.input
|
|
archive: ui.input
|
|
notes: ui.textarea
|
|
tags: ui.select
|
|
|
|
|
|
def register_page() -> None: # noqa: PLR0915
|
|
"""Register documents list and detail routes."""
|
|
|
|
@ui.page("/documents/new")
|
|
async def document_create_page(request: Request, session_factory: SessionFactoryDep) -> None:
|
|
document_service = DocumentService(session_factory=session_factory)
|
|
people_service = PeopleService(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 and type are required.")
|
|
|
|
people = sorted(await people_service.list_people(), key=lambda item: item.full_name.casefold())
|
|
role_catalog = list(await people_service.list_person_roles(active_only=False))
|
|
type_catalog = await document_service.list_document_types()
|
|
tag_catalog = await document_service.list_tags(active_only=True)
|
|
requested_person_id = parse_uuid(request.query_params.get("person_id"))
|
|
staged_links: list[StagedLinkedPerson] = []
|
|
if requested_person_id is not None and any(person.id == requested_person_id for person in people):
|
|
author_role_outcome = await run_ui_action(
|
|
operation="documents.create.preselect",
|
|
title="Author role unavailable",
|
|
action=lambda: people_service.read_person_role_by_semantic_key(AUTHOR_ROLE_SEMANTIC_KEY),
|
|
)
|
|
if author_role_outcome.ok and author_role_outcome.value is not None:
|
|
author_role = author_role_outcome.value
|
|
if author_role.is_active:
|
|
staged_links.append(StagedLinkedPerson(person_id=requested_person_id, role_id=author_role.id))
|
|
else:
|
|
ui.notify(
|
|
"The Author role is inactive, so the Person could not be preselected.",
|
|
type="warning",
|
|
)
|
|
elif request.query_params.get("person_id"):
|
|
ui.notify("The requested person could not be preselected.", type="warning")
|
|
linked_people = LinkedPeopleEditor(
|
|
people=people,
|
|
roles=role_catalog,
|
|
staged_links=staged_links,
|
|
)
|
|
form = _render_document_form_fields(
|
|
type_options={str(doc_type.id): doc_type.label for doc_type in type_catalog},
|
|
tag_options=[tag.label for tag in tag_catalog],
|
|
linked_people=linked_people,
|
|
)
|
|
|
|
return_to = request.query_params.get("return_to")
|
|
|
|
async def submit_create() -> None:
|
|
candidate_name = (form.name.value or "").strip()
|
|
candidate_type_id = _resolve_selected_document_type_id(form.document_type.value, form.type_options)
|
|
if not candidate_name:
|
|
ui.notify("Document name is required.", type="warning")
|
|
return
|
|
if candidate_type_id is None:
|
|
ui.notify("Document type is required.", type="warning")
|
|
return
|
|
|
|
parsed_date = parse_iso_date(form.document_date.value)
|
|
if form.document_date.value and parsed_date is None:
|
|
ui.notify("Exact date must use YYYY-MM-DD.", type="warning")
|
|
return
|
|
|
|
candidate = Document(
|
|
name=candidate_name,
|
|
document_type_id=candidate_type_id,
|
|
document_date=parsed_date,
|
|
document_date_raw=(form.document_date_raw.value or "").strip() or None,
|
|
location_created=(form.location.value or "").strip() or None,
|
|
notes=(form.notes.value or "").strip() or None,
|
|
archive_identifier=(form.archive.value or "").strip() or None,
|
|
)
|
|
|
|
created_outcome = await run_ui_action(
|
|
operation="documents.create",
|
|
title="Create failed",
|
|
action=lambda: create_document_with_people(
|
|
document=candidate,
|
|
links=linked_people.values(),
|
|
tag_labels=_resolve_selected_tag_labels(form.tags.value),
|
|
documents=document_service,
|
|
people=people_service,
|
|
),
|
|
)
|
|
if not created_outcome.ok or created_outcome.value is None:
|
|
return
|
|
created = created_outcome.value
|
|
|
|
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:
|
|
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")
|
|
|
|
documents_outcome = await run_ui_action(
|
|
operation="documents.list",
|
|
title="Load failed",
|
|
action=document_service.list_documents,
|
|
)
|
|
if not documents_outcome.ok:
|
|
return
|
|
documents = sorted(
|
|
documents_outcome.value or (),
|
|
key=lambda item: item.created_at,
|
|
reverse=True,
|
|
)
|
|
|
|
rows = [
|
|
DocumentTableRow(
|
|
id=doc.id,
|
|
name=doc.name,
|
|
authors=", ".join(_author_names(doc)),
|
|
tags=", ".join(_tag_labels(doc)),
|
|
document_date=compact_date(doc.document_date, doc.document_date_raw),
|
|
document_type=(doc.document_type_ref.label if doc.document_type_ref is not None else ""),
|
|
source_count=len(doc.sources),
|
|
transcription_status=_latest_job_status(doc),
|
|
)
|
|
for doc in documents
|
|
]
|
|
render_documents_table(rows)
|
|
|
|
@ui.page("/documents/{document_id}")
|
|
async def document_detail_page(request: Request, document_id: str, session_factory: SessionFactoryDep) -> None:
|
|
document_service = DocumentService(session_factory=session_factory)
|
|
sources_service = SourceService(session_factory=session_factory)
|
|
render_navigation_header(current_path="/documents")
|
|
settings = resolve_runtime_settings(request)
|
|
back_label = "Back to Documents"
|
|
back_target = "/documents"
|
|
from_context = request.query_params.get("from")
|
|
if from_context == "person":
|
|
person_id = parse_uuid(request.query_params.get("person_id"))
|
|
if person_id is not None:
|
|
back_label = "Back to Person"
|
|
back_target = f"/people/{person_id}"
|
|
elif from_context == "job":
|
|
job_id = parse_uuid(request.query_params.get("job_id"))
|
|
if job_id is not None:
|
|
back_label = "Back to Job"
|
|
back_target = f"/jobs/{job_id}"
|
|
|
|
parsed_doc_id = parsed_record_id(document_id, noun="Document")
|
|
if parsed_doc_id is None:
|
|
return
|
|
|
|
try:
|
|
document = await document_service.read_document_detail(document_id=parsed_doc_id)
|
|
except DocumentError:
|
|
render_record_not_found("Document")
|
|
return
|
|
except Exception as exc: # noqa: BLE001
|
|
show_error(exc, title="Load failed", operation="documents.read")
|
|
return
|
|
|
|
active_source = _resolve_active_source(document, parse_uuid(request.query_params.get("source_id")))
|
|
|
|
with ui.column().classes("w-full max-w-[1800px] mx-auto p-4 gap-4"):
|
|
type_display = document.document_type_ref.label if document.document_type_ref is not None else "Unspecified"
|
|
first_source = _resolve_active_source(document, None)
|
|
with section_header_row():
|
|
page_header(document.name, subtitle=f"Type: {type_display} | ID: {document.id}")
|
|
|
|
with ui.row().classes("items-center gap-2"):
|
|
ui.button(back_label, on_click=lambda: ui.navigate.to(back_target), icon="arrow_back").props("flat")
|
|
ui.button(
|
|
"Print",
|
|
on_click=lambda: ui.navigate.to(f"/documents/{document.id}/print"),
|
|
icon="print",
|
|
).props("flat").classes("text-xs")
|
|
ui.button(
|
|
"Edit Document",
|
|
on_click=lambda: ui.navigate.to(f"/documents/{document.id}/edit"),
|
|
icon="edit",
|
|
).classes("ui-btn-primary text-xs")
|
|
ui.button(
|
|
"Document Details",
|
|
on_click=lambda: ui.navigate.to(f"/documents/{document.id}/info"),
|
|
icon="info",
|
|
).props("flat").classes("text-xs")
|
|
ui.button(
|
|
"View Source Detail",
|
|
on_click=(
|
|
(
|
|
lambda: ui.navigate.to(
|
|
f"/sources/{first_source.id}?from=document&document_id={document.id}"
|
|
)
|
|
)
|
|
if first_source is not None
|
|
else (lambda: ui.notify("No source pages are linked yet.", type="warning"))
|
|
),
|
|
icon="description",
|
|
).props("flat").classes("text-xs")
|
|
destructive_button(
|
|
"Delete",
|
|
on_click=lambda: ui.navigate.to(f"/documents/{document.id}/delete"),
|
|
icon="delete",
|
|
extra_classes="text-xs",
|
|
)
|
|
|
|
with ui.grid().classes("w-full grid-cols-12 gap-4"):
|
|
_render_document_detail_viewer_zone(
|
|
document=document,
|
|
active_source=active_source,
|
|
base_url=str(request.base_url),
|
|
settings=settings,
|
|
)
|
|
_render_document_detail_revision_zone(
|
|
source=active_source,
|
|
sources_service=sources_service,
|
|
)
|
|
_render_bento_relations_zone(document)
|
|
|
|
@ui.page("/documents/{document_id}/info")
|
|
async def document_info_page(document_id: str, session_factory: SessionFactoryDep) -> None:
|
|
document_service = DocumentService(session_factory=session_factory)
|
|
render_navigation_header(current_path="/documents")
|
|
|
|
parsed_doc_id = parsed_record_id(document_id, noun="Document")
|
|
if parsed_doc_id is None:
|
|
return
|
|
|
|
try:
|
|
document = await document_service.read_document_detail(document_id=parsed_doc_id)
|
|
except DocumentError:
|
|
render_record_not_found("Document")
|
|
return
|
|
except Exception as exc: # noqa: BLE001
|
|
show_error(exc, title="Load failed", operation="documents.info.read")
|
|
return
|
|
|
|
with ui.column().classes("w-full max-w-6xl mx-auto p-4 gap-4"):
|
|
with section_header_row():
|
|
page_header("Document Info", subtitle=f"{document.name} ({document.id})")
|
|
ui.button(
|
|
"Back to Document",
|
|
on_click=lambda: ui.navigate.to(f"/documents/{document.id}"),
|
|
icon="arrow_back",
|
|
).props("flat")
|
|
_render_bento_metadata_zone(document)
|
|
|
|
@ui.page("/documents/{document_id}/jobs")
|
|
async def document_jobs_page(document_id: str, session_factory: SessionFactoryDep) -> RedirectResponse:
|
|
_ = session_factory
|
|
return RedirectResponse(url=f"/ui/jobs?document_id={document_id}")
|
|
|
|
@ui.page("/documents/{document_id}/sources")
|
|
async def document_sources_page(request: Request, document_id: str, session_factory: SessionFactoryDep) -> None:
|
|
document_service = DocumentService(session_factory=session_factory)
|
|
render_navigation_header(current_path="/documents")
|
|
settings = resolve_runtime_settings(request)
|
|
|
|
parsed_doc_id = parsed_record_id(document_id, noun="Document")
|
|
if parsed_doc_id is None:
|
|
return
|
|
|
|
try:
|
|
document = await document_service.read_document_detail(document_id=parsed_doc_id)
|
|
except DocumentError:
|
|
render_record_not_found("Document")
|
|
return
|
|
except Exception as exc: # noqa: BLE001
|
|
show_error(exc, title="Load failed", operation="documents.sources.read")
|
|
return
|
|
|
|
ordered_sources = _sorted_document_sources(document)
|
|
with ui.column().classes("w-full max-w-7xl mx-auto p-4 gap-4"):
|
|
with section_header_row():
|
|
page_header("Source Images", subtitle=f"{document.name} ({len(ordered_sources)} pages)")
|
|
ui.button(
|
|
"Back to Document",
|
|
on_click=lambda: ui.navigate.to(f"/documents/{document.id}"),
|
|
icon="arrow_back",
|
|
).props("flat")
|
|
|
|
if not ordered_sources:
|
|
render_empty_state("No source pages are linked yet.")
|
|
return
|
|
|
|
with ui.grid().classes("w-full grid-cols-2 md:grid-cols-3 xl:grid-cols-4 gap-3"):
|
|
for source in ordered_sources:
|
|
source_url = resolve_media_url(
|
|
source.file_path,
|
|
upload_dir=settings.upload_dir,
|
|
base_url=str(request.base_url),
|
|
)
|
|
with archival_card(extra_classes="gap-2"):
|
|
if source_url is None:
|
|
render_empty_state("Image unavailable.", extra_classes="text-xs")
|
|
else:
|
|
ui.image(source_url).classes("w-full aspect-[3/4] object-contain rounded-sm bg-black/5")
|
|
ui.label(f"Page {source.page_number}").classes("text-xs font-semibold")
|
|
ui.label(source.filename).classes("text-[11px] ui-text-muted break-all")
|
|
ui.button(
|
|
"Open Source Detail",
|
|
on_click=lambda _=None, source_id=source.id: ui.navigate.to(
|
|
f"/sources/{source_id}?from=document&document_id={document.id}"
|
|
),
|
|
icon="description",
|
|
).props("flat dense").classes("text-xs self-start")
|
|
|
|
@ui.page("/documents/{document_id}/edit")
|
|
async def document_edit_page(document_id: str, session_factory: SessionFactoryDep) -> None:
|
|
document_service = DocumentService(session_factory=session_factory)
|
|
people_service = PeopleService(session_factory=session_factory)
|
|
render_navigation_header(current_path="/documents")
|
|
|
|
parsed_doc_id = parsed_record_id(document_id, noun="Document")
|
|
if parsed_doc_id is None:
|
|
return
|
|
|
|
try:
|
|
document = await document_service.read_document_detail(document_id=parsed_doc_id)
|
|
except DocumentError:
|
|
render_record_not_found("Document")
|
|
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.")
|
|
|
|
people = sorted(await people_service.list_people(), key=lambda item: item.full_name.casefold())
|
|
role_catalog = list(await people_service.list_person_roles(active_only=False))
|
|
type_catalog = await document_service.list_document_types(active_only=False)
|
|
tag_catalog = await document_service.list_tags(active_only=False)
|
|
linked_people = LinkedPeopleEditor(
|
|
people=people,
|
|
roles=role_catalog,
|
|
initial_links=list(document.document_people),
|
|
)
|
|
form = _render_document_form_fields(
|
|
document=document,
|
|
type_options={str(doc_type.id): doc_type.label for doc_type in type_catalog},
|
|
tag_options=[tag.label for tag in tag_catalog],
|
|
linked_people=linked_people,
|
|
)
|
|
|
|
async def submit_edit() -> None:
|
|
candidate_name = (form.name.value or "").strip()
|
|
candidate_type_id = _resolve_selected_document_type_id(form.document_type.value, form.type_options)
|
|
if not candidate_name:
|
|
ui.notify("Document name is required.", type="warning")
|
|
return
|
|
if candidate_type_id is None:
|
|
ui.notify("Document type is required.", type="warning")
|
|
return
|
|
|
|
parsed_date = parse_iso_date(form.document_date.value)
|
|
if form.document_date.value and parsed_date is None:
|
|
ui.notify("Exact date must use YYYY-MM-DD.", type="warning")
|
|
return
|
|
|
|
candidate = Document(
|
|
id=document.id,
|
|
name=candidate_name,
|
|
document_type_id=candidate_type_id,
|
|
document_date=parsed_date,
|
|
document_date_raw=(form.document_date_raw.value or "").strip() or None,
|
|
location_created=(form.location.value or "").strip() or None,
|
|
notes=(form.notes.value or "").strip() or None,
|
|
archive_identifier=(form.archive.value or "").strip() or None,
|
|
created_at=document.created_at,
|
|
updated_at=document.updated_at,
|
|
)
|
|
|
|
save_outcome = await run_ui_action(
|
|
operation="documents.edit.save",
|
|
title="Save failed",
|
|
action=lambda: update_document_with_people(
|
|
document=candidate,
|
|
links=linked_people.values(),
|
|
tag_labels=_resolve_selected_tag_labels(form.tags.value),
|
|
documents=document_service,
|
|
people=people_service,
|
|
),
|
|
)
|
|
if not save_outcome.ok:
|
|
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:
|
|
document_service = DocumentService(session_factory=session_factory)
|
|
render_navigation_header(current_path="/documents")
|
|
|
|
parsed_doc_id = parsed_record_id(document_id, noun="Document")
|
|
if parsed_doc_id is None:
|
|
return
|
|
|
|
try:
|
|
document = await document_service.read_document_detail(document_id=parsed_doc_id)
|
|
except DocumentError:
|
|
render_record_not_found("Document")
|
|
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")
|
|
|
|
if document.sources or document.jobs:
|
|
render_delete_blocked_notice(
|
|
reason="Delete is blocked because related records exist.",
|
|
detail=dependency_summary([("Sources", bool(document.sources)), ("Jobs", bool(document.jobs))]),
|
|
guidance="Remove related records first, then retry deletion.",
|
|
back_label="Back to Document",
|
|
back_target=f"/documents/{document.id}",
|
|
)
|
|
return
|
|
|
|
ui.label("This action permanently deletes the document.").classes("text-xs ui-text-danger 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")
|
|
|
|
render_delete_actions(
|
|
confirm_label="Delete document permanently",
|
|
on_confirm=submit_delete,
|
|
cancel_target=f"/documents/{document.id}",
|
|
)
|
|
|
|
|
|
# --- Helper Sub-Components ---
|
|
|
|
|
|
def _render_document_form_fields(
|
|
*,
|
|
document: Document | None = None,
|
|
type_options: dict[str, str],
|
|
tag_options: list[str],
|
|
linked_people: LinkedPeopleEditor,
|
|
) -> DocumentFormFields:
|
|
with archival_card(extra_classes="gap-3"):
|
|
name_input = (
|
|
ui.input(label="Document name", value=document.name if document else "")
|
|
.props("outlined")
|
|
.classes("w-full ui-form-surface")
|
|
)
|
|
ordered_types = sorted(type_options.items(), key=lambda item: item[1].casefold())
|
|
type_display_to_id = {label: type_id for type_id, label in ordered_types}
|
|
type_display_options = list(type_display_to_id.keys())
|
|
selected_type = (
|
|
type_options[str(document.document_type_id)]
|
|
if document and document.document_type_id is not None and str(document.document_type_id) in type_options
|
|
else (type_display_options[0] if type_display_options else "")
|
|
)
|
|
type_input = (
|
|
ui.select(type_display_options, label="Document type").props("outlined").classes("w-full ui-form-surface")
|
|
)
|
|
if selected_type:
|
|
type_input.value = selected_type
|
|
|
|
with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-2"):
|
|
date_input = (
|
|
ui.input(
|
|
label="Document date",
|
|
value=document.document_date.isoformat() if document and document.document_date else "",
|
|
)
|
|
.props('outlined type="date"')
|
|
.classes("ui-form-surface")
|
|
)
|
|
date_raw_input = (
|
|
ui.input(
|
|
label="Approximate date",
|
|
value=document.document_date_raw if document and document.document_date_raw else "",
|
|
)
|
|
.props("outlined")
|
|
.classes("ui-form-surface")
|
|
)
|
|
|
|
location_input = (
|
|
ui.input(
|
|
label="Document location",
|
|
value=document.location_created if document and document.location_created else "",
|
|
)
|
|
.props("outlined")
|
|
.classes("w-full ui-form-surface")
|
|
)
|
|
archive_input = (
|
|
ui.input(
|
|
label="Archive identifier",
|
|
value=document.archive_identifier if document and document.archive_identifier else "",
|
|
)
|
|
.props("outlined")
|
|
.classes("w-full ui-form-surface")
|
|
)
|
|
notes_input = (
|
|
ui.textarea(label="Notes", value=document.notes if document and document.notes else "")
|
|
.props("outlined autogrow")
|
|
.classes("w-full ui-form-surface")
|
|
)
|
|
selected_tags = (
|
|
sorted(
|
|
[
|
|
link.tag_ref.label
|
|
for link in (document.document_tags if document is not None else [])
|
|
if link.tag_ref is not None
|
|
],
|
|
key=str.casefold,
|
|
)
|
|
if document is not None
|
|
else []
|
|
)
|
|
tags_input = (
|
|
ui.select(
|
|
sorted(tag_options, key=str.casefold),
|
|
label="Tags",
|
|
value=selected_tags,
|
|
multiple=True,
|
|
with_input=True,
|
|
new_value_mode="add-unique",
|
|
)
|
|
.props("outlined use-chips")
|
|
.classes("w-full ui-form-surface")
|
|
)
|
|
|
|
linked_people.render()
|
|
|
|
return DocumentFormFields(
|
|
name=name_input,
|
|
document_type=type_input,
|
|
type_options=type_display_to_id,
|
|
document_date=date_input,
|
|
document_date_raw=date_raw_input,
|
|
location=location_input,
|
|
archive=archive_input,
|
|
notes=notes_input,
|
|
tags=tags_input,
|
|
)
|
|
|
|
|
|
def _resolve_active_source(document: Document, requested_source_id: UUID | None) -> Source | None:
|
|
ordered = _sorted_document_sources(document)
|
|
if not ordered:
|
|
return None
|
|
if requested_source_id is None:
|
|
return ordered[0]
|
|
for source in ordered:
|
|
if source.id == requested_source_id:
|
|
return source
|
|
return ordered[0]
|
|
|
|
|
|
def _sorted_document_sources(document: Document) -> list[Source]:
|
|
return sorted(
|
|
document.sources,
|
|
key=lambda source: (source.page_number, source.upload_name.casefold(), source.filename.casefold()),
|
|
)
|
|
|
|
|
|
def _render_document_detail_viewer_zone(
|
|
*,
|
|
document: Document,
|
|
active_source: Source | None,
|
|
base_url: str,
|
|
settings: Settings,
|
|
) -> None:
|
|
with ui.column().classes("col-span-12 lg:col-span-4 gap-2"):
|
|
_render_document_source_navigation(document=document, active_source=active_source)
|
|
if active_source is None:
|
|
render_document_panzoom(media_url=None, filename="No source pages", count_label="0 Source Pages")
|
|
return
|
|
source_url = resolve_media_url(active_source.file_path, upload_dir=settings.upload_dir, base_url=base_url)
|
|
render_document_panzoom(
|
|
media_url=source_url,
|
|
filename=active_source.filename,
|
|
count_label=f"Page {active_source.page_number}",
|
|
)
|
|
|
|
|
|
def _render_document_source_navigation(*, document: Document, active_source: Source | None) -> None:
|
|
ordered = sorted(document.sources, key=lambda source: (source.page_number, source.id))
|
|
if not ordered or active_source is None:
|
|
with ui.row().classes("w-full justify-between items-center"):
|
|
ui.button("Previous Page", icon="chevron_left").props("flat dense disable")
|
|
ui.button("Next Page", icon="chevron_right").props("flat dense icon-right disable")
|
|
return
|
|
active_index = next((index for index, source in enumerate(ordered) if source.id == active_source.id), 0)
|
|
previous_source = ordered[active_index - 1] if active_index > 0 else None
|
|
next_source = ordered[active_index + 1] if active_index < len(ordered) - 1 else None
|
|
previous_target = f"/documents/{document.id}?source_id={previous_source.id}" if previous_source is not None else "#"
|
|
next_target = f"/documents/{document.id}?source_id={next_source.id}" if next_source is not None else "#"
|
|
with ui.row().classes("w-full justify-between items-center"):
|
|
previous = ui.button(
|
|
"Previous Page",
|
|
on_click=lambda: ui.navigate.to(previous_target),
|
|
icon="chevron_left",
|
|
).props("flat dense")
|
|
if previous_source is None:
|
|
previous.props("disable")
|
|
following = ui.button(
|
|
"Next Page",
|
|
on_click=lambda: ui.navigate.to(next_target),
|
|
icon="chevron_right",
|
|
).props("flat dense icon-right")
|
|
if next_source is None:
|
|
following.props("disable")
|
|
|
|
|
|
def _render_document_detail_revision_zone(*, source: Source | None, sources_service: SourceService) -> None:
|
|
with ui.column().classes("col-span-12 lg:col-span-6 gap-4"), archival_card(title="Editable Revision"):
|
|
if source is None:
|
|
render_empty_state("No source pages are linked yet.", italic=True)
|
|
return
|
|
seed_revision = source.revised_text if source.revised_text is not None else (source.raw_transcription or "")
|
|
revision_input = (
|
|
ui.textarea(
|
|
label="Revised transcription",
|
|
value=seed_revision,
|
|
)
|
|
.props("outlined autogrow")
|
|
.classes("w-full ui-form-surface")
|
|
)
|
|
save_state = ui.label(
|
|
f"Last saved: {source.date_revised.isoformat()}"
|
|
if source.date_revised is not None
|
|
else "No revision saved yet."
|
|
).classes("text-xs ui-text-muted")
|
|
|
|
async def submit_revision() -> None:
|
|
revised_text = (revision_input.value or "").strip()
|
|
if not revised_text:
|
|
ui.notify("Revised transcription cannot be empty.", type="warning")
|
|
return
|
|
save_outcome = await run_ui_action(
|
|
operation="documents.revision.save",
|
|
title="Save failed",
|
|
action=lambda: sources_service.upsert_revision_for_source(source_id=source.id, text=revised_text),
|
|
)
|
|
if not save_outcome.ok or save_outcome.value is None:
|
|
return
|
|
updated = save_outcome.value
|
|
source.revised_text = updated.revised_text
|
|
source.date_revised = updated.date_revised
|
|
save_state.text = (
|
|
f"Last saved: {updated.date_revised.isoformat()}"
|
|
if updated.date_revised is not None
|
|
else "Revision saved."
|
|
)
|
|
ui.notify("Revision saved", type="positive")
|
|
|
|
with ui.row().classes("w-full items-center gap-2 mt-2"):
|
|
ui.button("Save revision", on_click=submit_revision, icon="save").classes("ui-btn-primary")
|
|
ui.button(
|
|
"Reset",
|
|
on_click=lambda: _reset_document_revision_text(revision_input, source),
|
|
icon="refresh",
|
|
).props("flat")
|
|
|
|
|
|
def _first_source_path(document: Document) -> str | None:
|
|
first_source = _resolve_active_source(document, None)
|
|
return first_source.file_path if first_source is not None else None
|
|
|
|
|
|
def _render_bento_metadata_zone(document: Document) -> None:
|
|
author_names = _author_names(document)
|
|
|
|
with ui.column().classes("w-full gap-4"):
|
|
with archival_card(title="Archival Metadata"):
|
|
metadata_row("Author(s):", ", ".join(author_names) if author_names else "Not set")
|
|
metadata_row(
|
|
"Document Type:",
|
|
document.document_type_ref.label if document.document_type_ref is not None else "Not set",
|
|
)
|
|
tags = sorted(
|
|
[link.tag_ref.label for link in document.document_tags if link.tag_ref is not None],
|
|
key=str.casefold,
|
|
)
|
|
metadata_row("Tags:", ", ".join(tags) if tags else "Not set")
|
|
metadata_row("Document Date:", _detail_document_date(document.document_date, document.document_date_raw))
|
|
if document.location_created:
|
|
metadata_link_row(
|
|
"Location Created:",
|
|
document.location_created,
|
|
google_maps_search_url(document.location_created),
|
|
)
|
|
else:
|
|
metadata_row("Location Created:", "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 _detail_document_date(exact: date | None, approximate: str | None) -> str:
|
|
if exact is not None:
|
|
return exact.strftime("%m-%d-%Y")
|
|
return (approximate or "").strip() or "Unknown"
|
|
|
|
|
|
def _render_bento_relations_zone(document: Document) -> None:
|
|
with ui.column().classes("col-span-12 lg:col-span-2 gap-4"):
|
|
_render_related_people_card(document)
|
|
_render_document_processing_card(document)
|
|
|
|
|
|
def _render_related_people_card(document: Document) -> None:
|
|
with archival_card(title="Related People"):
|
|
if not document.document_people:
|
|
render_empty_state("No linked people yet.", italic=True)
|
|
return
|
|
|
|
grouped = _group_people_by_role(document)
|
|
with ui.column().classes("w-full gap-2"):
|
|
for role_label in sorted(grouped.keys(), key=str.casefold):
|
|
with ui.column().classes("w-full ui-row-surface p-2 gap-1"):
|
|
archival_badge(role_label)
|
|
for person in grouped[role_label]:
|
|
ui.button(
|
|
person.full_name,
|
|
on_click=lambda _=None, person_id=person.id: ui.navigate.to(
|
|
f"/people/{person_id}?from=document&document_id={document.id}"
|
|
),
|
|
icon="person",
|
|
).props("flat dense no-caps").classes("self-start text-xs font-semibold ui-link-primary")
|
|
|
|
|
|
def _render_document_processing_card(document: Document) -> None:
|
|
with archival_card(title="Source Pages & Transcriptions"):
|
|
metadata_row("Source pages:", str(len(document.sources)))
|
|
metadata_row("Transcription Jobs:", str(len(document.jobs)))
|
|
with ui.row().classes("w-full gap-2 mt-2 flex-wrap"):
|
|
ui.button(
|
|
"View Source Images",
|
|
on_click=lambda: ui.navigate.to(f"/documents/{document.id}/sources"),
|
|
icon="photo_library",
|
|
).props("flat dense text-xs").classes("ui-link-primary")
|
|
ui.button(
|
|
"View Transcription 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")
|
|
|
|
|
|
def _resolve_selected_document_type_id(selected_value: Any, type_options: dict[str, str]) -> UUID | None:
|
|
candidate = str(selected_value).strip() if selected_value is not None else ""
|
|
if not candidate:
|
|
return None
|
|
selected_id = type_options.get(candidate)
|
|
return parse_uuid(selected_id)
|
|
|
|
|
|
def _group_people_by_role(document: Document) -> dict[str, list[Any]]:
|
|
grouped: dict[str, list[Any]] = {}
|
|
for link in document.document_people:
|
|
if link.role_ref is not None and link.person is not None:
|
|
grouped.setdefault(link.role_ref.label, []).append(link.person)
|
|
for people in grouped.values():
|
|
people.sort(key=lambda person: person.full_name.casefold())
|
|
return grouped
|
|
|
|
|
|
def _author_names(document: Document) -> list[str]:
|
|
return sorted(
|
|
(
|
|
link.person.full_name
|
|
for link in document.document_people
|
|
if link.person is not None
|
|
and link.role_ref is not None
|
|
and link.role_ref.semantic_key == AUTHOR_ROLE_SEMANTIC_KEY
|
|
),
|
|
key=str.casefold,
|
|
)
|
|
|
|
|
|
def _tag_labels(document: Document) -> list[str]:
|
|
return sorted(
|
|
(
|
|
link.tag_ref.label
|
|
for link in document.document_tags
|
|
if link.tag_ref is not None and link.tag_ref.label.strip()
|
|
),
|
|
key=str.casefold,
|
|
)
|
|
|
|
|
|
def _resolve_selected_tag_labels(value: object) -> list[str]:
|
|
def flatten(item: object) -> list[str]:
|
|
if item is None:
|
|
return []
|
|
if isinstance(item, str):
|
|
return [item]
|
|
if isinstance(item, dict):
|
|
if "value" in item:
|
|
return flatten(item.get("value"))
|
|
if "label" in item:
|
|
return flatten(item.get("label"))
|
|
return []
|
|
if isinstance(item, (list, tuple, set)):
|
|
values: list[str] = []
|
|
for child in item:
|
|
values.extend(flatten(child))
|
|
return values
|
|
return [str(item)]
|
|
|
|
labels = [candidate.strip() for candidate in flatten(value) if candidate and candidate.strip()]
|
|
return list(dict.fromkeys(labels))
|
|
|
|
|
|
def _latest_job_status(document: Document) -> str | None:
|
|
if not document.jobs:
|
|
return None
|
|
latest = max(document.jobs, key=lambda job: (job.date_created, str(job.id)))
|
|
return latest.status.value
|
|
|
|
|
|
def _reset_document_revision_text(revision_input: ui.textarea, source: Source) -> None:
|
|
revision_input.value = source.revised_text if source.revised_text is not None else (source.raw_transcription or "")
|