bbchops/transcription
1
0
Fork 0
generated from john/python-template
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Jobs: jobs still stuck in queue. Fixes from testing.

Browse Source
This commit is contained in:
Jim Lancaster
2026-08-04 18:09:31 -05:00
parent 6c6589d8ff
commit 271633d1d5
12 changed files with 278 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+53 -7
View File
@@ -1,8 +1,54 @@
PROVIDER=openrouter
OPENROUTER_API_KEY=sk-or-...
# PROVIDER_MODEL= # optional: OpenRouter adapter supplies default
# --- NiceGUI Server ---
# HOST=`0.0.0.0` (default)
# PORT=8000 (default)
# LOG_LEVEL: [`critical`, `error`, `warning`, `info` (default), `debug`, `trace`]
# RELOAD=false (default)
# --- AI provider ---
# PROVIDER=[`openrouter`(default), `google_genai`]
PROVIDER=openrouter
# OPENROUTER_API_KEY - Required when `PROVIDER=openrouter`
OPENROUTER_API_KEY=your-api-key-goes-here
# GEMINI_API_KEY - Required when `PROVIDER=google_genai`
# PROVIDER_MODEL= specify model. If left blank OpenRouter will supply default.
PROVIDER_MODEL=google/gemini-2.5-flash
# OPENROUTER_HTTP_REFERER=https://example.com
# OPENROUTER_APP_TITLE=Historical Transcription MVP
# DATABASE_URL=sqlite:///./transcription.db
# UPLOAD_DIR=./uploads
# PROMPT_DIR=./prompts
# OPENROUTER_APP_TITLE="Google: Gemini 2.5 Flash (openrouter)"
# --- runtime environment ---
# ENVIRONMENT: [`development`(default), `test`, `production`]
# --- persistence ---
# Use nested settings with double underscore because env_nested_delimiter="__".
# SQLite example:
# DATABASE__DRIVER=sqlite
# DATABASE__PATH=app.db
#
# SQLite with custom relative path:
# DATABASE__DRIVER=sqlite
DATABASE__PATH=./data/transcription.db
#
# Postgres example:
# DATABASE__DRIVER=postgres
# DATABASE__HOST=localhost
# DATABASE__PORT=5432
# DATABASE__DATABASE=transcription
# DATABASE__USER=postgres
# DATABASE__PASSWORD=change-me
#
# Optional persistence flags:
# BOOTSTRAP_SCHEMA_ON_STARTUP=false
# SQLITE_CHECK_SAME_THREAD=false
# --- filesystem paths ---
UPLOAD_DIR="./data"
PROMPT_DIR="./prompts"
# --- worker reliability ---
WORKER_MAX_RETRIES=0
WORKER_RETRY_BACKOFF_SECONDS=0
# WORKER_PROVIDER_TIMEOUT_SECONDS=[0-20]
WORKER_PROVIDER_TIMEOUT_SECONDS=20
WORKER_MIN_TRANSCRIPTION_CHARS=0
WORKER_MIN_TRANSCRIPTION_LINES=0
WORKER_FAIL_ON_FINISH_REASON_LENGTH=false
.gitignore
+1
View File
@@ -17,3 +17,4 @@ wheels/
# Document images
uploads/*
data/*
src/transcription/services/documents.py
+2 -6
View File
@@ -217,12 +217,8 @@ class DocumentService(ServiceBase):
suggestion="Verify the person id and retry.",
)
if existing.document_people:
raise PersonDeleteBlockedError(
"Person delete blocked by linked documents",
category=ErrorCategory.VALIDATION,
suggestion="Remove linked DocumentPerson records first, then retry deletion.",
)
for link in list(existing.document_people):
await _session.delete(link)
await _session.delete(existing)
await self._finalize(session=_session, caller_session=session)
src/transcription/services/store.py
+64 -19
View File
@@ -41,6 +41,15 @@ class JobCreateResult:
source_ids: tuple[UUID, ...]
@dataclass(frozen=True)
class PendingStoredUpload:
"""Pre-staged upload artifact tied to a source id."""
source_id: UUID
original_filename: str
stored_path: Path
async def create_upload_job(
*,
filename: str,
@@ -50,14 +59,20 @@ async def create_upload_job(
) -> UploadJobResult:
"""Create upload-backed document and queued job records."""
runtime_settings = settings or get_settings()
document_id = uuid4()
source_id = uuid4()
stored_path = store_file(
filename=filename,
file_bytes=file_bytes,
settings=runtime_settings,
relative_directory=Path("documents") / str(document_id),
filename_stem=str(source_id),
)
try:
document, job = await _create_upload_records(
session=session,
document_id=document_id,
source_id=source_id,
original_filename=filename,
stored_path=stored_path,
)
@@ -99,15 +114,19 @@ async def create_job_for_document(
runtime_settings = settings or get_settings()
sorted_uploads = sorted(uploads, key=lambda item: Path(item[0]).name.casefold())
stored_uploads: list[tuple[str, Path]] = []
stored_uploads: list[PendingStoredUpload] = []
for filename, file_bytes in sorted_uploads:
source_id = uuid4()
stored_uploads.append(
(
filename,
store_file(
PendingStoredUpload(
source_id=source_id,
original_filename=filename,
stored_path=store_file(
filename=filename,
file_bytes=file_bytes,
settings=runtime_settings,
relative_directory=Path("documents") / str(document_id),
filename_stem=str(source_id),
),
)
)
@@ -122,8 +141,8 @@ async def create_job_for_document(
prompt_name=prompt_name,
)
except Exception as exc:
for _, stored_path in stored_uploads:
_best_effort_delete(stored_path)
for upload in stored_uploads:
_best_effort_delete(upload.stored_path)
raise UploadError(
"Failed to create job records from uploads",
category=ErrorCategory.INFRA_TRANSIENT,
@@ -142,10 +161,13 @@ async def create_job_for_document(
async def _create_upload_records(
*,
session: AsyncSession,
document_id: UUID,
source_id: UUID,
original_filename: str,
stored_path: Path,
) -> tuple[Document, Job]:
document = Document(
id=document_id,
name=Path(original_filename).name,
)
session.add(document)
@@ -156,6 +178,7 @@ async def _create_upload_records(
await session.flush()
source = Source(
id=source_id,
document_id=document.id,
page_number=1,
upload_name=Path(original_filename).name,
@@ -183,7 +206,7 @@ async def _create_job_for_document_records(
*,
session: AsyncSession,
document_id: UUID,
stored_uploads: Sequence[tuple[str, Path]],
stored_uploads: Sequence[PendingStoredUpload],
provider: str | None,
model: str | None,
prompt_name: str | None,
@@ -211,13 +234,14 @@ async def _create_job_for_document_records(
await session.flush()
source_ids: list[UUID] = []
for page_offset, (original_filename, stored_path) in enumerate(stored_uploads):
for page_offset, upload in enumerate(stored_uploads):
source = Source(
id=upload.source_id,
document_id=document_id,
page_number=next_page_number + page_offset,
upload_name=Path(original_filename).name,
filename=stored_path.name,
file_path=str(stored_path),
upload_name=Path(upload.original_filename).name,
filename=upload.stored_path.name,
file_path=str(upload.stored_path),
)
session.add(source)
await session.flush()
@@ -244,22 +268,41 @@ def _best_effort_delete(path: Path) -> None:
logger.warning("Failed to clean up upload file after DB error: %s", path)
def store_file(*, filename: str, file_bytes: bytes, settings: Settings | None = None) -> Path:
def store_file(
*,
filename: str,
file_bytes: bytes,
settings: Settings | None = None,
relative_directory: Path | None = None,
filename_stem: str | None = None,
) -> Path:
"""Persist an uploaded file to the configured upload directory."""
runtime_settings = settings or get_settings()
_validate_upload(filename=filename, file_bytes=file_bytes, supported_extensions=SUPPORTED_UPLOAD_EXTENSIONS)
return _store_file_bytes(filename=filename, file_bytes=file_bytes, settings=runtime_settings)
return _store_file_bytes(
filename=filename,
file_bytes=file_bytes,
settings=runtime_settings,
relative_directory=relative_directory,
filename_stem=filename_stem,
)
def store_person_portrait(*, filename: str, file_bytes: bytes, settings: Settings | None = None) -> Path:
"""Persist a portrait upload under uploads/portraits/person."""
def store_person_portrait(
*,
person_id: UUID,
filename: str,
file_bytes: bytes,
settings: Settings | None = None,
) -> Path:
"""Persist a portrait upload under persons/<person_id>."""
runtime_settings = settings or get_settings()
_validate_upload(filename=filename, file_bytes=file_bytes, supported_extensions=SUPPORTED_PORTRAIT_EXTENSIONS)
return _store_file_bytes(
filename=filename,
file_bytes=file_bytes,
settings=runtime_settings,
relative_directory=Path("portraits") / "person",
relative_directory=Path("persons") / str(person_id),
)
@@ -269,12 +312,13 @@ def _store_file_bytes(
file_bytes: bytes,
settings: Settings,
relative_directory: Path | None = None,
filename_stem: str | None = None,
) -> Path:
upload_dir = settings.upload_dir
target_dir = upload_dir if relative_directory is None else upload_dir / relative_directory
target_dir.mkdir(parents=True, exist_ok=True)
stored_name = _build_stored_filename(filename)
stored_name = _build_stored_filename(filename=filename, filename_stem=filename_stem)
stored_path = target_dir / stored_name
try:
@@ -315,7 +359,8 @@ def _validate_upload(*, filename: str, file_bytes: bytes, supported_extensions:
)
def _build_stored_filename(filename: str) -> str:
def _build_stored_filename(*, filename: str, filename_stem: str | None = None) -> str:
safe_name = Path(filename).name
suffix = Path(safe_name).suffix.lower()
return f"{uuid4()}{suffix}"
stem = filename_stem or str(uuid4())
return f"{stem}{suffix}"
src/transcription/ui/pages/people_page.py
+20 -22
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
from datetime import date
from urllib.parse import quote
from uuid import UUID
from uuid import uuid4
from fastapi import Request
from nicegui import ui
@@ -15,7 +16,6 @@ from transcription.errors import ErrorCategory
from transcription.services.documents import (
DocumentError,
DocumentService,
PersonDeleteBlockedError,
)
from transcription.services.store import UploadError, store_person_portrait
from transcription.ui.components.app_shell import render_navigation_header
@@ -44,11 +44,12 @@ def _parse_optional_date(value: str | None, *, label: str) -> date | None:
raise ValueError(f"{label} must use YYYY-MM-DD.") from exc
def _bind_portrait_file_picker(portrait_path_input: ui.input, *, settings: Settings) -> None:
def _bind_portrait_file_picker(portrait_path_input: ui.input, *, settings: Settings, person_id: UUID) -> None:
async def on_portrait_selected(event) -> None:
payload = await event.file.read()
try:
stored_path = store_person_portrait(
person_id=person_id,
filename=event.file.name,
file_bytes=payload,
settings=settings,
@@ -73,7 +74,8 @@ def _bind_portrait_file_picker(portrait_path_input: ui.input, *, settings: Setti
auto_upload=True,
label="Choose portrait file",
).props('accept=".jpg,.jpeg,.png,.gif,.webp,.bmp,.tif,.tiff"').classes("w-full")
ui.label("Portraits are stored under uploads/portraits/person.").classes("text-xs ui-text-muted")
portrait_dir = settings.upload_dir / "persons" / str(person_id)
ui.label(f"Portraits are stored under {portrait_dir}.").classes("text-xs ui-text-muted")
def _resolve_portrait_src(path: str | None) -> str | None:
@@ -145,6 +147,7 @@ def register_page() -> None: # noqa: PLR0915
apply_archival_theme()
people_service = DocumentService(session_factory=session_factory)
render_navigation_header(current_path="/people")
draft_person_id = uuid4()
with ui.column().classes("w-full max-w-4xl mx-auto p-4 gap-4"):
page_header("Create Person Record", subtitle="Full name is required.")
@@ -167,7 +170,11 @@ def register_page() -> None: # noqa: PLR0915
biography_input = ui.textarea(label="Biography").props("outlined bg-white autogrow").classes("w-full")
portrait_path_input = ui.input(label="Portrait path").props("outlined bg-white").classes("w-full")
_bind_portrait_file_picker(portrait_path_input, settings=_resolve_runtime_settings(request))
_bind_portrait_file_picker(
portrait_path_input,
settings=_resolve_runtime_settings(request),
person_id=draft_person_id,
)
async def submit_create() -> None:
full_name = (full_name_input.value or "").strip()
@@ -183,6 +190,7 @@ def register_page() -> None: # noqa: PLR0915
return
candidate = Person(
id=draft_person_id,
full_name=full_name,
display_name=(display_name_input.value or "").strip() or None,
maiden_name=(maiden_name_input.value or "").strip() or None,
@@ -357,7 +365,11 @@ def register_page() -> None: # noqa: PLR0915
portrait_path_input = (
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))
_bind_portrait_file_picker(
portrait_path_input,
settings=_resolve_runtime_settings(request),
person_id=person.id,
)
async def submit_edit() -> None:
full_name = (full_name_input.value or "").strip()
@@ -431,29 +443,15 @@ 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(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")
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"
)
return
ui.label(
f"This will also remove {len(person.document_people)} linked document relationship(s)."
).classes("text-xs text-red-800 font-bold mt-2")
ui.label("This action permanently deletes the person record.").classes("text-xs text-red-800 font-medium")
async def submit_delete() -> None:
try:
await people_service.delete_person(person)
except PersonDeleteBlockedError as exc:
ui.notify(exc.message, type="warning")
ui.navigate.to(f"/people/{person.id}/delete")
return
except DocumentError as exc:
if exc.category == ErrorCategory.NOT_FOUND:
ui.notify("Person not found.", type="warning")
src/transcription/ui/pages/sources_page.py
+11
View File
@@ -176,6 +176,17 @@ def register_page() -> None:
source.date_revised.isoformat() if source.date_revised else "Not revised",
)
with archival_card(title="Job Source Outcomes"):
if not source.job_sources:
ui.label("No job-source execution records found for this source.").classes("text-xs ui-text-muted")
else:
for job_source in sorted(source.job_sources, key=lambda item: item.executed_at, reverse=True):
with ui.column().classes("w-full gap-1 p-2 ui-row-surface rounded"):
metadata_row("Job ID:", str(job_source.job_id))
metadata_row("Status:", job_source.status.value)
metadata_row("Executed At:", job_source.executed_at.isoformat())
metadata_row("Error Detail:", job_source.error_detail or "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"
src/transcription/worker.py
+8 -1
View File
@@ -152,8 +152,15 @@ async def run_worker_loop(
wake_event.clear()
processed_any = False
while await process_next_queued_job(session_factory=session_factory):
while True:
with handle_worker_exceptions(operation="worker.process_next_queued_job"):
processed = await process_next_queued_job(session_factory=session_factory)
if not processed:
break
processed_any = True
continue
break
if wake_event is None and not processed_any:
await asyncio.sleep(poll_interval_seconds)
tests/services/test_document_service.py
+7 -3
View File
@@ -14,7 +14,6 @@ from transcription.db.models import Person
from transcription.db.models import Source
from transcription.services.documents import DocumentDeleteBlockedError
from transcription.services.documents import DocumentError
from transcription.services.documents import PersonDeleteBlockedError
from transcription.services.documents import DocumentService
@@ -153,7 +152,7 @@ async def test_update_person_refreshes_updated_timestamp(default_session_factory
@pytest.mark.asyncio
async def test_delete_person_blocks_when_linked_documents_exist(default_session_factory):
async def test_delete_person_removes_links_when_linked_documents_exist(default_session_factory):
service = DocumentService(session_factory=default_session_factory)
document = await service.create_document(
@@ -172,9 +171,14 @@ async def test_delete_person_blocks_when_linked_documents_exist(default_session_
)
)
with pytest.raises(PersonDeleteBlockedError):
await service.delete_person(person)
links = await service.list_document_people(person_id=person.id)
assert links == []
with pytest.raises(DocumentError):
await service.read_person_detail(person.id)
@pytest.mark.asyncio
async def test_delete_person_succeeds_when_unlinked(default_session_factory):
tests/services/test_store.py
+48
View File
@@ -1,3 +1,4 @@
from pathlib import Path
from uuid import uuid4
import pytest
@@ -9,7 +10,9 @@ from transcription.db.models import Job
from transcription.db.models import JobSource
from transcription.db.models import Source
from transcription.services.store import UploadError
from transcription.services.store import create_upload_job
from transcription.services.store import create_job_for_document
from transcription.services.store import store_person_portrait
@pytest.mark.asyncio
@@ -66,7 +69,52 @@ async def test_create_job_for_document_sorts_uploads_and_creates_links(async_ses
assert [source.upload_name for source in sources] == ["A_page.pdf", "b_page.pdf"]
assert all(source.filename.endswith(".pdf") for source in sources)
assert all("A_page" not in source.filename and "b_page" not in source.filename for source in sources)
assert all(Path(source.filename).stem == str(source.id) for source in sources)
assert all(Path(source.file_path).parent == (tmp_path / "documents" / str(document.id)) for source in sources)
job_sources = (await async_session.exec(select(JobSource).where(JobSource.job_id == result.job_id))).all()
assert len(job_sources) == 2
assert set(result.source_ids) == {job_source.source_id for job_source in job_sources}
@pytest.mark.asyncio
async def test_create_upload_job_stores_source_under_document_id_directory(async_session, tmp_path):
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
result = await create_upload_job(
filename="single-page.jpg",
file_bytes=b"image-bytes",
session=async_session,
settings=settings,
)
expected_parent = tmp_path / "documents" / str(result.document_id)
assert result.stored_path.parent == expected_parent
assert result.stored_path.exists()
source = (
await async_session.exec(
select(Source)
.where(Source.document_id == result.document_id)
.order_by(Source.page_number) # pyright: ignore[reportArgumentType]
)
).first()
assert source is not None
assert Path(source.filename).stem == str(source.id)
assert result.stored_path.name == source.filename
assert Path(source.file_path).parent == expected_parent
def test_store_person_portrait_stores_file_under_person_id_directory(tmp_path):
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
person_id = uuid4()
stored_path = store_person_portrait(
person_id=person_id,
filename="portrait.png",
file_bytes=b"portrait-bytes",
settings=settings,
)
assert stored_path.parent == (tmp_path / "persons" / str(person_id))
assert stored_path.exists()
tests/test_worker.py
+29
View File
@@ -0,0 +1,29 @@
import asyncio
import logging
import pytest
from transcription.worker import run_worker_loop
@pytest.mark.asyncio
async def test_run_worker_loop_survives_process_next_exception(monkeypatch, caplog):
calls = 0
stop_event = asyncio.Event()
async def _fake_process_next_queued_job(*, session=None, session_factory=None):
nonlocal calls
_ = (session, session_factory)
calls += 1
if calls == 1:
raise RuntimeError("boom")
stop_event.set()
return False
monkeypatch.setattr("transcription.worker.process_next_queued_job", _fake_process_next_queued_job)
with caplog.at_level(logging.ERROR):
await run_worker_loop(stop_event=stop_event, poll_interval_seconds=0)
assert calls == 2
assert "Worker loop exception" in caplog.text
tests/ui/test_people_page.py
+3 -4
View File
@@ -206,7 +206,7 @@ class TestPeoplePageRendering:
assert "This action permanently deletes the person record." in response.text
assert "Delete person permanently" in response.text
def test_person_delete_page_shows_blocked_state_when_linked_documents_exist(self, app_client):
def test_person_delete_page_warns_links_will_be_removed_when_linked_documents_exist(self, app_client):
_, client = app_client
async def _seed_links() -> str:
@@ -233,6 +233,5 @@ class TestPeoplePageRendering:
response = client.get(f"/ui/people/{person_id}/delete")
assert response.status_code == 200
assert "Delete is blocked because linked documents exist." in response.text
assert "Linked documents: 1" in response.text
assert "Go to Documents" in response.text
assert "This will also remove 1 linked document relationship(s)." in response.text
assert "Delete person permanently" in response.text
tests/ui/test_sources_page.py
+30
View File
@@ -9,6 +9,7 @@ from sqlmodel import select
from transcription.db import session_scope
from transcription.db.models import Document
from transcription.db.models import Job
from transcription.db.models import JobStatus
from transcription.db.models import Source
@@ -140,6 +141,35 @@ class TestSourcesPageRendering:
assert "Stored Filename:" in response.text
assert "Delete Source" in response.text
def test_source_detail_page_displays_job_source_status_and_error_detail(self, app_client, seed_job):
_, client = app_client
job_id = seed_job(
filename="failed-source.png",
status=JobStatus.FAILED,
transcription_text=None,
error_detail="Provider timed out",
)
async def _get_source_id() -> str:
async with session_scope() as session:
job = await session.get(Job, job_id)
assert job is not None
source = (
await session.exec(select(Source).where(Source.document_id == job.document_id))
).first()
assert source is not None
return str(source.id)
source_id = asyncio.run(_get_source_id())
response = client.get(f"/ui/sources/{source_id}")
assert response.status_code == 200
assert "JOB SOURCE OUTCOMES" in response.text
assert "Status:" in response.text
assert "failed" in response.text.lower()
assert "Error Detail:" in response.text
assert "Provider timed out" in response.text
def test_source_delete_page_blocks_when_source_is_job_linked(self, app_client, seed_job):
_, client = app_client
job_id = seed_job(filename="linked-source.png", transcription_text="linked text")