From 58faa00d7b72ad5d8eaf54beea3d69981de9ee69 Mon Sep 17 00:00:00 2001 From: Jim Lancaster <40281233+zoltan57@users.noreply.github.com> Date: Sat, 8 Aug 2026 14:59:45 -0500 Subject: [PATCH] AI metadata and api prompt results data capture now fixed --- src/transcription/providers/base.py | 3 + src/transcription/providers/openrouter.py | 70 +++++++++++++++++++++ src/transcription/services/transcription.py | 6 ++ src/transcription/services/workflows.py | 41 +++++++++--- tests/integration/test_pipeline_flow.py | 7 +++ tests/providers/test_openrouter.py | 13 +++- tests/services/test_v2_crud.py | 42 +++++++++++++ 7 files changed, 172 insertions(+), 10 deletions(-) diff --git a/src/transcription/providers/base.py b/src/transcription/providers/base.py index 6f68d4b..4aa9b66 100644 --- a/src/transcription/providers/base.py +++ b/src/transcription/providers/base.py @@ -1,6 +1,7 @@ """Provider interfaces and shared types for transcription adapters.""" from dataclasses import dataclass +from typing import Any from typing import Protocol @@ -28,6 +29,8 @@ class TranscriptionResult: usage_input_tokens: int | None = None usage_output_tokens: int | None = None usage_total_tokens: int | None = None + ai_metadata: dict[str, Any] | None = None + raw_api_response: dict[str, Any] | None = None class TranscriptionProvider(Protocol): diff --git a/src/transcription/providers/openrouter.py b/src/transcription/providers/openrouter.py index 86292c2..d7e0152 100644 --- a/src/transcription/providers/openrouter.py +++ b/src/transcription/providers/openrouter.py @@ -66,6 +66,13 @@ class OpenRouterTranscriptionProvider: model = self._get_optional_attr(response, "model") or self.model finish_reason = self._extract_finish_reason(response) usage_input_tokens, usage_output_tokens, usage_total_tokens = self._extract_usage(response) + ai_metadata = self._build_ai_metadata( + finish_reason=finish_reason, + usage_input_tokens=usage_input_tokens, + usage_output_tokens=usage_output_tokens, + usage_total_tokens=usage_total_tokens, + ) + raw_api_response = self._coerce_raw_response(response) logger.info("OpenRouter transcription completed using model=%s", model) return TranscriptionResult( text=text, @@ -76,8 +83,71 @@ class OpenRouterTranscriptionProvider: usage_input_tokens=usage_input_tokens, usage_output_tokens=usage_output_tokens, usage_total_tokens=usage_total_tokens, + ai_metadata=ai_metadata, + raw_api_response=raw_api_response, ) + def _build_ai_metadata( + self, + *, + finish_reason: str | None, + usage_input_tokens: int | None, + usage_output_tokens: int | None, + usage_total_tokens: int | None, + ) -> dict[str, Any] | None: + metadata: dict[str, Any] = {} + if finish_reason is not None: + metadata["finish_reason"] = finish_reason + + usage: dict[str, int] = {} + if usage_input_tokens is not None: + usage["input_tokens"] = usage_input_tokens + if usage_output_tokens is not None: + usage["output_tokens"] = usage_output_tokens + if usage_total_tokens is not None: + usage["total_tokens"] = usage_total_tokens + + if usage: + metadata["usage"] = usage + + return metadata or None + + def _coerce_raw_response(self, response: Any) -> dict[str, Any] | None: + payload = self._to_json_compatible(response) + if payload is None: + return None + if isinstance(payload, dict): + return payload + return {"response": payload} + + def _to_json_compatible(self, value: Any) -> Any: + if value is None or isinstance(value, str | int | float | bool): + return value + + if isinstance(value, dict): + return {str(key): self._to_json_compatible(item) for key, item in value.items()} + + if isinstance(value, list | tuple | set): + return [self._to_json_compatible(item) for item in value] + + for method_name in ("model_dump", "dict", "to_dict"): + serializer = getattr(value, method_name, None) + if callable(serializer): + try: + return self._to_json_compatible(serializer()) + except Exception: # noqa: BLE001 + continue + + object_dict = getattr(value, "__dict__", None) + if isinstance(object_dict, dict): + return { + str(key): self._to_json_compatible(item) + for key, item in object_dict.items() + if not str(key).startswith("_") + } + + return repr(value) + def _build_request(self, *, prompt_text: str, image_bytes: bytes, mime_type: str) -> OpenRouterRequest: image_b64 = base64.b64encode(image_bytes).decode("ascii") data_url = f"data:{mime_type};base64,{image_b64}" diff --git a/src/transcription/services/transcription.py b/src/transcription/services/transcription.py index 8b7d9f6..941685a 100644 --- a/src/transcription/services/transcription.py +++ b/src/transcription/services/transcription.py @@ -415,6 +415,8 @@ class TranscriptionService(ServiceBase): source_id: UUID, text: str | None, error_detail: str | None = None, + ai_metadata: dict[str, object] | None = None, + raw_api_response: dict[str, object] | None = None, provider: str | None = None, model: str | None = None, prompt_name: str = DEFAULT_PROMPT_FILE, @@ -462,11 +464,15 @@ class TranscriptionService(ServiceBase): source_id=source_id, status=JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED, raw_transcription=text, + ai_metadata=ai_metadata, + raw_api_response=raw_api_response, error_detail=error_detail, ) _session.add(job_source) else: job_source.raw_transcription = text + job_source.ai_metadata = ai_metadata + job_source.raw_api_response = raw_api_response job_source.error_detail = error_detail job_source.status = JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED job_source.executed_at = datetime.now(UTC) diff --git a/src/transcription/services/workflows.py b/src/transcription/services/workflows.py index 9dc9214..2df1eea 100644 --- a/src/transcription/services/workflows.py +++ b/src/transcription/services/workflows.py @@ -29,9 +29,13 @@ async def advance_job( ) -> Job | None: """Advance a single job by lifecycle status.""" settings = settings or get_settings() - match job.status: + current_status = _coerce_job_status(job.status) + match current_status: case JobStatus.QUEUED: return await process_queued_job(job=job, services=services, settings=settings, session=session) + case JobStatus.PROCESSING: + # Recover mid-flight jobs by continuing the queued processing path. + return await process_queued_job(job=job, services=services, settings=settings, session=session) case JobStatus.FAILED: if job.retry_count < settings.worker_max_retries: return await services.jobs.update_job_state( @@ -56,18 +60,20 @@ async def process_queued_job( ) -> Job | None: """Process one complete transcription attempt for a queued job.""" runtime_settings = settings or get_settings() - if job.status != JobStatus.QUEUED: + current_status = _coerce_job_status(job.status) + if current_status not in {JobStatus.QUEUED, JobStatus.PROCESSING}: logger.warning(f"Job {job.id} is not queued. Current status: {job.status}") return # Transaction A: claim job for processing. - if session is None: - job = await services.jobs.mark_job_status(job.id, JobStatus.PROCESSING) - else: - # If we're sharing the session, need to make sure setting the Job to PROCESSING is committed before we start - # the transcription, otherwise other workers may see the job as still QUEUED and try to process it. - job = await services.jobs.mark_job_status(job.id, JobStatus.PROCESSING, session=session) - await session.commit() + if current_status == JobStatus.QUEUED: + if session is None: + job = await services.jobs.mark_job_status(job.id, JobStatus.PROCESSING) + else: + # If we're sharing the session, need to make sure setting the Job to PROCESSING is committed before we start + # the transcription, otherwise other workers may see the job as still QUEUED and try to process it. + job = await services.jobs.mark_job_status(job.id, JobStatus.PROCESSING, session=session) + await session.commit() source_job = await services.jobs.read_job(job_id=job.id, session=session) sources = _resolve_job_sources(source_job) @@ -376,6 +382,8 @@ async def _finalize_batch_outcome( source_id=source.id, text=result.text, error_detail=None, + ai_metadata=result.ai_metadata, + raw_api_response=result.raw_api_response, provider=result.provider, model=result.model, prompt_name=result.prompt_name, @@ -402,6 +410,8 @@ async def _finalize_batch_outcome( source_id=source.id, text=result.text, error_detail=None, + ai_metadata=result.ai_metadata, + raw_api_response=result.raw_api_response, provider=result.provider, model=result.model, prompt_name=result.prompt_name, @@ -470,3 +480,16 @@ def _line_count(text: str) -> int: if not stripped: return 0 return sum(1 for line in stripped.splitlines() if line.strip()) + + +def _coerce_job_status(value: object) -> JobStatus | None: + if isinstance(value, JobStatus): + return value + + if isinstance(value, str): + lowered = value.strip().lower() + for member in JobStatus: + if lowered in {member.value.lower(), member.name.lower()}: + return member + + return None diff --git a/tests/integration/test_pipeline_flow.py b/tests/integration/test_pipeline_flow.py index 1726f1b..8e4e5aa 100644 --- a/tests/integration/test_pipeline_flow.py +++ b/tests/integration/test_pipeline_flow.py @@ -76,6 +76,8 @@ class TestPipelineSuccessFlow: provider="openrouter", model="test-model", prompt_name="transcribe_document.md", + ai_metadata={"finish_reason": "stop", "usage": {"total_tokens": 42}}, + raw_api_response={"id": "resp_123", "choices": [{"message": {"content": "Pipeline transcript"}}]}, ) monkeypatch.setattr( @@ -94,6 +96,11 @@ class TestPipelineSuccessFlow: assert job is not None assert job.status == JobStatus.TRANSCRIBED assert any(job_source.raw_transcription == "Pipeline transcript" for job_source in job.job_sources) + assert any(job_source.ai_metadata == {"finish_reason": "stop", "usage": {"total_tokens": 42}} for job_source in job.job_sources) + assert any( + job_source.raw_api_response == {"id": "resp_123", "choices": [{"message": {"content": "Pipeline transcript"}}]} + for job_source in job.job_sources + ) assert all(job_source.error_detail is None for job_source in job.job_sources) @pytest.mark.asyncio diff --git a/tests/providers/test_openrouter.py b/tests/providers/test_openrouter.py index 7e25090..261add0 100644 --- a/tests/providers/test_openrouter.py +++ b/tests/providers/test_openrouter.py @@ -78,7 +78,13 @@ class TestOpenRouterProviderTranscribe: """Transcribe returns normalized text from a valid response payload.""" response = { "model": "vendor/model-b", - "choices": [{"message": {"content": [{"text": "Line 1"}, {"text": "Line 2"}]}}], + "choices": [ + { + "message": {"content": [{"text": "Line 1"}, {"text": "Line 2"}]}, + "finish_reason": "stop", + } + ], + "usage": {"prompt_tokens": 10, "completion_tokens": 25, "total_tokens": 35}, } provider = OpenRouterTranscriptionProvider( settings=Settings(openrouter_api_key="test-key"), @@ -94,6 +100,11 @@ class TestOpenRouterProviderTranscribe: assert result.text == "Line 1\nLine 2" assert result.provider == "openrouter" assert result.model == "vendor/model-b" + assert result.ai_metadata == { + "finish_reason": "stop", + "usage": {"input_tokens": 10, "output_tokens": 25, "total_tokens": 35}, + } + assert result.raw_api_response == response @pytest.mark.asyncio async def test_maps_sdk_exception_to_provider_error(self): diff --git a/tests/services/test_v2_crud.py b/tests/services/test_v2_crud.py index 0bad960..29216df 100644 --- a/tests/services/test_v2_crud.py +++ b/tests/services/test_v2_crud.py @@ -211,3 +211,45 @@ async def test_source_delete_blocks_when_linked_to_multiple_jobs(default_session with pytest.raises(SourceDeleteBlockedError): await transcriptions.delete_source_from_job_context(job_id=job_one.id, source_id=source.id) + + +@pytest.mark.asyncio +async def test_update_job_source_transcription_persists_provider_json_payloads(default_session_factory): + documents = DocumentService(session_factory=default_session_factory) + jobs = JobService(session_factory=default_session_factory) + transcriptions = TranscriptionService(session_factory=default_session_factory) + + document = await documents.create_document(Document(id=uuid4(), name="provider-payloads-doc")) + job = await jobs.create_job(Job(document_id=document.id)) + source = await transcriptions.create_source( + Source( + document_id=document.id, + page_number=1, + upload_name="provider.jpg", + filename="provider.jpg", + file_path="uploads/provider.jpg", + ) + ) + await transcriptions.create_job_source( + JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING) + ) + + metadata = {"finish_reason": "stop", "usage": {"input_tokens": 11, "output_tokens": 22, "total_tokens": 33}} + raw_payload = {"id": "resp_xyz", "choices": [{"message": {"content": "provider transcript"}}]} + + await transcriptions.update_job_source_transcription( + job_id=job.id, + source_id=source.id, + text="provider transcript", + ai_metadata=metadata, + raw_api_response=raw_payload, + provider="openrouter", + model="test-model", + prompt_name="transcribe_document.md", + ) + + stored_rows = await transcriptions.list_job_sources(job_id=job.id) + assert len(stored_rows) == 1 + assert stored_rows[0].raw_transcription == "provider transcript" + assert stored_rows[0].ai_metadata == metadata + assert stored_rows[0].raw_api_response == raw_payload