8 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
15 changed files with 446 additions and 297 deletions
+5
View File
@@ -14,3 +14,8 @@ wheels/
# SQLite database
*.db
upload/
*.jpg
*.jpeg
*.png
+11 -6
View File
@@ -51,10 +51,15 @@ def _ensure_sqlite_compat_columns(connection: Connection) -> None:
inspector = inspect(connection)
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 "retry_count" not in columns:
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")
if "job" in table_names:
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"))
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")
+2
View File
@@ -65,6 +65,8 @@ class Transcript(SQLModel, table=True):
"""Revision number for this job's transcript history, starting at 0."""
provider: str
"""Name of the transcription provider used to generate this transcript."""
model: str
"""Model identifier used to generate this transcript revision."""
prompt_name: str
"""Name of the prompt used to generate this transcript."""
text: str | None = None
+3 -1
View File
@@ -28,12 +28,14 @@ class TranscriptionResult:
prompt_name: 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."""
return Transcript(
job_id=job_id,
revision=revision,
provider=self.provider,
prompt_name=self.prompt_name,
model=self.model,
text=self.text,
)
+4 -1
View File
@@ -35,7 +35,10 @@ class JobService(ServiceBase):
async with self._session_scope(session) as _session:
query = (
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)
.execution_options(populate_existing=True)
)
@@ -114,6 +114,7 @@ class TranscriptionService(ServiceBase):
job_id=job_id,
text=result.text,
provider=result.provider,
model=result.model,
prompt_name=result.prompt_name,
session=session,
)
@@ -125,6 +126,7 @@ class TranscriptionService(ServiceBase):
text: str | None,
error_detail: str | None = None,
provider: str | None = None,
model: str | None = None,
prompt_name: str = DEFAULT_PROMPT_FILE,
session: AsyncSession | None = None,
) -> Transcript:
@@ -139,6 +141,7 @@ class TranscriptionService(ServiceBase):
job_id=job_id,
revision=next_revision,
provider=provider or self.settings.provider.value,
model=model or _resolve_transcript_model(provider=self.provider, settings=self.settings),
prompt_name=prompt_name,
text=text,
error_detail=error_detail,
@@ -181,6 +184,17 @@ def _transcript_job_query(job_id: UUID):
) # 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(
image_path: str | Path,
*,
+2
View File
@@ -126,6 +126,7 @@ async def _finalize_transcribed(
text=result.text,
error_detail=None,
provider=result.provider,
model=result.model,
prompt_name=result.prompt_name,
session=local_session,
)
@@ -142,6 +143,7 @@ async def _finalize_transcribed(
text=result.text,
error_detail=None,
provider=result.provider,
model=result.model,
prompt_name=result.prompt_name,
session=session,
)
@@ -16,6 +16,88 @@ PANGOZOOM_CDN_URL = "https://unpkg.com/@panzoom/[email protected]/dist/panzoom.min.j
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
@@ -42,51 +124,6 @@ def _document_kind(document: Document) -> str:
return "image"
@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: center;
justify-content: center;
}
.document-panzoom-media {
width: auto;
height: auto;
display: block;
max-width: none;
max-height: none;
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 _attach_panzoom(host_id: str) -> None:
ui.run_javascript(
f"""
@@ -99,137 +136,76 @@ def _attach_panzoom(host_id: str) -> None:
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 1;
if (media && media.tagName === 'IMG') {{
if (media.naturalWidth <= 0 || media.naturalHeight <= 0) return 1;
return Math.min(
hostRect.width / media.naturalWidth,
hostRect.height / media.naturalHeight
);
}}
if (target.tagName === 'IFRAME') {{
const targetRect = target.getBoundingClientRect();
if (targetRect.width <= 0 || targetRect.height <= 0) return 1;
return hostRect.height / targetRect.height;
}}
if (hostRect.width <= 0 || hostRect.height <= 0) return null;
return 1;
}};
const initPanzoom = () => {{
if (window.__transcriptionPanzoom[{host_id!r}]) {{
window.__transcriptionPanzoom[{host_id!r}].destroy();
}}
const buildInstance = () => {{
cleanup();
if (host.__transcriptionWheelHandler) {{
host.removeEventListener('wheel', host.__transcriptionWheelHandler);
host.__transcriptionWheelHandler = null;
}}
const fitScale = computeFitScale();
if (fitScale === null) return false;
let fitScale = computeFitScale();
if (!Number.isFinite(fitScale) || fitScale <= 0) {{
fitScale = 1;
}}
const startScale = fitScale;
const minScale = 0.01;
const minScale = Math.min(fitScale, 0.01);
const instance = Panzoom(target, {{
maxScale: 32,
startX: 0,
startY: 0,
startScale: fitScale,
minScale: minScale,
startScale: startScale,
step: 0.18,
contain: 'inside',
roundPixels: true,
maxScale: 256,
step: 0.2,
roundPixels: false,
panOnlyWhenZoomed: true,
overflow: 'hidden',
}});
window.__transcriptionPanzoom[{host_id!r}] = instance;
const wheelHandler = (event) => instance.zoomWithWheel(event);
host.addEventListener('wheel', wheelHandler, {{ passive: false }});
requestAnimationFrame(() => {{
instance.reset({{ animate: false, force: true }});
instance.setOptions({{ contain: undefined }});
instance.reset({{ animate: false }});
}});
const wheelHandler = (event) => instance.zoomWithWheel(event);
host.__transcriptionWheelHandler = wheelHandler;
host.addEventListener('wheel', wheelHandler, {{ passive: 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', initPanzoom, {{ once: true }});
media.addEventListener('load', () => initWhenReady(), {{ once: true }});
return;
}}
if ((media && media.tagName === 'IMG' && (host.clientWidth <= 0 || host.clientHeight <= 0)) ||
(target.tagName === 'IFRAME' && (host.clientWidth <= 0 || host.clientHeight <= 0))) {{
requestAnimationFrame(initPanzoom);
return;
}}
initPanzoom();
initWhenReady();
}})();
"""
)
def render_document_panzoom(*, document: Document, height: str = "640px") -> 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 bg-blue-grey-10 text-grey-1 q-pa-md"):
with ui.row().classes("w-full items-center justify-between q-gutter-sm"):
with ui.column().classes("q-gutter-none"):
ui.label("Document preview").classes("text-subtitle1 text-weight-medium")
ui.label(document.filename).classes("text-caption text-grey-4")
with ui.button_group().props("flat outline"):
ui.button(
"Zoom In",
icon="add",
on_click=lambda: ui.run_javascript(f"window.__transcriptionPanzoom[{host_id!r}]?.zoomIn()"),
)
ui.button(
"Zoom Out",
icon="remove",
on_click=lambda: ui.run_javascript(f"window.__transcriptionPanzoom[{host_id!r}]?.zoomOut()"),
)
ui.button(
"Reset",
icon="restart_alt",
on_click=lambda: ui.run_javascript(f"window.__transcriptionPanzoom[{host_id!r}]?.reset()"),
)
ui.label("Scroll, pinch, or use the buttons to inspect the document.").classes(
"text-caption text-grey-4 q-mt-sm"
)
with (
ui.element("div")
.classes("w-full document-panzoom-host bg-blue-grey-9 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)
+13 -9
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from collections.abc import Sequence
from nicegui import ui
@@ -11,6 +12,8 @@ from transcription.models import Job
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:
if status == "queued":
@@ -31,21 +34,21 @@ def _metadata_row(label: str, value: str) -> None:
def _render_document_section(document: Document) -> None:
with ui.card().classes("w-full bg-blue-grey-10 text-grey-1 q-pa-md"):
with ui.card().classes("w-full q-pa-md"):
ui.label("Document").classes("text-subtitle1 text-weight-medium")
ui.separator().classes("q-my-sm bg-blue-grey-7")
ui.separator().classes("q-my-sm")
with ui.column().classes("w-full q-gutter-y-xs"):
_metadata_row("Filename", document.filename)
_metadata_row("File path", document.file_path)
ui.separator().classes("q-my-md bg-blue-grey-7")
ui.separator().classes("q-my-md")
render_document_panzoom(document=document)
def _render_transcript_section(transcripts: Sequence[Transcript]) -> None:
with ui.card().classes("w-full bg-blue-grey-10 text-grey-1 q-pa-md"):
with ui.card().classes("w-full q-pa-md"):
ui.label("Transcripts").classes("text-subtitle1 text-weight-medium")
ui.separator().classes("q-my-sm bg-blue-grey-7")
ui.separator().classes("q-my-sm")
if not transcripts:
ui.label("Transcript history is not available yet.").classes("text-body2 text-grey-3")
@@ -59,11 +62,11 @@ def _render_transcript_section(transcripts: Sequence[Transcript]) -> None:
_metadata_row("Created", transcript.created_at.isoformat())
if transcript.text:
ui.separator().classes("q-my-sm bg-blue-grey-7")
with ui.card().classes("w-full bg-blue-grey-9 text-grey-1 q-pa-sm"):
ui.separator().classes("q-my-sm")
with ui.card().classes("w-fullq-pa-sm"):
ui.markdown(transcript.text).classes("text-grey-1")
elif transcript.error_detail:
ui.separator().classes("q-my-sm bg-blue-grey-7")
ui.separator().classes("q-my-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(transcript.error_detail).classes("text-body2")
@@ -73,9 +76,10 @@ def _render_transcript_section(transcripts: Sequence[Transcript]) -> None:
def render_job_detail(*, job: Job, document: Document | None, transcripts: Sequence[Transcript]) -> None:
"""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
with ui.column().classes("w-full max-w-4xl q-gutter-md"):
with ui.card().classes("w-full bg-blue-grey-10 text-grey-1 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.column().classes("q-gutter-none"):
ui.label("Job overview").classes("text-h6 text-weight-bold")
@@ -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")
+45 -17
View File
@@ -8,13 +8,16 @@ from fastapi import Request
from nicegui import ui
from transcription.app_state import resolve_session_factory
from transcription.models import JobStatus
from transcription.services.jobs import JobService
from transcription.services.transcription import TranscriptionService
from transcription.ui.components.app_shell import render_navigation_header
from transcription.ui.components.job_detail import render_job_detail
from transcription.ui.components.error_presenter import show_error
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
def register_page() -> None:
@@ -50,22 +53,47 @@ def register_page() -> None:
jobs_service = JobService(session_factory=session_factory)
transcription_service = TranscriptionService(session_factory=session_factory)
render_navigation_header(current_path="/jobs")
ui.button(icon="arrow_back", on_click=ui.navigate.back)
try:
parsed_id = UUID(job_id)
except ValueError:
ui.label("Invalid job id")
ui.link("Back to jobs", "/jobs")
return
try:
job = await jobs_service.read_job(job_id=parsed_id)
except ValueError:
ui.label("Job not found")
ui.link("Back to jobs", "/jobs")
return
job = await jobs_service.read_job(job_id=UUID(job_id))
transcripts = list(await transcription_service.list_transcripts_by_job(parsed_id))
render_job_detail(job=job, document=job.document, transcripts=transcripts)
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")
ui.link("Back to jobs", "/jobs")
async def delete_transcript_by_id(transcript_id: UUID, revision: int) -> None:
try:
transcript = await transcription_service.read_transcript(transcript_id=transcript_id)
await transcription_service.delete_transcript(transcript)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Delete failed", operation="jobs.delete_transcript")
return
ui.notify(f"Deleted revision {revision}", type="positive")
await render_transcript_list.refresh()
@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,
)
),
)
await render_transcript_list()
+51 -12
View File
@@ -10,38 +10,63 @@ 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
def app_client(tmp_path: Path) -> tuple[FastAPI, TestClient]:
@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(
*,
@@ -49,10 +74,17 @@ def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., UUID]:
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:
document = Document(filename=filename, file_path=f"uploads/{filename}")
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()
@@ -60,16 +92,23 @@ def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., UUID]:
session.add(job)
await session.flush()
if transcript_text is not None or error_detail is not None:
session.add(
Transcript(
job_id=job.id,
provider="openrouter",
prompt_name="transcribe_document",
text=transcript_text,
error_detail=error_detail,
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
+63 -39
View File
@@ -1,53 +1,77 @@
"""Tests for the jobs page route."""
from uuid import UUID
from pathlib import Path
from uuid import uuid4
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from transcription.ui import register_pages
from transcription.ui.pages import jobs_page
from transcription.ui.pages.jobs_page import JobTableRow
@pytest.fixture
def client(monkeypatch):
"""Provide a minimal app client with jobs data patched for rendering."""
async def _fetch_jobs_stub():
return [
JobTableRow(
id=UUID("00000000-0000-0000-0000-000000000001"),
status="queued",
filename="sample.pdf",
retry_count=2,
created_at="2026-01-01T12:00:00+00:00",
updated_at="2026-01-01T12:01:00+00:00",
)
]
monkeypatch.setattr(jobs_page, "fetch_jobs", _fetch_jobs_stub)
app = FastAPI()
register_pages(app)
with TestClient(app) as test_client:
yield test_client
from transcription.models import JobStatus
@pytest.mark.integration
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, monkeypatch):
"""GET /ui/jobs returns the page shell and jobs controls."""
response = client.get("/ui/jobs")
async def _fetch_jobs_empty():
return []
monkeypatch.setattr(jobs_page, "fetch_jobs", _fetch_jobs_empty)
def test_jobs_page_renders_empty_state(self, app_client):
"""GET /ui/jobs renders the page and empty-state text when no jobs exist."""
_, client = app_client
response = client.get("/ui/jobs")
assert response.status_code == 200
assert "Transcription Jobs" in response.text
assert "No jobs yet." 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 -29
View File
@@ -1,39 +1,18 @@
"""Tests for UI page registration wiring."""
import pytest
from fastapi import FastAPI
from transcription.ui import register_pages
@pytest.mark.integration
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):
"""register_pages registers pages and mounts NiceGUI at /ui."""
calls: list[str] = []
def test_ui_mount_serves_registered_pages(self, app_client):
"""Mounted UI routes respond successfully when the full app is created."""
_, client = app_client
def _record_upload() -> None:
calls.append("upload")
upload_response = client.get("/ui/upload")
jobs_response = client.get("/ui/jobs")
def _record_jobs() -> None:
calls.append("jobs")
def _record_run_with(
_app: FastAPI,
*,
mount_path: str,
show_welcome_message: bool,
dark: bool,
) -> None:
calls.append(f"run_with:{mount_path}:{show_welcome_message}:{dark}")
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:True"]
assert upload_response.status_code == 200
assert jobs_response.status_code == 200
+9 -29
View File
@@ -1,55 +1,35 @@
"""Tests for the upload page route."""
from pathlib import Path
"""Tests for upload and entry-point routes."""
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
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."""
_, client = app_client
response = client.get("/", follow_redirects=False)
assert response.status_code == 307
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."""
_, client = app_client
response = client.get("/ui", follow_redirects=False)
assert response.status_code == 307
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."""
_, client = app_client
response = client.get("/ui/upload")
assert response.status_code == 200
assert "Upload Document" in response.text
assert "Select document file" in response.text
assert "Upload" in response.text
assert "Jobs" in response.text