2 Commits
Author SHA1 Message Date
John Lancaster f7192a33dc document form 2026-08-04 00:36:27 -05:00
John Lancaster 5932c0d3a1 global theming 2026-08-03 22:56:04 -05:00
16 changed files with 426 additions and 265 deletions
+1
View File
@@ -12,6 +12,7 @@ def create_cli_app() -> FastAPI:
def main() -> None:
settings = parse_cli_settings()
uvicorn.run(
"transcription.__main__:create_cli_app",
factory=True,
+5 -5
View File
@@ -5,12 +5,12 @@ from fastapi import APIRouter
router = APIRouter()
def healthz() -> dict[str, str]:
"""Return a simple health status payload."""
return {"status": "ok"}
@router.get("/healthz")
def healthz_route() -> dict[str, str]:
"""Route wrapper for health status payload."""
return healthz()
def healthz() -> dict[str, str]:
"""Return a simple health status payload."""
return {"status": "ok"}
+2 -5
View File
@@ -13,7 +13,6 @@ 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,8 +24,7 @@ from .db import dispose_database_runtime
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 .ui.pages import register_pages
from .worker import worker_consumer_lifespan
logger = logging.getLogger(__name__)
@@ -98,8 +96,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
def health() -> dict[str, str]:
return {"status": "ok"}
app.include_router(health_router)
register_error_handlers(app)
register_pages(app)
app.include_router(health_router)
ui.colors(**THEME_COLORS)
return app
+39 -12
View File
@@ -9,8 +9,8 @@ from typing import Optional
from uuid import UUID
from uuid import uuid4
from sqlalchemy import Column
from sqlalchemy import JSON
from sqlalchemy import Column
from sqlalchemy import UniqueConstraint
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm.exc import DetachedInstanceError
@@ -67,7 +67,23 @@ class Document(SQLModel, table=True):
jobs: list["Job"] = Relationship(back_populates="document", sa_relationship_kwargs={"lazy": "selectin"})
sources: list["Source"] = Relationship(back_populates="document", sa_relationship_kwargs={"lazy": "selectin"})
document_people: list["DocumentPerson"] = Relationship(back_populates="document", sa_relationship_kwargs={"lazy": "selectin"})
document_people: list["DocumentPerson"] = Relationship(
back_populates="document", sa_relationship_kwargs={"lazy": "selectin"}
)
@property
def authors(self):
"""Return linked people whose role is AUTHOR."""
return [
link.person
for link in self.document_people
if link.role == DocumentPersonRole.AUTHOR and link.person is not None
]
@property
def author(self):
"""Return the first linked author for convenience in read paths."""
return self.authors[0] if self.authors else None
class Person(SQLModel, table=True):
@@ -92,13 +108,24 @@ class Person(SQLModel, table=True):
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
document_people: list["DocumentPerson"] = Relationship(back_populates="person", sa_relationship_kwargs={"lazy": "selectin"})
document_people: list["DocumentPerson"] = Relationship(
back_populates="person", sa_relationship_kwargs={"lazy": "selectin"}
)
@property
def authored_documents(self):
"""Return documents where this person is linked as AUTHOR."""
return [
link.document
for link in self.document_people
if link.role == DocumentPersonRole.AUTHOR and link.document is not None
]
class DocumentPerson(SQLModel, table=True):
"""Associates documents with people in a given role."""
__tablename__ = "document_person"
__tablename__: str = "document_person"
id: UUID = Field(default_factory=uuid4, primary_key=True)
document_id: UUID = Field(foreign_key="document.id")
@@ -106,12 +133,14 @@ class DocumentPerson(SQLModel, table=True):
role: DocumentPersonRole = Field(default=DocumentPersonRole.AUTHOR)
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
__table_args__ = (
UniqueConstraint("document_id", "person_id", "role", name="uq_document_person_role"),
)
__table_args__ = (UniqueConstraint("document_id", "person_id", "role", name="uq_document_person_role"),)
document: Optional["Document"] = Relationship(back_populates="document_people", sa_relationship_kwargs={"lazy": "selectin"})
person: Optional["Person"] = Relationship(back_populates="document_people", sa_relationship_kwargs={"lazy": "selectin"})
document: Optional["Document"] = Relationship(
back_populates="document_people", sa_relationship_kwargs={"lazy": "selectin"}
)
person: Optional["Person"] = Relationship(
back_populates="document_people", sa_relationship_kwargs={"lazy": "selectin"}
)
class Job(SQLModel, table=True):
@@ -185,7 +214,7 @@ class Source(SQLModel, table=True):
class JobSource(SQLModel, table=True):
"""A single AI execution record for one source page."""
__tablename__ = "job_source"
__tablename__: str = "job_source"
id: UUID = Field(default_factory=uuid4, primary_key=True)
job_id: UUID = Field(foreign_key="job.id")
@@ -199,5 +228,3 @@ class JobSource(SQLModel, table=True):
job: Optional["Job"] = Relationship(back_populates="job_sources", sa_relationship_kwargs={"lazy": "selectin"})
source: Optional["Source"] = Relationship(back_populates="job_sources", sa_relationship_kwargs={"lazy": "selectin"})
-33
View File
@@ -1,33 +0,0 @@
"""UI page registration exports."""
from fastapi import FastAPI
from nicegui import ui
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
from transcription.ui.pages.upload_page import register_page as register_upload_page
from transcription.ui.resources import read_css
_THEME_REGISTERED_STATE_KEY = "transcription_ui_theme_registered"
def _register_global_styles(app: FastAPI) -> None:
if getattr(app.state, _THEME_REGISTERED_STATE_KEY, False):
return
ui.add_css(read_css("theme.css"), shared=True)
setattr(app.state, _THEME_REGISTERED_STATE_KEY, True)
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_pages()
register_people_page()
register_sources_page()
register_jobs_page()
ui.run_with(app, mount_path="/ui", show_welcome_message=False, dark=False)
@@ -0,0 +1,111 @@
import logging
from uuid import UUID
from nicegui import ui
from nicegui.binding import bindable_dataclass
from sqlmodel.ext.asyncio.session import AsyncSession
from ...db.models import Document
from ...db.models import DocumentPersonRole
from ...services.people import get_name_options
from .cards import archival_card
PROPS = "outlined bg-white"
CREATE_NEW_PERSON_OPTION = "__create_new_person__"
logger = logging.getLogger(__name__)
@bindable_dataclass
class DocumentEditForm:
id: UUID | None = None
name: str | None = None
document_type: str | None = None
document_date: str | None = None
document_date_raw: str | None = None
location_created: str | None = None
archive_identifier: str | None = None
notes: str | None = None
author_id: str | None = None
@classmethod
def from_table_model(cls, model: Document):
existing_author = next(
(item for item in model.document_people if item.role == DocumentPersonRole.AUTHOR),
None,
)
return cls(
id=model.id,
name=model.name,
document_type=model.document_type,
document_date=model.document_date.isoformat() if model.document_date else None,
document_date_raw=model.document_date_raw,
location_created=model.location_created,
archive_identifier=model.archive_identifier,
notes=model.notes,
author_id=str(existing_author.person_id) if existing_author is not None else "",
)
def save(self, session: AsyncSession) -> None:
"""Save the form data to the database."""
doc = session.get(Document, self.id)
if not doc:
logger.error("Document with ID %s not found in the database.", self.id)
return
async def render_document_edit_form(document: Document, session: AsyncSession) -> DocumentEditForm:
edit_form = DocumentEditForm.from_table_model(document)
with archival_card(extra_classes="gap-3"):
(ui.input("Document name").classes("w-full").props("autofocus").props(PROPS).bind_value(edit_form, "name"))
(
ui.input("Document type")
.classes("w-full")
.props("autofocus")
.props(PROPS)
.bind_value(edit_form, "document_type")
)
with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-2"):
(
ui.input("Exact date (YYYY-MM-DD)")
.props(PROPS)
.props('type="date"')
.bind_value(edit_form, "document_date")
)
(ui.input("Approximate date").bind_value(edit_form, "document_date_raw").props(PROPS))
ui.input("Document location").classes("w-full").props(PROPS).bind_value(edit_form, "location_created")
ui.input("Archive identifier").classes("w-full").props(PROPS).bind_value(edit_form, "archive_identifier")
(
ui.textarea("Notes")
.classes("w-full")
.props(PROPS)
.props("autogrow")
.bind_value(edit_form, "notes")
.props("rows=4")
)
people = await get_name_options(session=session)
author_options = {"": "No author", CREATE_NEW_PERSON_OPTION: "Create new item"} | {
str(person_id): name for person_id, name in people.items()
}
def on_author_change(event) -> None:
selected = str(event.value or "").strip()
if selected == CREATE_NEW_PERSON_OPTION:
ui.navigate.to("/people/new")
(
ui.select(
author_options,
label="Author (Person)",
value=edit_form.author_id or "",
on_change=on_author_change,
)
.classes("w-full")
.props(PROPS)
.bind_value(edit_form, "author_id")
)
ui.link("Create new person", "/people/new").classes("text-xs ui-link-primary font-medium")
return edit_form
+23
View File
@@ -0,0 +1,23 @@
from fastapi import FastAPI
from nicegui import ui
from transcription.ui.pages.documents import register_pages as register_documents_pages
from ..theme import register_global_styles
from .jobs_page import register_page as register_jobs_page
from .people_page import register_page as register_people_page
from .sources_page import register_page as register_sources_page
from .upload_page import register_page as register_upload_page
__all__ = ["register_pages"]
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_pages()
register_people_page()
register_sources_page()
register_jobs_page()
ui.run_with(app, mount_path="/ui", show_welcome_message=False, dark=False)
@@ -4,6 +4,7 @@ from fastapi import Request
from nicegui import ui
from ...dependency import ServicesDep
from ...dependency import SessionFactoryDep
from .delete_document import render_delete_document_page
from .document_detail import render_document_detail_page
from .document_overview import render_document_overview_page
@@ -29,8 +30,8 @@ def register_pages() -> 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)
async def document_edit_page(document_id: str, services: ServicesDep, session_factory: SessionFactoryDep) -> None:
await render_document_edit_page(document_id, services=services, session_factory=session_factory)
@ui.page("/documents/{document_id}/delete")
async def document_delete_page(document_id: str, services: ServicesDep) -> None:
@@ -8,16 +8,82 @@ from transcription.db.models import DocumentPerson
from transcription.db.models import DocumentPersonRole
from transcription.services.documents import DocumentError
from ....db.session import SessionFactory
from ...components.app_shell import render_navigation_header
from ...components.cards import archival_card
from ...components.document_form import CREATE_NEW_PERSON_OPTION
from ...components.document_form import DocumentEditForm
from ...components.document_form import render_document_edit_form
from ...components.error_presenter import show_error
from ...dependency import ServicesDep
from ...theme import page_header
CREATE_NEW_PERSON_OPTION = "__create_new_person__"
def _build_updated_document(edit_form: DocumentEditForm, document: Document) -> tuple[Document | None, str | None]:
candidate_name = (edit_form.name or "").strip()
candidate_type = (edit_form.document_type or "").strip()
if not candidate_name:
return None, "Document name is required."
if not candidate_type:
return None, "Document type is required."
parsed_date: date | None = None
candidate_date_text = (edit_form.document_date or "").strip()
if candidate_date_text:
try:
parsed_date = date.fromisoformat(candidate_date_text)
except ValueError:
return None, "Exact date must use YYYY-MM-DD."
return (
Document(
id=document.id,
name=candidate_name,
document_type=candidate_type,
document_date=parsed_date,
document_date_raw=(edit_form.document_date_raw or "").strip() or None,
location_created=(edit_form.location_created or "").strip() or None,
notes=(edit_form.notes or "").strip() or None,
archive_identifier=(edit_form.archive_identifier or "").strip() or None,
created_at=document.created_at,
updated_at=document.updated_at,
),
None,
)
async def render_document_edit_page(document_id: str, services: ServicesDep) -> None:
async def _sync_author_links(
services: ServicesDep,
document: Document,
selected_author: str,
) -> None:
existing_author_links = [
link for link in document.document_people if link.role == DocumentPersonRole.AUTHOR
]
if not selected_author:
for link in existing_author_links:
await services.documents.delete_document_person(link)
return
selected_author_id = UUID(selected_author)
if any(link.person_id == selected_author_id for link in existing_author_links):
return
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,
)
)
async def render_document_edit_page(
document_id: str,
services: ServicesDep,
session_factory: SessionFactory,
) -> None:
render_navigation_header(current_path="/documents")
try:
@@ -38,129 +104,29 @@ async def render_document_edit_page(document_id: str, services: ServicesDep) ->
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")
edit_form = await render_document_edit_form(document=document, session=session_factory())
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")
candidate, validation_error = _build_updated_document(edit_form, document)
if validation_error:
ui.notify(validation_error, type="warning")
return
if not candidate_type:
ui.notify("Document type is required.", type="warning")
if candidate is None:
ui.notify("Unable to build updated document.", 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()
selected_author = (edit_form.author_id 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,
)
)
await _sync_author_links(services=services, document=document, selected_author=selected_author)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Author update failed", operation="documents.edit.link_author")
return
+20 -15
View File
@@ -23,7 +23,6 @@ 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.jobs import render_jobs_table
from transcription.ui.theme import apply_archival_theme
from transcription.ui.theme import page_header
from transcription.worker import resolve_worker_notifier
@@ -36,7 +35,7 @@ def register_page() -> None: # noqa: PLR0915
@ui.page("/jobs")
async def jobs_page(session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
jobs_service = JobService(session_factory=session_factory)
render_navigation_header(current_path="/jobs")
@@ -68,12 +67,14 @@ def register_page() -> None: # noqa: PLR0915
@ui.page("/jobs/new")
async def job_create_page(request: Request, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
documents_service = DocumentService(session_factory=session_factory)
render_navigation_header(current_path="/jobs")
with ui.column().classes("w-full max-w-4xl mx-auto p-4 gap-4"):
page_header("Create Processing Job", subtitle="Queue source files for AI transcription and entity processing.")
page_header(
"Create Processing Job", subtitle="Queue source files for AI transcription and entity processing."
)
documents = await documents_service.list_documents()
if not documents:
@@ -88,7 +89,9 @@ def register_page() -> None: # noqa: PLR0915
on_click=lambda: ui.navigate.to("/documents/new?return_to=jobs_new"),
icon="note_add",
).classes("ui-btn-primary")
ui.button("Back to Jobs", on_click=lambda: ui.navigate.to("/jobs"), icon="arrow_back").props("flat")
ui.button("Back to Jobs", on_click=lambda: ui.navigate.to("/jobs"), icon="arrow_back").props(
"flat"
)
return
uploaded_files: list[tuple[str, bytes]] = []
@@ -136,18 +139,16 @@ def register_page() -> None: # noqa: PLR0915
with ui.column().classes("gap-1 w-full mt-2"):
for index, (filename, _) in ordered_uploads:
with ui.row().classes(
"w-full items-center justify-between ui-row-surface p-2"
):
with ui.row().classes("w-full items-center justify-between ui-row-surface p-2"):
ui.label(Path(filename).name).classes("text-xs font-mono ui-text-primary")
ui.button(icon="delete", on_click=lambda idx=index: remove_file(idx)).props(
"flat round dense color=negative text-xs"
)
with ui.row().classes("w-full justify-end mt-2"):
ui.button("Clear files", on_click=clear_files, icon="clear_all").props("flat dense").classes(
"text-xs text-red-800"
)
ui.button("Clear files", on_click=clear_files, icon="clear_all").props(
"flat dense"
).classes("text-xs text-red-800")
async def on_upload(event) -> None:
payload = await event.file.read()
@@ -204,7 +205,7 @@ def register_page() -> None: # noqa: PLR0915
@ui.page("/jobs/{job_id}")
async def job_detail_page(job_id: str, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
jobs_service = JobService(session_factory=session_factory)
render_navigation_header(current_path="/jobs")
@@ -256,7 +257,7 @@ def register_page() -> None: # noqa: PLR0915
@ui.page("/jobs/{job_id}/delete")
async def job_delete_page(job_id: str, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
jobs_service = JobService(session_factory=session_factory)
render_navigation_header(current_path="/jobs")
@@ -282,7 +283,9 @@ def register_page() -> None: # noqa: PLR0915
ui.label("Delete is blocked while the job is processing.").classes(
"text-xs text-red-800 font-bold mt-2"
)
ui.label("Wait for processing to complete, then retry delete.").classes("text-xs ui-text-muted italic")
ui.label("Wait for processing to complete, then retry delete.").classes(
"text-xs ui-text-muted italic"
)
with ui.row().classes("w-full items-center gap-2 mt-4"):
ui.button(
"Back to Job",
@@ -296,7 +299,9 @@ def register_page() -> None: # noqa: PLR0915
ui.label("This action permanently deletes the job.").classes("text-xs text-red-800 font-medium")
if job.job_sources:
ui.label("Related JobSource links will be removed as part of delete.").classes("text-xs ui-text-muted")
ui.label("Related JobSource links will be removed as part of delete.").classes(
"text-xs ui-text-muted"
)
async def submit_delete() -> None:
try:
+49 -33
View File
@@ -9,15 +9,15 @@ from uuid import UUID
from fastapi import Request
from nicegui import ui
from transcription.config import Settings, get_settings
from transcription.config import Settings
from transcription.config import get_settings
from transcription.db.models import Person
from transcription.errors import ErrorCategory
from transcription.services.documents import (
DocumentError,
DocumentService,
PersonDeleteBlockedError,
)
from transcription.services.store import UploadError, store_person_portrait
from transcription.services.documents import DocumentError
from transcription.services.documents import DocumentService
from transcription.services.documents import PersonDeleteBlockedError
from transcription.services.store import UploadError
from transcription.services.store import store_person_portrait
from transcription.ui.components.app_shell import render_navigation_header
from transcription.ui.components.cards import archival_card
from transcription.ui.components.data_display import metadata_row
@@ -25,9 +25,9 @@ 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.people import PersonTableRow, render_people_table
from transcription.ui.components.table.people import PersonTableRow
from transcription.ui.components.table.people import render_people_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
@@ -104,7 +104,7 @@ def register_page() -> None: # noqa: PLR0915
@ui.page("/people")
async def people_page(session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
people_service = DocumentService(session_factory=session_factory)
render_navigation_header(current_path="/people")
@@ -142,7 +142,7 @@ def register_page() -> None: # noqa: PLR0915
@ui.page("/people/new")
async def person_create_page(request: Request, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
people_service = DocumentService(session_factory=session_factory)
render_navigation_header(current_path="/people")
@@ -211,7 +211,7 @@ def register_page() -> None: # noqa: PLR0915
@ui.page("/people/{person_id}")
async def person_detail_page(person_id: str, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
people_service = DocumentService(session_factory=session_factory)
render_navigation_header(current_path="/people")
@@ -271,9 +271,7 @@ def register_page() -> None: # noqa: PLR0915
with ui.column().classes("col-span-12 lg:col-span-4 gap-4"):
with archival_card(title="Biography"):
ui.label(person.biography or "No biography recorded.").classes(
"p-2 ui-note-box text-xs w-full"
)
ui.label(person.biography or "No biography recorded.").classes("p-2 ui-note-box text-xs w-full")
with archival_card(title="Linked Documents"):
if not person.document_people:
@@ -285,9 +283,7 @@ def register_page() -> None: # noqa: PLR0915
document = link.document
if document is None:
continue
with ui.row().classes(
"w-full justify-between items-center ui-row-surface p-2"
):
with ui.row().classes("w-full justify-between items-center ui-row-surface p-2"):
with ui.column().classes("gap-0"):
ui.label(document.name).classes("text-xs font-semibold ui-text-primary")
ui.label(f"Role: {link.role.value}").classes("text-[10px] ui-text-muted")
@@ -301,7 +297,7 @@ def register_page() -> None: # noqa: PLR0915
@ui.page("/people/{person_id}/edit")
async def person_edit_page(person_id: str, request: Request, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
people_service = DocumentService(session_factory=session_factory)
render_navigation_header(current_path="/people")
@@ -329,7 +325,9 @@ def register_page() -> None: # noqa: PLR0915
display_name_input = ui.input(label="Display name", value=person.display_name or "").props(
"outlined bg-white"
)
maiden_name_input = ui.input(label="Maiden name", value=person.maiden_name or "").props("outlined bg-white")
maiden_name_input = ui.input(label="Maiden name", value=person.maiden_name or "").props(
"outlined bg-white"
)
with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-3"):
birth_date_input = ui.input(
@@ -339,7 +337,9 @@ def register_page() -> None: # noqa: PLR0915
birth_date_raw_input = ui.input(
label="Birth date (approximate)", value=person.birth_date_raw or ""
).props("outlined bg-white")
birth_place_input = ui.input(label="Birth place", value=person.birth_place or "").props("outlined bg-white")
birth_place_input = ui.input(label="Birth place", value=person.birth_place or "").props(
"outlined bg-white"
)
with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-3"):
death_date_input = ui.input(
@@ -349,13 +349,19 @@ def register_page() -> None: # noqa: PLR0915
death_date_raw_input = ui.input(
label="Death date (approximate)", value=person.death_date_raw or ""
).props("outlined bg-white")
death_place_input = ui.input(label="Death place", value=person.death_place or "").props("outlined bg-white")
death_place_input = ui.input(label="Death place", value=person.death_place or "").props(
"outlined bg-white"
)
biography_input = (
ui.textarea(label="Biography", value=person.biography or "").props("outlined bg-white autogrow").classes("w-full")
ui.textarea(label="Biography", value=person.biography or "")
.props("outlined bg-white autogrow")
.classes("w-full")
)
portrait_path_input = (
ui.input(label="Portrait path", value=person.portrait_path or "").props("outlined bg-white").classes("w-full")
ui.input(label="Portrait path", value=person.portrait_path or "")
.props("outlined bg-white")
.classes("w-full")
)
_bind_portrait_file_picker(portrait_path_input, settings=_resolve_runtime_settings(request))
@@ -401,11 +407,13 @@ def register_page() -> None: # noqa: PLR0915
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"/people/{person.id}"), icon="arrow_back").props("flat")
ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/people/{person.id}"), icon="arrow_back").props(
"flat"
)
@ui.page("/people/{person_id}/delete")
async def person_delete_page(person_id: str, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
people_service = DocumentService(session_factory=session_factory)
render_navigation_header(current_path="/people")
@@ -431,21 +439,27 @@ def register_page() -> None: # noqa: PLR0915
ui.label(f"Person: {person.full_name}").classes("text-sm font-semibold ui-text-primary")
if person.document_people:
ui.label("Delete is blocked because linked documents exist.").classes("text-xs text-red-800 font-bold mt-2")
ui.label("Delete is blocked because linked documents exist.").classes(
"text-xs text-red-800 font-bold mt-2"
)
ui.label(f"Linked documents: {len(person.document_people)}").classes("text-xs ui-text-muted")
ui.label("Remove document links first, then retry deletion.").classes("text-xs ui-text-muted italic")
ui.label("Remove document links 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 Person",
on_click=lambda: ui.navigate.to(f"/people/{person.id}"),
icon="arrow_back",
).classes("ui-btn-primary text-xs")
ui.button("Go to Documents", on_click=lambda: ui.navigate.to("/documents"), icon="description").props(
"flat text-xs"
)
ui.button(
"Go to Documents", on_click=lambda: ui.navigate.to("/documents"), icon="description"
).props("flat text-xs")
return
ui.label("This action permanently deletes the person record.").classes("text-xs text-red-800 font-medium")
ui.label("This action permanently deletes the person record.").classes(
"text-xs text-red-800 font-medium"
)
async def submit_delete() -> None:
try:
@@ -475,4 +489,6 @@ def register_page() -> None: # noqa: PLR0915
icon="delete_forever",
variant="solid",
)
ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/people/{person.id}"), icon="arrow_back").props("flat")
ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/people/{person.id}"), icon="arrow_back").props(
"flat"
)
+27 -23
View File
@@ -9,21 +9,20 @@ from fastapi import Request
from fastapi.responses import RedirectResponse
from nicegui import ui
from transcription.db.models import JobSource, Source
from transcription.services.documents import DocumentError, DocumentService
from transcription.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 (
TranscriptionNotFoundError,
TranscriptionService,
)
from transcription.services.transcription import TranscriptionNotFoundError
from transcription.services.transcription import TranscriptionService
from transcription.ui.components.app_shell import render_navigation_header
from transcription.ui.components.cards import archival_card
from transcription.ui.components.data_display import metadata_row
from transcription.ui.components.document_panzoom import render_document_panzoom
from transcription.ui.components.error_presenter import show_error
from transcription.ui.components.primitives import section_header_row
from transcription.ui.components.table.sources import SourceTableRow, render_sources_table
from transcription.ui.theme import apply_archival_theme
from transcription.ui.components.table.sources import SourceTableRow
from transcription.ui.components.table.sources import render_sources_table
from transcription.ui.theme import page_header
from ...db.session import SessionFactoryDep
@@ -34,7 +33,7 @@ def register_page() -> None:
@ui.page("/sources")
async def sources_page(request: Request, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
sources_service = TranscriptionService(session_factory=session_factory)
jobs_service = JobService(session_factory=session_factory)
documents_service = DocumentService(session_factory=session_factory)
@@ -56,7 +55,7 @@ def register_page() -> 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())))
sources = sorted(document.sources, key=lambda item: (item.page_number, item.upload_name.casefold()))
elif job_id is not None:
job = await jobs_service.read_job(job_id=job_id)
job_label = str(job.id)
@@ -65,7 +64,9 @@ def register_page() -> None:
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)))
sources = sorted(
await sources_service.list_sources(), key=lambda item: (item.document_id, item.page_number)
)
except DocumentError:
ui.label("Document not found").classes("text-h6 text-red-800 p-4")
return
@@ -89,9 +90,9 @@ def register_page() -> None:
if back_path is not None:
back_label = "Back to Document" if document_id is not None else "Back to Job"
ui.button(back_label, on_click=lambda route=back_path: ui.navigate.to(route), icon="arrow_back").classes(
"ui-btn-primary text-xs"
)
ui.button(
back_label, on_click=lambda route=back_path: ui.navigate.to(route), icon="arrow_back"
).classes("ui-btn-primary text-xs")
# Format source records into read-model rows for the table renderer
rows = [
@@ -108,7 +109,6 @@ def register_page() -> None:
@ui.page("/sources/{source_id}")
async def source_detail_page(source_id: str, request: Request, session_factory: SessionFactoryDep) -> None:
apply_archival_theme()
sources_service = TranscriptionService(session_factory=session_factory)
render_navigation_header(current_path="/sources")
@@ -131,7 +131,9 @@ def register_page() -> None:
with ui.column().classes("w-full max-w-[1800px] mx-auto p-4 gap-4"):
with section_header_row():
page_header(f"Source Page {source.page_number}: {source.upload_name}", subtitle=f"Source ID: {source.id}")
page_header(
f"Source Page {source.page_number}: {source.upload_name}", subtitle=f"Source ID: {source.id}"
)
if back_path is not None:
back_label = (
@@ -141,9 +143,9 @@ def register_page() -> None:
if "job_id" in request.query_params
else "Back to Sources"
)
ui.button(back_label, on_click=lambda route=back_path: ui.navigate.to(route), icon="arrow_back").classes(
"ui-btn-primary text-xs"
)
ui.button(
back_label, on_click=lambda route=back_path: ui.navigate.to(route), icon="arrow_back"
).classes("ui-btn-primary text-xs")
else:
ui.button("Back to Sources", on_click=lambda: ui.navigate.to("/sources"), icon="arrow_back").props(
"flat text-xs"
@@ -167,9 +169,9 @@ def register_page() -> None:
)
with archival_card(title="Automated Raw Transcription"):
ui.textarea(value=_source_transcription_text(source) or "").props("outlined autogrow readonly bg-white").classes(
"w-full text-xs font-mono"
)
ui.textarea(value=_source_transcription_text(source) or "").props(
"outlined autogrow readonly bg-white"
).classes("w-full text-xs font-mono")
with archival_card(title="Curated Human Transcription"):
revision_input = (
@@ -194,7 +196,9 @@ def register_page() -> None:
ui.navigate.to(request.url.path + _back_query(request.query_params))
with ui.row().classes("w-full items-center gap-2 mt-2"):
ui.button("Save Revision", on_click=save_revision, icon="save").classes("ui-btn-primary text-xs")
ui.button("Save Revision", on_click=save_revision, icon="save").classes(
"ui-btn-primary text-xs"
)
@ui.page("/documents/{document_id}/sources")
async def document_sources_page(document_id: str) -> RedirectResponse:
+18 -8
View File
@@ -1,5 +1,12 @@
from fastapi import FastAPI
from nicegui import app as nicegui_app
from nicegui import ui
from .resources import read_css
_THEME_REGISTERED_STATE_KEY = "transcription_ui_theme_registered"
# Runtime bridge for Quasar color slots. Visual ownership remains in theme.css tokens/classes.
THEME_COLORS = {
"primary": "#5e6572",
@@ -12,17 +19,20 @@ THEME_COLORS = {
"warning": "#a9b4c2",
}
_THEME_APPLIED = False
def apply_archival_theme() -> None:
"""Apply runtime color slots once; visual styling is defined in theme.css."""
global _THEME_APPLIED
if _THEME_APPLIED:
def register_global_styles(app: FastAPI) -> None:
if getattr(app.state, _THEME_REGISTERED_STATE_KEY, False):
return
ui.colors(**THEME_COLORS)
_THEME_APPLIED = True
theme_css = read_css("theme.css")
try:
ui.add_css(theme_css, shared=True)
except RuntimeError:
# NiceGUI can retain stale slot state across test app factories.
ui.add_head_html(f"<style>{theme_css}</style>", shared=True)
setattr(app.state, _THEME_REGISTERED_STATE_KEY, True)
nicegui_app.colors(**THEME_COLORS)
def page_header(title: str, subtitle: str | None = None) -> None:
+4 -12
View File
@@ -8,19 +8,11 @@ from transcription import __main__ as entrypoint
@pytest.mark.unit
def test_main_passes_constructed_app_to_uvicorn(monkeypatch):
"""Non-reload execution keeps the parsed settings instance in the app."""
def test_main_uses_cli_factory_import_string(monkeypatch):
"""Startup uses an importable factory so Uvicorn owns app creation."""
settings = SimpleNamespace(host="127.0.0.1", port=8123, log_level="debug", reload=False)
application = object()
captured = {}
def create_app(*, settings: object) -> object:
assert settings is expected_settings
return application
expected_settings = settings
monkeypatch.setattr(entrypoint, "parse_cli_settings", lambda: settings)
monkeypatch.setattr(entrypoint, "create_app", create_app)
monkeypatch.setattr(
entrypoint.uvicorn,
"run",
@@ -30,8 +22,8 @@ def test_main_passes_constructed_app_to_uvicorn(monkeypatch):
entrypoint.main()
assert captured == {
"application": application,
"factory": False,
"application": "transcription.__main__:create_cli_app",
"factory": True,
"host": "127.0.0.1",
"port": 8123,
"log_level": "debug",
+43 -2
View File
@@ -1,7 +1,5 @@
"""Tests for the V2 SQLModel persistence layer and relationships."""
from datetime import UTC
from datetime import datetime
from uuid import UUID
import pytest
@@ -203,3 +201,46 @@ class TestRelationships:
assert len(document.jobs) == 1
assert len(document.sources) == 1
assert len(document.document_people) == 1
def test_document_exposes_author_via_role_filtered_relation(self, session):
document = _persist_document(session)
author = _persist_person(session, full_name="Author Person")
recipient = _persist_person(session, full_name="Recipient Person")
session.add(DocumentPerson(document_id=document.id, person_id=author.id, role=DocumentPersonRole.AUTHOR))
session.add(DocumentPerson(document_id=document.id, person_id=recipient.id, role=DocumentPersonRole.RECIPIENT))
session.commit()
session.refresh(document)
assert [person.full_name for person in document.authors] == ["Author Person"]
assert document.author is not None
assert document.author.full_name == "Author Person"
def test_person_exposes_authored_documents_via_role_filtered_relation(self, session):
authored_document = _make_document(name="Authored Doc")
recipient_only_document = _make_document(name="Recipient Doc")
session.add(authored_document)
session.add(recipient_only_document)
session.commit()
session.refresh(authored_document)
session.refresh(recipient_only_document)
person = _persist_person(session, full_name="Dual Role Person")
session.add(
DocumentPerson(
document_id=authored_document.id,
person_id=person.id,
role=DocumentPersonRole.AUTHOR,
)
)
session.add(
DocumentPerson(
document_id=recipient_only_document.id,
person_id=person.id,
role=DocumentPersonRole.RECIPIENT,
)
)
session.commit()
session.refresh(person)
assert [document.name for document in person.authored_documents] == ["Authored Doc"]
+1 -1
View File
@@ -5,7 +5,7 @@ import re
import pytest
from fastapi import FastAPI
from transcription.ui import register_pages
from transcription.ui.pages import register_pages
from transcription.ui.resources import read_css