diff --git a/src/transcription/config.py b/src/transcription/config.py index f3aa56f..d56c024 100644 --- a/src/transcription/config.py +++ b/src/transcription/config.py @@ -75,7 +75,7 @@ LOGGING_CONFIG: dict[str, object] = { "disable_existing_loggers": False, "formatters": { "standard": { - "format": "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s", + "format": "%(asctime)s %(levelname)-8s | %(message)s", "datefmt": "%Y-%m-%d %H:%M:%S", } }, diff --git a/src/transcription/services/documents.py b/src/transcription/services/documents.py index 273a6fe..c2277f1 100644 --- a/src/transcription/services/documents.py +++ b/src/transcription/services/documents.py @@ -15,7 +15,6 @@ from ..errors import AppError from ..errors import ErrorCategory from ..models import Document from ..models import Job -from ..models import JobStatus from .base import ServiceBase from .store import store_file @@ -178,10 +177,7 @@ async def _create_upload_records( session.add(document) await session.flush() - job = Job( - document_id=document.id, - status=JobStatus.QUEUED, - ) + job = Job(document_id=document.id) session.add(job) await session.commit() await session.refresh(document) diff --git a/src/transcription/worker.py b/src/transcription/worker.py index 844fc42..33a0930 100644 --- a/src/transcription/worker.py +++ b/src/transcription/worker.py @@ -70,18 +70,40 @@ async def _get_queue_item(queue: asyncio.Queue[UUID]) -> AsyncGenerator[UUID]: queue.task_done() -async def process_job(job: Job, services: ServiceBundle) -> None: +async def process_job( + job: Job, + services: ServiceBundle, + settings: Settings | None = None, + session: AsyncSession | None = None, +) -> Job | None: """Process a single job using the service bundle.""" + settings = settings or get_settings() match job.status: case JobStatus.QUEUED: job.status = JobStatus.PROCESSING try: - await services.transcriptions.transcribe_image(job.document.file_path) + # await services.transcriptions.transcribe_image(job.document.file_path) job.status = JobStatus.TRANSCRIBED except TranscriptionError as exc: job.status = JobStatus.FAILED job.error_message = str(exc) - await services.jobs.update_job(job) + logger.error( + "Job failed operation=worker.process_job job_id=%s document_id=%s error=%s", + job.id, + job.document.id, + exc, + ) + case JobStatus.FAILED: + if job.retry_count < settings.worker_max_retries: + job.retry_count += 1 + job.status = JobStatus.QUEUED + logger.info(f"Queuing job {job.id} for retry {job.retry_count}/{settings.worker_max_retries}") + else: + logger.error(f"Job {job.id} has failed and reached max retries.") + return + case _: + return + return await services.jobs.update_job(job, session=session) async def run_worker_loop(