18 Commits
Author SHA1 Message Date
John Lancaster b8998025e2 delete button on transcript header 2026-06-29 20:35:21 -05:00
John Lancaster 002eb572e9 header slot 2026-06-29 20:19:19 -05:00
John Lancaster d44c7de684 gitignore updates 2026-06-29 19:05:24 -05:00
John Lancaster 282b0fb967 ui test updates 2026-06-29 19:05:15 -05:00
John Lancaster a9a47c3906 model used being carried thru 2026-06-29 19:04:55 -05:00
John Lancaster 9ada09accf job detail page stuff 2026-06-29 18:33:22 -05:00
John Lancaster 67b0980664 reworked zooming 2026-06-29 18:00:22 -05:00
John Lancaster e35a8ec060 coloring and page tweaks 2026-06-29 17:34:05 -05:00
John Lancaster 3a141bd4cc page tweaks 2026-06-29 14:08:35 -05:00
John Lancaster 8129f5a9e8 ui instructions 2026-06-29 13:59:36 -05:00
John Lancaster 58b4c381a4 added revisions to transcription table 2026-06-29 13:59:26 -05:00
John Lancaster 5719debbaa messing with zoom/reset 2026-06-29 08:06:02 -05:00
John Lancaster e7c7ab71b4 zooming fix 2026-06-28 22:20:48 -05:00
John Lancaster ca5c9f787f started panzoom thing 2026-06-28 22:10:06 -05:00
John Lancaster 8064821503 app_shell tweak 2026-06-28 15:19:32 -05:00
John Lancaster 593388ef3a test updates 2026-06-28 15:18:31 -05:00
John Lancaster 83ee7b31e0 styling 2026-06-28 15:18:06 -05:00
John Lancaster 455a01d7c4 started app shell 2026-06-28 14:58:29 -05:00
23 changed files with 870 additions and 223 deletions
+6
View File
@@ -0,0 +1,6 @@
---
description: Copilot rules for modifying the UI
applyTo: 'src/transcription/ui/**/*.py'
---
The UI is forbidden from directly touching the database. All interactions should be done using the [services](../../src/transcription/services/)
+5
View File
@@ -14,3 +14,8 @@ wheels/
# SQLite database # SQLite database
*.db *.db
upload/
*.jpg
*.jpeg
*.png
+9 -1
View File
@@ -8,6 +8,7 @@ from contextlib import asynccontextmanager
from fastapi import FastAPI from fastapi import FastAPI
from fastapi import status from fastapi import status
from fastapi.responses import RedirectResponse from fastapi.responses import RedirectResponse
from fastapi.staticfiles import StaticFiles
from .api.errors import register_error_handlers from .api.errors import register_error_handlers
from .api.health import router as health_router from .api.health import router as health_router
@@ -25,7 +26,7 @@ from .worker import worker_consumer_lifespan
async def _lifespan(app: FastAPI): async def _lifespan(app: FastAPI):
configure_logging() configure_logging()
settings = get_settings() settings = getattr(app.state, "settings", None) or get_settings()
app.state.settings = settings app.state.settings = settings
app.state.services = ServiceBundle() app.state.services = ServiceBundle()
app.state.runtime = initialize_database_runtime(settings=settings) app.state.runtime = initialize_database_runtime(settings=settings)
@@ -52,6 +53,13 @@ async def _lifespan(app: FastAPI):
def create_app() -> FastAPI: def create_app() -> FastAPI:
"""Create and configure the FastAPI application.""" """Create and configure the FastAPI application."""
app = FastAPI(title="Transcription", lifespan=_lifespan) app = FastAPI(title="Transcription", lifespan=_lifespan)
settings = get_settings()
app.state.settings = settings
app.mount(
"/uploads",
StaticFiles(directory=settings.upload_dir, check_dir=False),
name="uploads",
)
@app.get("/", include_in_schema=False) @app.get("/", include_in_schema=False)
async def root_redirect() -> RedirectResponse: async def root_redirect() -> RedirectResponse:
+9 -4
View File
@@ -51,10 +51,15 @@ def _ensure_sqlite_compat_columns(connection: Connection) -> None:
inspector = inspect(connection) inspector = inspect(connection)
table_names = set(inspector.get_table_names()) table_names = set(inspector.get_table_names())
if "job" not in table_names:
return
columns = {column["name"] for column in inspector.get_columns("job")} if "job" in table_names:
if "retry_count" not in columns: job_columns = {column["name"] for column in inspector.get_columns("job")}
if "retry_count" not in job_columns:
connection.execute(text("ALTER TABLE job ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0")) connection.execute(text("ALTER TABLE job ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0"))
logger.warning("Applied SQLite compatibility schema patch table=job column=retry_count default=0") logger.warning("Applied SQLite compatibility schema patch table=job column=retry_count default=0")
if "transcript" in table_names:
transcript_columns = {column["name"] for column in inspector.get_columns("transcript")}
if "model" not in transcript_columns:
connection.execute(text("ALTER TABLE transcript ADD COLUMN model VARCHAR NOT NULL DEFAULT 'unknown'"))
logger.warning("Applied SQLite compatibility schema patch table=transcript column=model default=unknown")
+12 -6
View File
@@ -1,16 +1,16 @@
"""SQLModel domain models for the transcription system. """SQLModel domain models for the transcription system.
Three models capture the MVP lifecycle: Three models capture the MVP lifecycle:
Document -> one-to-many -> Job -> one-to-one -> Transcript Document -> one-to-many -> Job -> one-to-many -> Transcript
""" """
from datetime import UTC from datetime import UTC
from datetime import datetime from datetime import datetime
from enum import StrEnum from enum import StrEnum
from typing import Optional
from uuid import UUID from uuid import UUID
from uuid import uuid4 from uuid import uuid4
from sqlalchemy import UniqueConstraint
from sqlmodel import Field from sqlmodel import Field
from sqlmodel import Relationship from sqlmodel import Relationship
from sqlmodel import SQLModel from sqlmodel import SQLModel
@@ -47,7 +47,7 @@ class Job(SQLModel, table=True):
# --- relationships --- # --- relationships ---
document: Document = Relationship(back_populates="jobs") document: Document = Relationship(back_populates="jobs")
transcript: Optional["Transcript"] = Relationship(back_populates="job") transcripts: list["Transcript"] = Relationship(back_populates="job")
@property @property
def filename(self) -> str: def filename(self) -> str:
@@ -59,10 +59,14 @@ class Transcript(SQLModel, table=True):
"""The output of a transcription job.""" """The output of a transcription job."""
id: UUID = Field(default_factory=uuid4, primary_key=True) id: UUID = Field(default_factory=uuid4, primary_key=True)
job_id: UUID = Field(foreign_key="job.id", unique=True) job_id: UUID = Field(foreign_key="job.id")
"""ID for the associated job. There's a 1-1 relationship bewteen transcripts and jobs.""" """ID for the associated job."""
revision: int = Field(default=0, ge=0)
"""Revision number for this job's transcript history, starting at 0."""
provider: str provider: str
"""Name of the transcription provider used to generate this transcript.""" """Name of the transcription provider used to generate this transcript."""
model: str
"""Model identifier used to generate this transcript revision."""
prompt_name: str prompt_name: str
"""Name of the prompt used to generate this transcript.""" """Name of the prompt used to generate this transcript."""
text: str | None = None text: str | None = None
@@ -71,5 +75,7 @@ class Transcript(SQLModel, table=True):
"""Details of any error that occurred during transcription.""" """Details of any error that occurred during transcription."""
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
__table_args__ = (UniqueConstraint("job_id", "revision", name="uq_transcript_job_revision"),)
# --- relationships --- # --- relationships ---
job: Job = Relationship(back_populates="transcript") job: Job = Relationship(back_populates="transcripts")
+3 -1
View File
@@ -28,12 +28,14 @@ class TranscriptionResult:
prompt_name: str prompt_name: str
model: str model: str
def to_transcript(self, job_id: UUID) -> Transcript: def to_transcript(self, job_id: UUID, *, revision: int = 0) -> Transcript:
"""Convert a TranscriptionResult to a Transcript model instance.""" """Convert a TranscriptionResult to a Transcript model instance."""
return Transcript( return Transcript(
job_id=job_id, job_id=job_id,
revision=revision,
provider=self.provider, provider=self.provider,
prompt_name=self.prompt_name, prompt_name=self.prompt_name,
model=self.model,
text=self.text, text=self.text,
) )
+4 -1
View File
@@ -35,7 +35,10 @@ class JobService(ServiceBase):
async with self._session_scope(session) as _session: async with self._session_scope(session) as _session:
query = ( query = (
select(Job) select(Job)
.options(selectinload(Job.document)) # pyright: ignore[reportArgumentType] .options(
selectinload(Job.document), # pyright: ignore[reportArgumentType]
selectinload(Job.transcripts), # pyright: ignore[reportArgumentType]
)
.where(Job.id == job_id) .where(Job.id == job_id)
.execution_options(populate_existing=True) .execution_options(populate_existing=True)
) )
+67 -13
View File
@@ -4,10 +4,12 @@ from __future__ import annotations
import logging import logging
import mimetypes import mimetypes
from collections.abc import Sequence
from contextlib import contextmanager from contextlib import contextmanager
from pathlib import Path from pathlib import Path
from uuid import UUID from uuid import UUID
from sqlalchemy import func
from sqlalchemy.ext.asyncio import async_sessionmaker from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.orm import selectinload from sqlalchemy.orm import selectinload
from sqlmodel import select from sqlmodel import select
@@ -108,38 +110,90 @@ class TranscriptionService(ServiceBase):
settings=self.settings, settings=self.settings,
provider=self.provider, provider=self.provider,
) )
await self.create_transcript(transcript=result.to_transcript(job_id=job_id), session=session) await self.create_transcript_for_job(
job_id=job_id,
text=result.text,
provider=result.provider,
model=result.model,
prompt_name=result.prompt_name,
session=session,
)
async def upsert_transcript_by_job( async def create_transcript_for_job(
self, self,
*, *,
job_id: UUID, job_id: UUID,
text: str | None, text: str | None,
error_detail: str | None, error_detail: str | None = None,
provider: str | None = None, provider: str | None = None,
model: str | None = None,
prompt_name: str = DEFAULT_PROMPT_FILE, prompt_name: str = DEFAULT_PROMPT_FILE,
session: AsyncSession | None = None, session: AsyncSession | None = None,
) -> Transcript: ) -> Transcript:
"""Create or update a transcript for a job id.""" """Create a new transcript revision for a job id."""
async with self._session_scope(session) as _session: async with self._session_scope(session) as _session:
transcript = (await _session.exec(select(Transcript).where(Transcript.job_id == job_id))).first() rev_query = select(func.max(Transcript.revision)).where(Transcript.job_id == job_id)
if transcript is None: rev_result = await _session.exec(rev_query)
max_revision = -1 if (rev := rev_result.one_or_none()) is None else rev
next_revision = max_revision + 1
transcript = Transcript( transcript = Transcript(
job_id=job_id, job_id=job_id,
revision=next_revision,
provider=provider or self.settings.provider.value, provider=provider or self.settings.provider.value,
model=model or _resolve_transcript_model(provider=self.provider, settings=self.settings),
prompt_name=prompt_name, prompt_name=prompt_name,
text=text,
error_detail=error_detail,
) )
transcript.text = text
transcript.error_detail = error_detail
if provider is not None:
transcript.provider = provider
transcript.prompt_name = prompt_name
_session.add(transcript) _session.add(transcript)
await self._finalize(session=_session, caller_session=session, refresh=(transcript,)) await self._finalize(session=_session, caller_session=session, refresh=(transcript,))
return transcript return transcript
async def read_latest_transcript_by_job(
self,
job_id: UUID,
*,
session: AsyncSession | None = None,
) -> Transcript | None:
"""Read the latest transcript revision for a job id."""
async with self._session_scope(session) as _session:
query = _transcript_job_query(job_id=job_id).limit(1)
result = await _session.exec(query)
return result.one_or_none()
async def list_transcripts_by_job(
self,
job_id: UUID,
*,
session: AsyncSession | None = None,
) -> Sequence[Transcript]:
"""List transcript revisions for a job id in ascending revision order."""
async with self._session_scope(session) as _session:
query = _transcript_job_query(job_id=job_id)
result = await _session.exec(query)
return result.all()
def _transcript_job_query(job_id: UUID):
return (
select(Transcript)
.where(Transcript.job_id == job_id)
.options(selectinload(Transcript.job)) # pyright: ignore[reportArgumentType]
.order_by(Transcript.revision) # pyright: ignore[reportArgumentType]
) # fmt: skip
def _resolve_transcript_model(*, provider: TranscriptionProvider, settings: Settings) -> str:
provider_model = getattr(provider, "model", None)
if isinstance(provider_model, str) and provider_model.strip():
return provider_model
if settings.provider_model and settings.provider_model.strip():
return settings.provider_model
return "unknown"
async def transcribe_document_image( async def transcribe_document_image(
image_path: str | Path, image_path: str | Path,
+8 -6
View File
@@ -121,11 +121,12 @@ async def _finalize_transcribed(
"""Transaction B: transcript + TRANSCRIBED in one commit.""" """Transaction B: transcript + TRANSCRIBED in one commit."""
if session is None: if session is None:
async with services.jobs._session_scope() as local_session: async with services.jobs._session_scope() as local_session:
await services.transcriptions.upsert_transcript_by_job( await services.transcriptions.create_transcript_for_job(
job_id=job.id, job_id=job.id,
text=result.text, text=result.text,
error_detail=None, error_detail=None,
provider=result.provider, provider=result.provider,
model=result.model,
prompt_name=result.prompt_name, prompt_name=result.prompt_name,
session=local_session, session=local_session,
) )
@@ -137,11 +138,12 @@ async def _finalize_transcribed(
await local_session.commit() await local_session.commit()
return updated_job return updated_job
await services.transcriptions.upsert_transcript_by_job( await services.transcriptions.create_transcript_for_job(
job_id=job.id, job_id=job.id,
text=result.text, text=result.text,
error_detail=None, error_detail=None,
provider=result.provider, provider=result.provider,
model=result.model,
prompt_name=result.prompt_name, prompt_name=result.prompt_name,
session=session, session=session,
) )
@@ -165,7 +167,7 @@ async def _finalize_retry(
"""Transaction C: transcript error + QUEUED + retry increment in one commit.""" """Transaction C: transcript error + QUEUED + retry increment in one commit."""
if session is None: if session is None:
async with services.jobs._session_scope() as local_session: async with services.jobs._session_scope() as local_session:
await services.transcriptions.upsert_transcript_by_job( await services.transcriptions.create_transcript_for_job(
job_id=job.id, job_id=job.id,
text=None, text=None,
error_detail=format_error_detail(error), error_detail=format_error_detail(error),
@@ -180,7 +182,7 @@ async def _finalize_retry(
) )
await local_session.commit() await local_session.commit()
else: else:
await services.transcriptions.upsert_transcript_by_job( await services.transcriptions.create_transcript_for_job(
job_id=job.id, job_id=job.id,
text=None, text=None,
error_detail=format_error_detail(error), error_detail=format_error_detail(error),
@@ -210,7 +212,7 @@ async def _finalize_failed(
"""Transaction B: transcript error + FAILED in one commit.""" """Transaction B: transcript error + FAILED in one commit."""
if session is None: if session is None:
async with services.jobs._session_scope() as local_session: async with services.jobs._session_scope() as local_session:
await services.transcriptions.upsert_transcript_by_job( await services.transcriptions.create_transcript_for_job(
job_id=job.id, job_id=job.id,
text=None, text=None,
error_detail=format_error_detail(error), error_detail=format_error_detail(error),
@@ -225,7 +227,7 @@ async def _finalize_failed(
await local_session.commit() await local_session.commit()
return updated_job return updated_job
await services.transcriptions.upsert_transcript_by_job( await services.transcriptions.create_transcript_for_job(
job_id=job.id, job_id=job.id,
text=None, text=None,
error_detail=format_error_detail(error), error_detail=format_error_detail(error),
+32 -1
View File
@@ -1,14 +1,45 @@
"""UI page registration exports.""" """UI page registration exports."""
from pathlib import Path
from fastapi import FastAPI from fastapi import FastAPI
from nicegui import app as nicegui_app
from nicegui import ui from nicegui import ui
from transcription.ui.pages.jobs_page import register_page as register_jobs_page from transcription.ui.pages.jobs_page import register_page as register_jobs_page
from transcription.ui.pages.upload_page import register_page as register_upload_page from transcription.ui.pages.upload_page import register_page as register_upload_page
_THEME_REGISTERED_STATE_KEY = "transcription_ui_theme_registered"
_THEME_COLORS: dict[str, str] = {
"primary": "#6f97e8",
"secondary": "#92b5f5",
"accent": "#7fc0de",
"dark": "#22304a",
"dark_page": "#1a2538",
"positive": "#86c8ad",
"negative": "#d98a9a",
"info": "#7ebdda",
"warning": "#e2c083",
}
def _register_global_styles(app: FastAPI) -> None:
if getattr(app.state, _THEME_REGISTERED_STATE_KEY, False):
return
nicegui_app.colors(**_THEME_COLORS)
css_path = Path(__file__).resolve().parent / "static" / "colors.css"
if css_path.exists():
ui.add_css(css_path, shared=True)
setattr(app.state, _THEME_REGISTERED_STATE_KEY, True)
def register_pages(app: FastAPI) -> None: def register_pages(app: FastAPI) -> None:
"""Register all NiceGUI pages and mount them onto the FastAPI app.""" """Register all NiceGUI pages and mount them onto the FastAPI app."""
_register_global_styles(app)
register_upload_page() register_upload_page()
register_jobs_page() register_jobs_page()
ui.run_with(app, mount_path="/ui", show_welcome_message=False) ui.run_with(app, mount_path="/ui", show_welcome_message=False, dark=True)
@@ -0,0 +1,7 @@
"""Reusable UI component exports."""
from transcription.ui.components.app_shell import NAV_ITEMS
from transcription.ui.components.app_shell import render_navigation_header
from transcription.ui.components.document_panzoom import render_document_panzoom
__all__ = ["NAV_ITEMS", "render_document_panzoom", "render_navigation_header"]
@@ -0,0 +1,59 @@
"""Reusable app shell primitives for page-level layout."""
from __future__ import annotations
from nicegui import ui
NAV_ITEMS: tuple[tuple[str, str, str], ...] = (
("Upload", "/upload", "upload_file"),
("Jobs", "/jobs", "work_history"),
)
def _is_active_path(*, current_path: str, item_path: str) -> bool:
if item_path == "/jobs":
return current_path == "/jobs" or current_path.startswith("/jobs/")
return current_path == item_path
def _button_props(*, icon: str, is_active: bool) -> str:
if is_active:
return f"icon={icon} no-caps unelevated color=primary text-color=white"
return f"icon={icon} no-caps outline color=secondary text-color=secondary"
def _button_classes(*, is_active: bool) -> str:
base = "w-full sm:w-auto min-h-[40px] px-3 rounded-lg text-body2 text-weight-medium"
if is_active:
return f"{base}"
return f"{base}"
def _render_nav_button(*, label: str, path: str, icon: str, current_path: str) -> None:
is_active = _is_active_path(current_path=current_path, item_path=path)
button = ui.button(
label,
icon=icon,
on_click=lambda _=None, route=path: ui.navigate.to(route),
)
button.props(_button_props(icon=icon, is_active=is_active)).classes(_button_classes(is_active=is_active))
def _normalize_path(current_path: str | None) -> str:
normalized = (current_path or "").strip()
if not normalized:
return "/upload"
return normalized.rstrip("/") or "/"
def render_navigation_header(*, current_path: str | None = None) -> None:
"""Render a shared app header with links for top-level pages."""
normalized_path = _normalize_path(current_path)
with (
ui.header(elevated=True).props("bordered").classes("bg-dark text-white q-px-sm q-py-xs"),
ui.row().classes("w-full items-center justify-end q-gutter-xs"),
ui.element("div").classes("w-full grid grid-cols-2 gap-2 sm:flex sm:justify-start sm:gap-2"),
):
for label, path, icon in NAV_ITEMS:
_render_nav_button(label=label, path=path, icon=icon, current_path=normalized_path)
@@ -0,0 +1,211 @@
"""Panzoom-backed document preview component."""
from __future__ import annotations
from functools import lru_cache
from pathlib import Path
from urllib.parse import quote
from uuid import uuid4
from nicegui import ui
from transcription.config import get_settings
from transcription.models import Document
PANGOZOOM_CDN_URL = "https://unpkg.com/@panzoom/[email protected]/dist/panzoom.min.js"
UPLOADS_URL_PREFIX = "/uploads"
def render_document_panzoom(*, document: Document) -> None:
"""Render a document preview with pan and zoom interactions."""
_register_panzoom_assets()
host_id = f"document-panzoom-{uuid4().hex}"
document_url = _document_url(document)
document_kind = _document_kind(document)
with ui.card().classes("w-full q-pa-md"):
with ui.row().classes("w-full items-center justify-between no-wrap"):
ui.label("Document preview").classes("text-subtitle1 text-weight-medium")
ui.label(document.filename).classes("text-caption text-grey-4 ellipsis").style(
"max-width: 60%; text-align: right;"
)
with (
ui.element("div").classes("w-full document-panzoom-host rounded-borders q-mt-md")
# .style(f"height: {height};")
) as host:
host.props(f"id={host_id}")
with ui.element("div").classes("document-panzoom-surface"):
if document_kind == "pdf":
ui.html(
f'<iframe class="document-panzoom-iframe" '
f'src="{document_url}" title="{document.filename}" '
"data-panzoom-target></iframe>"
)
else:
ui.html(
f'<img class="document-panzoom-media" '
f'src="{document_url}" alt="{document.filename}" '
"data-panzoom-target data-panzoom-media />"
)
_attach_panzoom(host_id)
@lru_cache(maxsize=1)
def _register_panzoom_assets() -> None:
ui.add_head_html(
f'<script src="{PANGOZOOM_CDN_URL}"></script>',
shared=True,
)
ui.add_head_html(
"""
<style>
.document-panzoom-host {
overflow: hidden;
touch-action: none;
}
.document-panzoom-surface {
width: 100%;
height: 100%;
display: flex;
align-items: flex-start;
justify-content: flex-start;
}
.document-panzoom-media {
width: auto;
height: auto;
display: block;
max-width: 100%;
max-height: 100%;
user-select: none;
-webkit-user-drag: none;
}
.document-panzoom-iframe {
width: 100%;
height: 100%;
border: 0;
pointer-events: none;
background: white;
}
</style>
""",
shared=True,
)
def _document_url(document: Document) -> str:
file_path = Path(document.file_path)
upload_dir = get_settings().upload_dir
relative_path: Path
try:
relative_path = file_path.resolve().relative_to(upload_dir.resolve())
except ValueError:
parts = file_path.parts
if "uploads" in parts:
uploads_index = parts.index("uploads")
relative_path = Path(*parts[uploads_index + 1 :])
else:
relative_path = Path(file_path.name)
encoded_relative_path = "/".join(quote(part) for part in relative_path.parts)
return f"{UPLOADS_URL_PREFIX}/{encoded_relative_path}"
def _document_kind(document: Document) -> str:
suffix = Path(document.file_path).suffix.lower()
if suffix == ".pdf":
return "pdf"
return "image"
def _attach_panzoom(host_id: str) -> None:
ui.run_javascript(
f"""
(function() {{
if (!window.Panzoom) return;
window.__transcriptionPanzoom = window.__transcriptionPanzoom || {{}};
const host = document.getElementById({host_id!r});
if (!host) return;
const target = host.querySelector('[data-panzoom-target]');
const media = host.querySelector('[data-panzoom-media]');
if (!target) return;
const cleanup = () => {{
const existing = window.__transcriptionPanzoom[{host_id!r}];
if (existing?.resizeObserver) existing.resizeObserver.disconnect();
if (existing?.wheelHandler) host.removeEventListener('wheel', existing.wheelHandler);
if (existing?.instance) existing.instance.destroy();
}};
const computeFitScale = () => {{
const hostRect = host.getBoundingClientRect();
if (hostRect.width <= 0 || hostRect.height <= 0) return null;
return 1;
}};
const buildInstance = () => {{
cleanup();
const fitScale = computeFitScale();
if (fitScale === null) return false;
const minScale = Math.min(fitScale, 0.01);
const instance = Panzoom(target, {{
startX: 0,
startY: 0,
startScale: fitScale,
minScale: minScale,
maxScale: 256,
step: 0.2,
roundPixels: false,
panOnlyWhenZoomed: true,
overflow: 'hidden',
}});
const wheelHandler = (event) => instance.zoomWithWheel(event);
host.addEventListener('wheel', wheelHandler, {{ passive: false }});
requestAnimationFrame(() => {{
instance.reset({{ animate: false }});
}});
const resizeObserver = new ResizeObserver(() => {{
const nextFitScale = computeFitScale();
if (nextFitScale === null) return;
instance.setOptions({{
startScale: nextFitScale,
minScale: Math.min(nextFitScale, 0.01),
}});
instance.reset({{ animate: false }});
}});
resizeObserver.observe(host);
window.__transcriptionPanzoom[{host_id!r}] = {{
instance,
wheelHandler,
resizeObserver,
}};
return true;
}};
const initWhenReady = (retries = 15) => {{
if (buildInstance()) return;
if (retries <= 0) return;
requestAnimationFrame(() => initWhenReady(retries - 1));
}};
if (media && media.tagName === 'IMG' && !media.complete) {{
media.addEventListener('load', () => initWhenReady(), {{ once: true }});
return;
}}
initWhenReady();
}})();
"""
)
+29 -17
View File
@@ -2,11 +2,17 @@
from __future__ import annotations from __future__ import annotations
import logging
from collections.abc import Sequence
from nicegui import ui from nicegui import ui
from transcription.models import Document from transcription.models import Document
from transcription.models import Job from transcription.models import Job
from transcription.models import Transcript from transcription.models import Transcript
from transcription.ui.components.document_panzoom import render_document_panzoom
logger = logging.getLogger(__name__)
def _status_chip_classes(status: str) -> str: def _status_chip_classes(status: str) -> str:
@@ -22,56 +28,62 @@ def _status_chip_classes(status: str) -> str:
def _metadata_row(label: str, value: str) -> None: def _metadata_row(label: str, value: str) -> None:
with ui.row().classes("w-full items-start justify-between no-wrap q-gutter-x-md"): with ui.row().classes("w-full items-start justify-between q-gutter-x-md"):
ui.label(label).classes("text-caption text-grey-7 text-uppercase") ui.label(label).classes("text-caption text-grey-5 text-uppercase w-28")
ui.label(value).classes("text-body2 text-right") ui.label(value).classes("text-body2 text-right text-grey-1 break-all")
def _render_document_section(document: Document) -> None: def _render_document_section(document: Document) -> None:
with ui.card().classes("w-full bg-grey-1 q-pa-md"): with ui.card().classes("w-full q-pa-md"):
ui.label("Document").classes("text-subtitle1 text-weight-medium") ui.label("Document").classes("text-subtitle1 text-weight-medium")
ui.separator().classes("q-my-sm") ui.separator().classes("q-my-sm")
with ui.column().classes("w-full q-gutter-y-xs"): with ui.column().classes("w-full q-gutter-y-xs"):
_metadata_row("Filename", document.filename) _metadata_row("Filename", document.filename)
_metadata_row("File path", document.file_path) _metadata_row("File path", document.file_path)
ui.separator().classes("q-my-md")
render_document_panzoom(document=document)
def _render_transcript_section(transcript: Transcript | None) -> None:
def _render_transcript_section(transcripts: Sequence[Transcript]) -> None:
with ui.card().classes("w-full q-pa-md"): with ui.card().classes("w-full q-pa-md"):
ui.label("Transcript").classes("text-subtitle1 text-weight-medium") ui.label("Transcripts").classes("text-subtitle1 text-weight-medium")
ui.separator().classes("q-my-sm") ui.separator().classes("q-my-sm")
if transcript is None: if not transcripts:
ui.label("Transcript not available yet.").classes("text-body2 text-grey-8") ui.label("Transcript history is not available yet.").classes("text-body2 text-grey-3")
return return
for transcript in transcripts:
with ui.column().classes("w-full q-gutter-y-xs"): with ui.column().classes("w-full q-gutter-y-xs"):
_metadata_row("Revision", str(transcript.revision))
_metadata_row("Provider", transcript.provider) _metadata_row("Provider", transcript.provider)
_metadata_row("Prompt", transcript.prompt_name) _metadata_row("Prompt", transcript.prompt_name)
_metadata_row("Created", transcript.created_at.isoformat()) _metadata_row("Created", transcript.created_at.isoformat())
if transcript.text: if transcript.text:
ui.separator().classes("q-my-sm") ui.separator().classes("q-my-sm")
with ui.card().classes("w-full bg-grey-1 q-pa-sm"): with ui.card().classes("w-fullq-pa-sm"):
ui.markdown(transcript.text) ui.markdown(transcript.text).classes("text-grey-1")
return elif transcript.error_detail:
if transcript.error_detail:
ui.separator().classes("q-my-sm") ui.separator().classes("q-my-sm")
with ui.card().classes("w-full bg-red-1 text-red-10 q-pa-sm"): with ui.card().classes("w-full bg-red-1 text-red-10 q-pa-sm"):
ui.label("Failure detail").classes("text-caption text-uppercase") ui.label("Failure detail").classes("text-caption text-uppercase")
ui.label(transcript.error_detail).classes("text-body2") ui.label(transcript.error_detail).classes("text-body2")
ui.separator().classes("q-my-md bg-blue-grey-7")
def render_job_detail(*, job: Job, document: Document | None, transcript: Transcript | None) -> None:
def render_job_detail(*, job: Job, document: Document | None, transcripts: Sequence[Transcript]) -> None:
"""Render all sections for the job detail page.""" """Render all sections for the job detail page."""
logger.debug("Rendering job detail for job ID %s with %d transcripts", job.id, len(transcripts))
status_text = job.status.value status_text = job.status.value
with ui.column().classes("w-full max-w-4xl q-gutter-md"): with ui.column().classes("w-full max-w-4xl q-gutter-md"):
with ui.card().classes("w-full q-pa-lg"): with ui.card().classes("w-full q-pa-lg"):
with ui.row().classes("w-full items-center justify-between q-gutter-md"): with ui.row().classes("w-full items-center justify-between q-gutter-md"):
with ui.column().classes("q-gutter-none"): with ui.column().classes("q-gutter-none"):
ui.label("Job overview").classes("text-h6 text-weight-bold") ui.label("Job overview").classes("text-h6 text-weight-bold")
ui.label(str(job.id)).classes("text-caption text-grey-7") ui.label(str(job.id)).classes("text-caption text-grey-5")
status_chip_classes = ( status_chip_classes = (
"q-px-sm q-py-xs rounded-borders " "q-px-sm q-py-xs rounded-borders "
"text-weight-medium text-capitalize " "text-weight-medium text-capitalize "
@@ -79,7 +91,7 @@ def render_job_detail(*, job: Job, document: Document | None, transcript: Transc
) )
ui.label(status_text).classes(status_chip_classes) ui.label(status_text).classes(status_chip_classes)
ui.separator().classes("q-my-md") ui.separator().classes("q-my-md bg-blue-grey-7")
with ui.column().classes("w-full q-gutter-y-xs"): with ui.column().classes("w-full q-gutter-y-xs"):
_metadata_row("Created", job.created_at.isoformat()) _metadata_row("Created", job.created_at.isoformat())
_metadata_row("Updated", job.updated_at.isoformat()) _metadata_row("Updated", job.updated_at.isoformat())
@@ -88,4 +100,4 @@ def render_job_detail(*, job: Job, document: Document | None, transcript: Transc
if document is not None: if document is not None:
_render_document_section(document) _render_document_section(document)
_render_transcript_section(transcript) _render_transcript_section(transcripts)
@@ -1,11 +1,14 @@
"""Common logic for generating table widgets.""" """Common logic for generating table widgets."""
import logging
from collections.abc import Callable from collections.abc import Callable
from typing import Any from typing import Any
from nicegui import events from nicegui import events
from nicegui import ui from nicegui import ui
logger = logging.getLogger(__name__)
def _extract_row_id(args: Any) -> str | None: def _extract_row_id(args: Any) -> str | None:
if isinstance(args, dict): if isinstance(args, dict):
@@ -37,6 +40,7 @@ def _bind_row_click_handler(
on_row_click_id(row_id) on_row_click_id(row_id)
table.on("rowClick", handle_row_click) table.on("rowClick", handle_row_click)
logger.debug("Row click handler bound to table")
def build_table( def build_table(
@@ -63,6 +67,7 @@ def build_table(
.classes(classes) .classes(classes)
.props('table-style="table-layout: fixed; width: 100%;"') .props('table-style="table-layout: fixed; width: 100%;"')
) )
logger.info("Table built with %d rows and %d columns", len(rows), len(columns))
if on_row_click_id is not None: if on_row_click_id is not None:
_bind_row_click_handler(table, on_row_click_id=on_row_click_id) _bind_row_click_handler(table, on_row_click_id=on_row_click_id)
return table return table
@@ -0,0 +1,86 @@
"""Reusable transcript UI components."""
from __future__ import annotations
from collections.abc import Awaitable
from collections.abc import Callable
from datetime import datetime
from typing import Any
from nicegui import ui
from transcription.models import Transcript
type TranscriptAction = Callable[[Transcript], Awaitable[None] | None]
def render_transcript_revision_row(
*,
transcript: Transcript,
initially_expanded: bool = False,
classes: str = "w-full",
on_delete: TranscriptAction | None = None,
) -> Any:
"""Render one collapsible row for a single transcript revision."""
status_label = "Failed" if transcript.error_detail else "Transcribed"
header = f"Revision {transcript.revision} | {status_label}"
caption = f"{transcript.provider} | {transcript.model} | {_format_created_at(transcript.created_at)}"
expansion = ui.expansion(value=initially_expanded, group="group").classes(
f"{classes} rounded-borders bg-blue-grey-10"
)
with expansion, ui.column().classes("w-full q-gutter-y-sm q-pa-sm"):
with expansion.add_slot("header"), ui.row().classes("w-full items-start justify-between q-gutter-md"):
with ui.column().classes("q-gutter-none"):
ui.label(header).classes("text-subtitle1 text-weight-medium")
ui.label(caption).classes("text-caption text-grey-5")
if on_delete is not None:
with ui.dialog() as delete_dialog, ui.card().classes("q-pa-md"):
ui.label("Delete this transcript revision?").classes("text-body1")
with ui.row().classes("w-full justify-end q-gutter-sm"):
ui.button("Cancel", on_click=lambda: delete_dialog.submit(False)).props("flat")
ui.button("Delete", on_click=lambda: delete_dialog.submit(True)).props(
'unelevated color="negative"'
)
async def delete_current_transcript() -> None:
delete_dialog.open()
confirmed = await delete_dialog
if not confirmed:
return
maybe_awaitable = on_delete(transcript)
if isinstance(maybe_awaitable, Awaitable):
await maybe_awaitable
with ui.column(align_items="center").classes("self-center q-gutter-none"):
ui.button(icon="delete", on_click=delete_current_transcript).props(
'flat round dense color="negative"'
)
_metadata_row(label="Provider", value=transcript.provider)
_metadata_row(label="Model", value=transcript.model)
_metadata_row(label="Created", value=_format_created_at(transcript.created_at))
if transcript.text:
with ui.card().classes("w-full q-pa-sm"):
ui.markdown(transcript.text)
if transcript.error_detail:
with ui.card().classes("w-full bg-red-1 text-red-10 q-pa-sm"):
ui.label("Failure detail").classes("text-caption text-uppercase")
ui.label(transcript.error_detail).classes("text-body2")
return expansion
def _format_created_at(value: datetime) -> str:
"""Return a compact UTC-like timestamp for row captions."""
return value.strftime("%Y-%m-%d %H:%M:%S %Z")
def _metadata_row(*, label: str, value: str) -> None:
with ui.row().classes("w-md items-start justify-between q-gutter-x-md"):
ui.label(label).classes("text-caption text-grey-5 text-uppercase")
ui.label(value).classes("text-body2 text-right text-grey-1 break-all")
+68 -70
View File
@@ -4,33 +4,34 @@ from __future__ import annotations
from uuid import UUID from uuid import UUID
from fastapi import Request
from nicegui import ui from nicegui import ui
from sqlalchemy.orm import selectinload
from sqlmodel import desc
from sqlmodel import select
from transcription.db import get_session from transcription.app_state import resolve_session_factory
from transcription.models import Document from transcription.models import JobStatus
from transcription.models import Job from transcription.services.jobs import JobService
from transcription.models import Transcript from transcription.services.transcription import TranscriptionService
from transcription.ui.components.app_shell import render_navigation_header
from transcription.ui.components.error_presenter import show_error from transcription.ui.components.error_presenter import show_error
from transcription.ui.components.error_presenter import summarize_error
from transcription.ui.components.job_detail import render_job_detail
from transcription.ui.components.table.jobs import JobTableRow
from transcription.ui.components.table.jobs import render_jobs_table from transcription.ui.components.table.jobs import render_jobs_table
from ..components.document_panzoom import render_document_panzoom
from ..components.table.jobs import JobTableRow
from ..components.transcript import render_transcript_revision_row
async def fetch_job_rows() -> list[JobTableRow]:
"""Return jobs for display in most-recent-first order.""" def register_page() -> None:
async with get_session() as session: """Register jobs list and detail routes."""
jobs = (
await session.exec( @ui.page("/jobs")
select(Job) async def jobs_page(request: Request) -> None:
.options(selectinload(Job.document)) # pyright: ignore[reportArgumentType] session_factory = resolve_session_factory(request.app.state)
.order_by(desc(Job.created_at)) jobs_service = JobService(session_factory=session_factory)
) render_navigation_header(current_path="/jobs")
).all()
return [ @ui.refreshable
async def render_table() -> None:
jobs = [
JobTableRow( JobTableRow(
id=job.id, id=job.id,
status=job.status.value, status=job.status.value,
@@ -39,63 +40,60 @@ async def fetch_job_rows() -> list[JobTableRow]:
created_at=job.created_at.isoformat(), created_at=job.created_at.isoformat(),
updated_at=job.updated_at.isoformat(), updated_at=job.updated_at.isoformat(),
) )
for job in jobs for job in await jobs_service.list_jobs()
] ]
async def fetch_job_detail(job_id: UUID) -> tuple[Job | None, Document | None, Transcript | None]:
"""Return job, document, and transcript for detail view."""
async with get_session() as session:
job = await session.get(Job, job_id)
if job is None:
return None, None, None
document = await session.get(Document, job.document_id)
transcript = (await session.exec(select(Transcript).where(Transcript.job_id == job.id))).first()
return job, document, transcript
def register_page() -> None:
"""Register jobs list and detail routes."""
@ui.page("/jobs")
async def jobs_page() -> None:
ui.label("Transcription Jobs")
status = ui.label("Ready")
@ui.refreshable
async def render_table() -> None:
jobs = await fetch_job_rows()
render_jobs_table(jobs) render_jobs_table(jobs)
async def refresh() -> None: ui.button("Refresh", on_click=render_table.refresh, icon="refresh")
status.text = "Refreshing..."
try:
await render_table.refresh()
status.text = "Refreshed"
except Exception as exc: # noqa: BLE001
status.text = f"Refresh failed: {summarize_error(exc, operation='jobs.refresh')}"
show_error(exc, title="Jobs refresh failed", operation="jobs.refresh")
ui.button("Refresh", on_click=refresh)
await render_table() await render_table()
ui.link("Back to upload", "/upload")
@ui.page("/jobs/{job_id}") @ui.page("/jobs/{job_id}")
async def job_detail_page(job_id: str) -> None: async def job_detail_page(job_id: str, request: Request) -> None:
ui.label("Job Detail") session_factory = resolve_session_factory(request.app.state)
jobs_service = JobService(session_factory=session_factory)
transcription_service = TranscriptionService(session_factory=session_factory)
render_navigation_header(current_path="/jobs")
job = await jobs_service.read_job(job_id=UUID(job_id))
with ui.splitter(value=30).classes("w-full h-[calc(100vh-64px)]") as splitter:
with splitter.before, ui.column(align_items="stretch").classes("w-full h-full p-4 gap-3"):
render_document_panzoom(document=job.document)
with splitter.after, ui.column(align_items="stretch").classes("w-full h-full p-4 gap-3"):
with ui.row():
ui.button(icon="arrow_back", on_click=ui.navigate.back)
with ui.row().classes("w-full items-center justify-between"):
ui.label(f"{job.id}").classes("text-h6 text-weight-bold")
match job.status:
case JobStatus.TRANSCRIBED:
ui.chip(job.status.value.upper(), color="green", text_color="white").props("outline")
case _:
ui.label(f"{job.status.value}").classes("text-subtitle1 text-weight-medium")
async def delete_transcript_by_id(transcript_id: UUID, revision: int) -> None:
try: try:
parsed_id = UUID(job_id) transcript = await transcription_service.read_transcript(transcript_id=transcript_id)
except ValueError: await transcription_service.delete_transcript(transcript)
ui.label("Invalid job id") except Exception as exc: # noqa: BLE001
ui.link("Back to jobs", "/jobs") show_error(exc, title="Delete failed", operation="jobs.delete_transcript")
return return
job, document, transcript = await fetch_job_detail(parsed_id) ui.notify(f"Deleted revision {revision}", type="positive")
if job is None: await render_transcript_list.refresh()
ui.label("Job not found")
ui.link("Back to jobs", "/jobs")
return
render_job_detail(job=job, document=document, transcript=transcript) @ui.refreshable
async def render_transcript_list() -> None:
refreshed_job = await jobs_service.read_job(job_id=UUID(job_id))
for i, transcript in enumerate(refreshed_job.transcripts):
render_transcript_revision_row(
transcript=transcript,
initially_expanded=(i == 0),
on_delete=(
lambda _transcript, tid=transcript.id, rev=transcript.revision: delete_transcript_by_id(
tid,
rev,
)
),
)
ui.link("Back to jobs", "/jobs") await render_transcript_list()
+2 -3
View File
@@ -8,6 +8,7 @@ from nicegui import ui
from transcription.app_state import resolve_session_factory from transcription.app_state import resolve_session_factory
from transcription.db import get_session from transcription.db import get_session
from transcription.services.store import create_upload_job from transcription.services.store import create_upload_job
from transcription.ui.components.app_shell import render_navigation_header
from transcription.ui.components.upload import render_upload_widget from transcription.ui.components.upload import render_upload_widget
from transcription.worker import resolve_worker_notifier from transcription.worker import resolve_worker_notifier
@@ -17,6 +18,7 @@ def register_page() -> None:
@ui.page("/upload", title="Upload Document") @ui.page("/upload", title="Upload Document")
def upload_page(request: Request) -> None: def upload_page(request: Request) -> None:
render_navigation_header(current_path="/upload")
session_factory = resolve_session_factory(request.app.state) session_factory = resolve_session_factory(request.app.state)
async def submit_upload(filename: str, file_bytes: bytes): async def submit_upload(filename: str, file_bytes: bytes):
@@ -29,6 +31,3 @@ def register_page() -> None:
notify_worker = resolve_worker_notifier(request.app.state) notify_worker = resolve_worker_notifier(request.app.state)
render_upload_widget(submitter=submit_upload, notifier=notify_worker) render_upload_widget(submitter=submit_upload, notifier=notify_worker)
with ui.row():
ui.link("View jobs", "/jobs")
+30
View File
@@ -0,0 +1,30 @@
:root {
/* Soft blue-night palette tokens */
--ctp-rosewater: #f2dde5;
--ctp-flamingo: #edcfd8;
--ctp-pink: #dcc7de;
--ctp-mauve: #a9bde5;
--ctp-red: #d98a9a;
--ctp-maroon: #d39aa5;
--ctp-peach: #d7af8c;
--ctp-yellow: #e2c083;
--ctp-green: #86c8ad;
--ctp-teal: #77bfbe;
--ctp-sky: #7ebdda;
--ctp-sapphire: #74aed0;
--ctp-blue: #92b5f5;
--ctp-lavender: #6f97e8;
--ctp-text: #d8e2f5;
--ctp-subtext1: #bfcae0;
--ctp-subtext0: #a9b6cf;
--ctp-overlay2: #95a3bf;
--ctp-overlay1: #7c8ca9;
--ctp-overlay0: #657490;
--ctp-surface2: #4d5f7c;
--ctp-surface1: #394a65;
--ctp-surface0: #2a3954;
--ctp-base: #1f2b42;
--ctp-mantle: #1a2538;
--ctp-crust: #141e30;
}
+118
View File
@@ -0,0 +1,118 @@
"""Shared fixtures for UI integration tests."""
from __future__ import annotations
import asyncio
from collections.abc import Callable
from pathlib import Path
from uuid import UUID
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlmodel import delete
from transcription.app import create_app
from transcription.config import Settings
from transcription.config import _settings
from transcription.db import create_all
from transcription.db import get_session
from transcription.db import initialize_database_runtime
from transcription.models import Document
from transcription.models import Job
from transcription.models import JobStatus
from transcription.models import Transcript
TranscriptSeed = tuple[int, str | None, str | None]
@pytest.fixture(scope="session")
def app_client(tmp_path_factory: pytest.TempPathFactory) -> tuple[FastAPI, TestClient]:
"""Provide a real application and test client backed by in-memory SQLite."""
tmp_path = tmp_path_factory.mktemp("ui")
settings = Settings(
openrouter_api_key="test-key",
database_url="sqlite:///:memory:",
environment="test",
bootstrap_schema_on_startup=True,
upload_dir=tmp_path / "uploads",
prompt_dir=tmp_path / "prompts",
)
_settings.set(settings)
app = create_app()
app.state.runtime = initialize_database_runtime(settings=settings)
asyncio.run(create_all(engine=app.state.runtime.engine))
with TestClient(app) as client:
yield app, client
@pytest.fixture(autouse=True)
def clear_ui_database(app_client: tuple[FastAPI, TestClient]) -> None:
"""Reset UI-facing tables before each test for isolation."""
app, _ = app_client
async def _clear() -> None:
async with get_session(session_factory=app.state.runtime.session_factory) as session:
await session.exec(delete(Transcript))
await session.exec(delete(Job))
await session.exec(delete(Document))
await session.commit()
asyncio.run(_clear())
@pytest.fixture
def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., UUID]:
"""Return a helper for inserting a document/job/transcript trio."""
app, _ = app_client
fixtures_dir = Path(__file__).resolve().parents[1] / "fixtures" / "images" / "valid"
def _seed(
*,
filename: str = "sample.pdf",
status: JobStatus = JobStatus.TRANSCRIBED,
transcript_text: str | None = "Sample transcript text",
error_detail: str | None = None,
transcript_revisions: list[TranscriptSeed] | None = None,
source_file: Path | None = None,
) -> UUID:
async def _insert() -> UUID:
async with get_session(session_factory=app.state.runtime.session_factory) as session:
stored_path = app.state.settings.upload_dir / filename
stored_path.parent.mkdir(parents=True, exist_ok=True)
source_path = source_file or fixtures_dir / "small_png.png"
stored_path.write_bytes(source_path.read_bytes())
document = Document(filename=filename, file_path=str(stored_path))
session.add(document)
await session.flush()
job = Job(document_id=document.id, status=status, retry_count=0)
session.add(job)
await session.flush()
revisions = transcript_revisions
if revisions is None and (transcript_text is not None or error_detail is not None):
revisions = [(0, transcript_text, error_detail)]
if revisions is not None:
for revision, revision_text, revision_error in revisions:
session.add(
Transcript(
job_id=job.id,
revision=revision,
provider="openrouter",
model="google/gemini-2.5-flash",
prompt_name="transcribe_document",
text=revision_text,
error_detail=revision_error,
)
)
await session.commit()
return job.id
return asyncio.run(_insert())
return _seed
+64 -24
View File
@@ -1,37 +1,77 @@
"""Tests for the jobs page route.""" """Tests for the jobs page route."""
from pathlib import Path
from uuid import uuid4
import pytest import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from transcription.ui import register_pages from transcription.models import JobStatus
from transcription.ui.pages import jobs_page
@pytest.fixture
def client(monkeypatch):
"""Provide a minimal app client with jobs data patched for rendering."""
async def _fetch_jobs_stub():
return []
monkeypatch.setattr(jobs_page, "fetch_jobs", _fetch_jobs_stub)
app = FastAPI()
register_pages(app)
with TestClient(app) as test_client:
yield test_client
@pytest.mark.integration @pytest.mark.integration
class TestPageRendering: class TestPageRendering:
"""Verify the jobs page is available and includes the main controls.""" """Verify jobs routes render correctly with real app wiring."""
def test_jobs_page_renders_expected_controls(self, client): def test_jobs_page_renders_empty_state(self, app_client):
"""GET /ui/jobs returns the page shell and jobs controls.""" """GET /ui/jobs renders the page and empty-state text when no jobs exist."""
_, client = app_client
response = client.get("/ui/jobs") response = client.get("/ui/jobs")
assert response.status_code == 200 assert response.status_code == 200
assert "Transcription Jobs" in response.text assert "Transcription Jobs" in response.text
assert "Refresh" in response.text assert "No jobs yet." in response.text
assert "Back to upload" in response.text
def test_jobs_page_lists_seeded_jobs(self, app_client, seed_job):
"""GET /ui/jobs lists seeded jobs from the in-memory database."""
_, client = app_client
seed_job(filename="sample.pdf", status=JobStatus.TRANSCRIBED, transcript_text="done")
response = client.get("/ui/jobs")
assert response.status_code == 200
assert "sample.pdf" in response.text
assert "transcribed" in response.text
def test_job_detail_page_renders_seeded_job(self, app_client, seed_job):
"""GET /ui/jobs/{job_id} renders detail content for a real seeded job."""
_, client = app_client
fixture_path = Path(__file__).resolve().parents[1] / "fixtures" / "images" / "valid" / "single_page_pdf.pdf"
job_id = seed_job(
filename="detail.pdf",
status=JobStatus.TRANSCRIBED,
transcript_revisions=[
(0, None, "first attempt failed"),
(1, "hello", None),
],
source_file=fixture_path,
)
response = client.get(f"/ui/jobs/{job_id}")
assert response.status_code == 200
assert "Job Detail" in response.text
assert "Job overview" in response.text
assert "detail.pdf" in response.text
assert "Transcripts" in response.text
assert "Revision" in response.text
assert "first attempt failed" in response.text
assert "hello" in response.text
assert "Document preview" in response.text
assert "/uploads/detail.pdf" in response.text
def test_job_detail_page_rejects_invalid_id(self, app_client):
"""GET /ui/jobs/{job_id} shows validation feedback for malformed IDs."""
_, client = app_client
response = client.get("/ui/jobs/not-a-uuid")
assert response.status_code == 200
assert "Invalid job id" in response.text
def test_job_detail_page_handles_missing_job(self, app_client):
"""GET /ui/jobs/{job_id} shows not-found state for unknown IDs."""
_, client = app_client
missing_id = uuid4()
response = client.get(f"/ui/jobs/{missing_id}")
assert response.status_code == 200
assert "Job not found" in response.text
+8 -28
View File
@@ -1,38 +1,18 @@
"""Tests for UI page registration wiring.""" """Tests for UI page registration wiring."""
import pytest import pytest
from fastapi import FastAPI
from transcription.ui import register_pages
@pytest.mark.integration @pytest.mark.integration
class TestPageRegistration: class TestPageRegistration:
"""Verify page registration and route wiring.""" """Verify page registration and mounted UI routes."""
def test_register_pages_wires_upload_jobs_and_mount(self, monkeypatch): def test_ui_mount_serves_registered_pages(self, app_client):
"""register_pages registers pages and mounts NiceGUI at /ui.""" """Mounted UI routes respond successfully when the full app is created."""
calls: list[str] = [] _, client = app_client
def _record_upload() -> None: upload_response = client.get("/ui/upload")
calls.append("upload") jobs_response = client.get("/ui/jobs")
def _record_jobs() -> None: assert upload_response.status_code == 200
calls.append("jobs") assert jobs_response.status_code == 200
def _record_run_with(
_app: FastAPI,
*,
mount_path: str,
show_welcome_message: bool,
) -> None:
calls.append(f"run_with:{mount_path}:{show_welcome_message}")
monkeypatch.setattr("transcription.ui.register_upload_page", _record_upload)
monkeypatch.setattr("transcription.ui.register_jobs_page", _record_jobs)
monkeypatch.setattr("transcription.ui.ui.run_with", _record_run_with)
app = FastAPI()
register_pages(app)
assert calls == ["upload", "jobs", "run_with:/ui:False"]
+10 -30
View File
@@ -1,55 +1,35 @@
"""Tests for the upload page route.""" """Tests for upload and entry-point routes."""
from pathlib import Path
import pytest import pytest
from fastapi.testclient import TestClient
from transcription.app import create_app
from transcription.config import Settings
from transcription.config import _settings
@pytest.fixture
def client(tmp_path: Path):
"""Provide a real app client backed by in-memory SQLite."""
settings = Settings(
openrouter_api_key="test-key",
database_url="sqlite:///:memory:",
environment="test",
upload_dir=tmp_path / "uploads",
prompt_dir=tmp_path / "prompts",
)
_settings.set(settings)
app = create_app()
with TestClient(app) as test_client:
yield test_client
@pytest.mark.integration @pytest.mark.integration
class TestPageRendering: class TestPageRendering:
"""Verify the upload page is available and includes the main controls.""" """Verify upload-related routes return working pages."""
def test_root_redirects_to_ui(self, client): def test_root_redirects_to_ui(self, app_client):
"""GET / redirects to the UI mount point.""" """GET / redirects to the UI mount point."""
_, client = app_client
response = client.get("/", follow_redirects=False) response = client.get("/", follow_redirects=False)
assert response.status_code == 307 assert response.status_code == 307
assert response.headers["location"] == "/ui" assert response.headers["location"] == "/ui"
def test_ui_redirects_to_upload(self, client): def test_ui_redirects_to_upload(self, app_client):
"""GET /ui redirects to the upload page.""" """GET /ui redirects to the upload page."""
_, client = app_client
response = client.get("/ui", follow_redirects=False) response = client.get("/ui", follow_redirects=False)
assert response.status_code == 307 assert response.status_code == 307
assert response.headers["location"] == "/ui/upload" assert response.headers["location"] == "/ui/upload"
def test_upload_page_renders_expected_controls(self, client): def test_upload_page_renders_expected_controls(self, app_client):
"""GET /ui/upload returns the page shell and upload controls.""" """GET /ui/upload returns the page shell and upload controls."""
_, client = app_client
response = client.get("/ui/upload") response = client.get("/ui/upload")
assert response.status_code == 200 assert response.status_code == 200
assert "Upload Document" in response.text assert "Upload Document" in response.text
assert "Select document file" in response.text assert "Select document file" in response.text
assert "View jobs" in response.text assert "Upload" in response.text
assert "Jobs" in response.text