Step 4 implemented

src/transcription/services/__init__.py

@@ -8,6 +8,12 @@ from transcription.services.transcription import (
     load_prompt_text,
     transcribe_document_image,
 )
+from transcription.services.upload import (
+    SUPPORTED_UPLOAD_EXTENSIONS,
+    UploadError,
+    UploadJobResult,
+    create_upload_job,
+)
 
 __all__ = [
     "DEFAULT_PROMPT_FILE",
@@ -16,4 +22,9 @@ __all__ = [
     "load_image_payload",
     "load_prompt_text",
     "transcribe_document_image",
+    "SUPPORTED_UPLOAD_EXTENSIONS",
+    "UploadError",
+    "UploadJobResult",
+    "create_upload_job",
 ]
+

src/transcription/services/upload.py

@@ -0,0 +1,122 @@
+"""Upload service for storing files and creating queued transcription jobs."""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from uuid import UUID, uuid4
+
+from sqlmodel import Session
+
+from transcription.config import Settings, get_settings
+from transcription.db import get_session
+from transcription.models import Document, Job, JobStatus
+
+logger = logging.getLogger(__name__)
+
+SUPPORTED_UPLOAD_EXTENSIONS = {".jpg", ".jpeg", ".png", ".tif", ".tiff", ".pdf"}
+
+
+class UploadError(RuntimeError):
+    """Raised when uploaded content cannot be persisted safely."""
+
+
+@dataclass(frozen=True)
+class UploadJobResult:
+    """Summary of created upload records."""
+
+    document_id: UUID
+    job_id: UUID
+    stored_path: Path
+    original_filename: str
+
+
+def create_upload_job(
+    *,
+    filename: str,
+    file_bytes: bytes,
+    session: Session | None = None,
+    settings: Settings | None = None,
+) -> UploadJobResult:
+    """Persist an uploaded file and create document/job records."""
+    runtime_settings = settings or get_settings()
+    _validate_upload(filename=filename, file_bytes=file_bytes)
+
+    upload_dir = runtime_settings.upload_dir
+    upload_dir.mkdir(parents=True, exist_ok=True)
+
+    stored_name = _build_stored_filename(filename)
+    stored_path = upload_dir / stored_name
+
+    try:
+        stored_path.write_bytes(file_bytes)
+    except OSError as exc:
+        raise UploadError(f"Failed to persist upload file: {stored_path}") from exc
+
+    try:
+        if session is not None:
+            document, job = _create_upload_records(session=session, original_filename=filename, stored_path=stored_path)
+        else:
+            with get_session() as local_session:
+                document, job = _create_upload_records(
+                    session=local_session,
+                    original_filename=filename,
+                    stored_path=stored_path,
+                )
+    except Exception as exc:  # noqa: BLE001
+        _best_effort_delete(stored_path)
+        raise UploadError("Failed to create upload database records") from exc
+
+    logger.info("Created upload job document_id=%s job_id=%s", document.id, job.id)
+    return UploadJobResult(
+        document_id=document.id,
+        job_id=job.id,
+        stored_path=stored_path,
+        original_filename=Path(filename).name,
+    )
+
+
+def _validate_upload(*, filename: str, file_bytes: bytes) -> None:
+    if not file_bytes:
+        raise UploadError("Upload payload is empty")
+
+    safe_name = Path(filename).name
+    if not safe_name:
+        raise UploadError("Upload filename is required")
+
+    suffix = Path(safe_name).suffix.lower()
+    if suffix not in SUPPORTED_UPLOAD_EXTENSIONS:
+        raise UploadError(f"Unsupported upload extension: {suffix}")
+
+
+def _build_stored_filename(filename: str) -> str:
+    safe_name = Path(filename).name
+    return f"{uuid4()}_{safe_name}"
+
+
+def _create_upload_records(*, session: Session, original_filename: str, stored_path: Path) -> tuple[Document, Job]:
+    document = Document(
+        filename=Path(original_filename).name,
+        file_path=str(stored_path),
+    )
+    session.add(document)
+    session.flush()
+
+    job = Job(
+        document_id=document.id,
+        status=JobStatus.QUEUED,
+    )
+    session.add(job)
+    session.commit()
+    session.refresh(document)
+    session.refresh(job)
+    return document, job
+
+
+def _best_effort_delete(path: Path) -> None:
+    try:
+        if path.exists():
+            path.unlink()
+    except OSError:
+        logger.warning("Failed to clean up upload file after DB error: %s", path)

src/transcription/worker.py

@@ -0,0 +1,95 @@
+"""Background worker for queued transcription jobs."""
+
+from __future__ import annotations
+
+import logging
+import time
+from datetime import datetime, timezone
+from threading import Event
+
+from sqlmodel import Session, select
+
+from transcription.db import get_session
+from transcription.models import Document, Job, JobStatus, Transcript
+from transcription.services.transcription import transcribe_document_image
+
+logger = logging.getLogger(__name__)
+
+
+def process_next_queued_job(*, session: Session | None = None) -> bool:
+    """Process the next queued job and persist terminal outcome.
+
+    Returns True when a job was processed, False when no queued job exists.
+    """
+    if session is None:
+        with get_session() as local_session:
+            return _process_next_queued_job(session=local_session)
+    return _process_next_queued_job(session=session)
+
+
+def _process_next_queued_job(*, session: Session) -> bool:
+    job = session.exec(
+        select(Job)
+        .where(Job.status == JobStatus.QUEUED)
+        .order_by(Job.created_at)
+    ).first()
+
+    if job is None:
+        return False
+
+    logger.info("Picked queued job id=%s", job.id)
+    job.status = JobStatus.PROCESSING
+    job.updated_at = datetime.now(timezone.utc)
+    session.add(job)
+    session.commit()
+    session.refresh(job)
+
+    document = session.get(Document, job.document_id)
+    if document is None:
+        _upsert_transcript(session=session, job_id=job.id, text=None, error_detail="Document not found")
+        job.status = JobStatus.FAILED
+        job.updated_at = datetime.now(timezone.utc)
+        session.add(job)
+        session.commit()
+        logger.error("Job failed because document was missing job_id=%s", job.id)
+        return True
+
+    try:
+        result = transcribe_document_image(document.file_path)
+        _upsert_transcript(session=session, job_id=job.id, text=result.text, error_detail=None)
+        job.status = JobStatus.TRANSCRIBED
+        logger.info("Job transcribed job_id=%s provider=%s", job.id, result.provider)
+    except Exception as exc:  # noqa: BLE001
+        _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=str(exc))
+        job.status = JobStatus.FAILED
+        logger.exception("Job failed job_id=%s", job.id)
+
+    job.updated_at = datetime.now(timezone.utc)
+    session.add(job)
+    session.commit()
+    return True
+
+
+def _upsert_transcript(*, session: Session, job_id, text: str | None, error_detail: str | None) -> Transcript:
+    transcript = session.exec(select(Transcript).where(Transcript.job_id == job_id)).first()
+    if transcript is None:
+        transcript = Transcript(job_id=job_id)
+
+    transcript.text = text
+    transcript.error_detail = error_detail
+    session.add(transcript)
+    session.commit()
+    session.refresh(transcript)
+    return transcript
+
+
+def run_worker_loop(*, stop_event: Event | None = None, poll_interval_seconds: float = 1.0) -> None:
+    """Run worker polling loop until stop_event is set."""
+    while True:
+        if stop_event is not None and stop_event.is_set():
+            logger.info("Worker stop event received")
+            return
+
+        processed = process_next_queued_job()
+        if not processed:
+            time.sleep(poll_interval_seconds)