generated from john/python-template
904 lines
37 KiB
Python
904 lines
37 KiB
Python
"""Source persistence, media policy, revisions, and transcription execution."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import logging
|
|
from collections.abc import Sequence
|
|
from contextlib import contextmanager
|
|
from dataclasses import dataclass
|
|
from datetime import UTC
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from pydantic import BaseModel
|
|
from pydantic import ConfigDict
|
|
from pydantic import Field
|
|
from pydantic import JsonValue
|
|
from pydantic import TypeAdapter
|
|
from pydantic import ValidationError
|
|
from sqlalchemy import func
|
|
from sqlalchemy import literal
|
|
from sqlalchemy import tuple_
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
from sqlmodel import col
|
|
from sqlmodel import select
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
|
|
from transcription.config import Settings
|
|
from transcription.config import get_settings
|
|
from transcription.db.models import ExecutionAttempt
|
|
from transcription.db.models import Job
|
|
from transcription.db.models import JobSource
|
|
from transcription.db.models import JobSourceStatus
|
|
from transcription.db.models import Source
|
|
from transcription.errors import ErrorCategory
|
|
from transcription.providers import ProviderAuthError
|
|
from transcription.providers import ProviderError
|
|
from transcription.providers import ProviderResponseError
|
|
from transcription.providers import RequestManifest
|
|
from transcription.providers import SourceEvidenceReference
|
|
from transcription.providers import TranscriptionMetadata
|
|
from transcription.providers import TranscriptionProvider
|
|
from transcription.providers import TranscriptionResult
|
|
from transcription.providers import TransportEvidence
|
|
from transcription.providers import get_transcription_provider
|
|
|
|
from ..db.loading import orm_attribute
|
|
from ..db.loading import selectinload
|
|
from .base import ServiceBase
|
|
from .errors import PromptLoadError
|
|
from .errors import SourceDeleteBlockedError
|
|
from .errors import TranscriptionError
|
|
from .errors import TranscriptionNotFoundError
|
|
from .source_media import lookup_source_mime_type
|
|
from .source_media import supported_source_formats
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DEFAULT_PROMPT_FILE = "transcribe_document.md"
|
|
JSON_OBJECT_ADAPTER = TypeAdapter(dict[str, JsonValue])
|
|
|
|
|
|
class PromptExecution(BaseModel):
|
|
"""Resolved prompt inputs captured for one page execution."""
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True, str_strip_whitespace=True)
|
|
|
|
prompt_name: str = Field(min_length=1, pattern=r"^[^/\\]+$")
|
|
prompt_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
|
|
system_prompt: str | None
|
|
user_prompt: str = Field(min_length=1)
|
|
temperature: float | None = Field(ge=0.0, le=2.0)
|
|
top_p: float | None = Field(ge=0.0, le=1.0)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SourceNavigation:
|
|
"""Adjacent Source identifiers within one ordered Document."""
|
|
|
|
previous_id: UUID | None
|
|
next_id: UUID | None
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ProviderInput:
|
|
"""Resolved immutable bytes and evidence identity for one provider request."""
|
|
|
|
path: Path
|
|
digest_sha256: str
|
|
byte_size: int
|
|
media_type: str
|
|
|
|
|
|
def build_provider_input(source: Source, *, upload_dir: Path) -> ProviderInput:
|
|
"""Describe the stored Source bytes that a provider request will carry.
|
|
|
|
Stored pages are normalized upright at ingest, so the file on disk is the
|
|
exact payload sent to the provider and ``file_hash`` already identifies it.
|
|
"""
|
|
return ProviderInput(
|
|
path=(upload_dir / Path(source.file_path)).resolve(),
|
|
digest_sha256=source.file_hash.lower(),
|
|
byte_size=source.file_size_bytes,
|
|
media_type=source_mime_type(source.file_path),
|
|
)
|
|
|
|
|
|
class SourceService(ServiceBase):
|
|
"""Manage source records, media payloads, revisions, and page execution output."""
|
|
|
|
def __init__(
|
|
self,
|
|
session_factory: async_sessionmaker[AsyncSession] | None = None,
|
|
settings: Settings | None = None,
|
|
):
|
|
super().__init__(session_factory=session_factory, settings=settings)
|
|
self._provider: TranscriptionProvider | None = None
|
|
|
|
@property
|
|
def provider(self) -> TranscriptionProvider:
|
|
if self._provider is None:
|
|
self._provider = get_transcription_provider(settings=self.settings)
|
|
return self._provider
|
|
|
|
async def aclose(self) -> None:
|
|
"""Close provider-owned network resources when they were initialized."""
|
|
if self._provider is None:
|
|
return
|
|
close = getattr(self._provider, "aclose", None)
|
|
if close is not None:
|
|
await close()
|
|
self._provider = None
|
|
|
|
async def _read_source(
|
|
self,
|
|
*,
|
|
session: AsyncSession,
|
|
source_id: UUID,
|
|
options: Sequence[Any] = (),
|
|
suggestion: str = "Verify the source id and retry.",
|
|
) -> Source:
|
|
return await self._get_or_raise(
|
|
Source,
|
|
source_id,
|
|
session=session,
|
|
error=TranscriptionNotFoundError,
|
|
noun="Source",
|
|
suggestion=suggestion,
|
|
options=options,
|
|
)
|
|
|
|
async def create_source(self, source: Source, *, session: AsyncSession | None = None) -> Source:
|
|
"""Create a new source page record in the database."""
|
|
async with self._session_scope(session) as _session:
|
|
_session.add(source)
|
|
await self._finalize(session=_session, caller_session=session, refresh=(source,))
|
|
return source
|
|
|
|
async def read_source(self, source_id: UUID, *, session: AsyncSession | None = None) -> Source:
|
|
"""Read an existing source page record."""
|
|
async with self._session_scope(session) as _session:
|
|
return await self._read_source(session=_session, source_id=source_id)
|
|
|
|
async def read_source_detail(self, source_id: UUID, *, session: AsyncSession | None = None) -> Source:
|
|
"""Read a source page record with job-source context for UI detail rendering."""
|
|
async with self._session_scope(session) as _session:
|
|
query = (
|
|
select(Source)
|
|
.options(
|
|
selectinload(Source.document),
|
|
selectinload(Source.job_sources).selectinload(orm_attribute(JobSource.job)),
|
|
)
|
|
.where(Source.id == source_id)
|
|
.execution_options(populate_existing=True)
|
|
)
|
|
source = (await _session.exec(query)).first()
|
|
|
|
if source is None:
|
|
raise TranscriptionNotFoundError(
|
|
f"Source with id {source_id} not found",
|
|
category=ErrorCategory.NOT_FOUND,
|
|
suggestion="Verify the source id and retry.",
|
|
)
|
|
return source
|
|
|
|
async def read_source_navigation(
|
|
self,
|
|
source_id: UUID,
|
|
*,
|
|
session: AsyncSession | None = None,
|
|
) -> SourceNavigation:
|
|
"""Return adjacent Sources ordered within the current Document."""
|
|
async with self._session_scope(session) as _session:
|
|
source = await self._read_source(session=_session, source_id=source_id)
|
|
position = (col(Source.page_number), col(Source.id))
|
|
current = (literal(source.page_number), literal(source_id))
|
|
|
|
previous_query = (
|
|
select(col(Source.id))
|
|
.where(col(Source.document_id) == source.document_id)
|
|
.where(tuple_(*position) < tuple_(*current))
|
|
.order_by(col(Source.page_number).desc(), col(Source.id).desc())
|
|
.limit(1)
|
|
)
|
|
next_query = (
|
|
select(col(Source.id))
|
|
.where(col(Source.document_id) == source.document_id)
|
|
.where(tuple_(*position) > tuple_(*current))
|
|
.order_by(col(Source.page_number), col(Source.id))
|
|
.limit(1)
|
|
)
|
|
|
|
return SourceNavigation(
|
|
previous_id=(await _session.exec(previous_query)).first(),
|
|
next_id=(await _session.exec(next_query)).first(),
|
|
)
|
|
|
|
async def update_source(self, source: Source, *, session: AsyncSession | None = None) -> Source:
|
|
"""Update an existing source page record."""
|
|
async with self._session_scope(session) as _session:
|
|
merged = await _session.merge(source)
|
|
await self._finalize(session=_session, caller_session=session, refresh=(merged,))
|
|
return merged
|
|
|
|
async def delete_source(self, source: Source, *, session: AsyncSession | None = None) -> None:
|
|
"""Delete a source only when it has no retained execution evidence."""
|
|
await self.delete_unlinked_source(source_id=source.id, session=session)
|
|
|
|
async def delete_unlinked_source(self, *, source_id: UUID, session: AsyncSession | None = None) -> None:
|
|
"""Delete a source only when no JobSource links exist."""
|
|
async with self._session_scope(session) as _session:
|
|
source = await self._read_source(
|
|
session=_session,
|
|
source_id=source_id,
|
|
options=(selectinload(Source.job_sources),),
|
|
)
|
|
|
|
if source.job_sources:
|
|
raise SourceDeleteBlockedError(
|
|
"Source delete blocked because retained execution evidence exists",
|
|
category=ErrorCategory.VALIDATION,
|
|
suggestion="Preserve the source or use an explicit evidence-retention workflow.",
|
|
)
|
|
|
|
source_file_path = source.file_path
|
|
await _session.delete(source)
|
|
await self._finalize(session=_session, caller_session=session)
|
|
|
|
self._delete_source_file(source_file_path=source_file_path)
|
|
|
|
async def list_sources(
|
|
self,
|
|
*,
|
|
document_id: UUID | None = None,
|
|
session: AsyncSession | None = None,
|
|
) -> Sequence[Source]:
|
|
"""List source pages, optionally filtered by document."""
|
|
async with self._session_scope(session) as _session:
|
|
query = select(Source)
|
|
if document_id is not None:
|
|
query = query.where(Source.document_id == document_id)
|
|
result = await _session.exec(query)
|
|
return result.all()
|
|
|
|
async def query_sources(
|
|
self,
|
|
*,
|
|
document_id: UUID | None = None,
|
|
page_number: int | None = None,
|
|
session: AsyncSession | None = None,
|
|
) -> Sequence[Source]:
|
|
"""Query source pages using the provided filters."""
|
|
async with self._session_scope(session) as _session:
|
|
query = select(Source)
|
|
if document_id is not None:
|
|
query = query.where(Source.document_id == document_id)
|
|
if page_number is not None:
|
|
query = query.where(Source.page_number == page_number)
|
|
result = await _session.exec(query)
|
|
return result.all()
|
|
|
|
async def list_sources_detail(
|
|
self,
|
|
*,
|
|
document_id: UUID | None = None,
|
|
job_id: UUID | None = None,
|
|
session: AsyncSession | None = None,
|
|
) -> Sequence[Source]:
|
|
"""List source pages with document/job link context for UI rendering."""
|
|
async with self._session_scope(session) as _session:
|
|
query = select(Source).options(
|
|
selectinload(Source.document),
|
|
# Both are needed by Source.latest_job_source and
|
|
# latest_error_detail: recency comes from the parent job, and
|
|
# failure detail lives on the attempt, not the junction row.
|
|
selectinload(Source.job_sources).selectinload(orm_attribute(JobSource.job)),
|
|
selectinload(Source.job_sources).selectinload(orm_attribute(JobSource.execution_attempts)),
|
|
)
|
|
if document_id is not None:
|
|
query = query.where(Source.document_id == document_id)
|
|
if job_id is not None:
|
|
query = query.join(JobSource, col(JobSource.source_id) == col(Source.id)).where(
|
|
col(JobSource.job_id) == job_id
|
|
)
|
|
|
|
return list((await _session.exec(query)).all())
|
|
|
|
async def create_job_source(
|
|
self,
|
|
job_source: JobSource,
|
|
*,
|
|
session: AsyncSession | None = None,
|
|
) -> JobSource:
|
|
"""Create a new job_source execution record in the database."""
|
|
async with self._session_scope(session) as _session:
|
|
_session.add(job_source)
|
|
try:
|
|
await self._finalize(session=_session, caller_session=session, refresh=(job_source,))
|
|
except IntegrityError as exc:
|
|
raise self._job_source_conflict(job_id=job_source.job_id, source_id=job_source.source_id) from exc
|
|
return job_source
|
|
|
|
async def read_job_source(self, job_source_id: UUID, *, session: AsyncSession | None = None) -> JobSource:
|
|
"""Read an existing job_source record."""
|
|
async with self._session_scope(session) as _session:
|
|
job_source = await _session.get(
|
|
JobSource,
|
|
job_source_id,
|
|
options=(selectinload(JobSource.source),),
|
|
)
|
|
if job_source is None:
|
|
raise TranscriptionNotFoundError(
|
|
f"JobSource with id {job_source_id} not found",
|
|
category=ErrorCategory.NOT_FOUND,
|
|
suggestion="Verify the job source id and retry.",
|
|
)
|
|
return job_source
|
|
|
|
async def read_job_source_for_job(
|
|
self,
|
|
*,
|
|
job_id: UUID,
|
|
source_id: UUID,
|
|
session: AsyncSession | None = None,
|
|
) -> JobSource:
|
|
"""Read the unique JobSource association for one job and Source."""
|
|
async with self._session_scope(session) as _session:
|
|
job_source = (
|
|
await _session.exec(
|
|
select(JobSource).where(JobSource.job_id == job_id).where(JobSource.source_id == source_id)
|
|
)
|
|
).first()
|
|
if job_source is None:
|
|
raise TranscriptionNotFoundError(
|
|
f"Source {source_id} is not linked to Job {job_id}",
|
|
category=ErrorCategory.NOT_FOUND,
|
|
suggestion="Refresh the Job and retry.",
|
|
)
|
|
return job_source
|
|
|
|
async def update_job_source(self, job_source: JobSource, *, session: AsyncSession | None = None) -> JobSource:
|
|
"""Update an existing job_source record."""
|
|
async with self._session_scope(session) as _session:
|
|
merged = await _session.merge(job_source)
|
|
await self._finalize(session=_session, caller_session=session, refresh=(merged,))
|
|
return merged
|
|
|
|
async def delete_job_source(self, job_source: JobSource, *, session: AsyncSession | None = None) -> None:
|
|
"""Delete a job_source record."""
|
|
async with self._session_scope(session) as _session:
|
|
await _session.delete(job_source)
|
|
await self._finalize(session=_session, caller_session=session)
|
|
|
|
async def delete_source_from_job_context(
|
|
self,
|
|
*,
|
|
job_id: UUID,
|
|
source_id: UUID,
|
|
session: AsyncSession | None = None,
|
|
) -> None:
|
|
"""Delete a source from an active job context with dependency guardrails.
|
|
|
|
Policy:
|
|
- Allowed when exactly one JobSource link exists and it points at ``job_id``.
|
|
- Blocked when additional JobSource links exist (history/shared dependencies).
|
|
"""
|
|
async with self._session_scope(session) as _session:
|
|
source = await self._read_source(
|
|
session=_session,
|
|
source_id=source_id,
|
|
options=(selectinload(Source.job_sources),),
|
|
)
|
|
|
|
linked_job_sources = list(source.job_sources)
|
|
attempt_count = (
|
|
await _session.exec(
|
|
select(func.count()).select_from(ExecutionAttempt).where(ExecutionAttempt.source_id == source_id)
|
|
)
|
|
).one()
|
|
if attempt_count:
|
|
raise SourceDeleteBlockedError(
|
|
"Source delete blocked because immutable evidence exists",
|
|
category=ErrorCategory.VALIDATION,
|
|
suggestion="Preserve the source or use an explicit evidence-retention workflow.",
|
|
)
|
|
|
|
matching_links = [job_source for job_source in linked_job_sources if job_source.job_id == job_id]
|
|
if not matching_links:
|
|
raise TranscriptionNotFoundError(
|
|
f"Source {source_id} is not linked to job {job_id}",
|
|
category=ErrorCategory.NOT_FOUND,
|
|
suggestion="Open the source from its linked job context and retry.",
|
|
)
|
|
|
|
if len(linked_job_sources) > len(matching_links):
|
|
raise SourceDeleteBlockedError(
|
|
"Source delete blocked by related job history",
|
|
category=ErrorCategory.VALIDATION,
|
|
suggestion="Remove additional JobSource links first, then retry deletion.",
|
|
)
|
|
|
|
for job_source in matching_links:
|
|
await _session.delete(job_source)
|
|
|
|
source_file_path = source.file_path
|
|
await _session.delete(source)
|
|
await self._finalize(session=_session, caller_session=session)
|
|
|
|
self._delete_source_file(source_file_path=source_file_path)
|
|
|
|
def _delete_source_file(self, *, source_file_path: str) -> None:
|
|
"""Best-effort cleanup for source media files."""
|
|
candidate_path = Path(source_file_path)
|
|
resolved_path = candidate_path if candidate_path.is_absolute() else self.settings.upload_dir / candidate_path
|
|
|
|
if not resolved_path.exists():
|
|
return
|
|
|
|
try:
|
|
resolved_path.unlink()
|
|
logger.info("Deleted source file: %s", resolved_path)
|
|
except OSError:
|
|
logger.warning("Failed to delete source file: %s", resolved_path)
|
|
|
|
async def list_job_sources(
|
|
self,
|
|
*,
|
|
job_id: UUID | None = None,
|
|
session: AsyncSession | None = None,
|
|
) -> Sequence[JobSource]:
|
|
"""List job-source records, optionally filtered by job."""
|
|
async with self._session_scope(session) as _session:
|
|
query = select(JobSource).options(
|
|
selectinload(JobSource.job),
|
|
selectinload(JobSource.source),
|
|
)
|
|
if job_id is not None:
|
|
query = query.where(JobSource.job_id == job_id)
|
|
result = await _session.exec(query)
|
|
return result.all()
|
|
|
|
async def update_job_source_transcription(
|
|
self,
|
|
*,
|
|
job_id: UUID,
|
|
source_id: UUID,
|
|
text: str | None,
|
|
error_detail: str | None = None,
|
|
ai_metadata: TranscriptionMetadata | dict[str, JsonValue] | None = None,
|
|
raw_api_response: dict[str, JsonValue] | None = None,
|
|
provider: str | None = None,
|
|
model: str | None = None,
|
|
request_manifest: RequestManifest | None = None,
|
|
quality_warnings: dict[str, JsonValue] | None = None,
|
|
timing_breakdown: dict[str, JsonValue] | None = None,
|
|
transport_evidence: TransportEvidence | None = None,
|
|
failure_phase: str | None = None,
|
|
error_category: str | None = None,
|
|
started_at: datetime | None = None,
|
|
finished_at: datetime | None = None,
|
|
duration_ms: int | None = None,
|
|
session: AsyncSession | None = None,
|
|
) -> JobSource:
|
|
"""Persist transcription fields for one source within a specific job."""
|
|
async with self._session_scope(session) as _session:
|
|
job = await self._get_or_raise(
|
|
Job,
|
|
job_id,
|
|
session=_session,
|
|
error=TranscriptionNotFoundError,
|
|
noun="Job",
|
|
suggestion="Verify the job id and retry.",
|
|
)
|
|
|
|
source = await self._read_source(session=_session, source_id=source_id)
|
|
|
|
if source.document_id != job.document_id:
|
|
raise TranscriptionError(
|
|
f"Source {source_id} does not belong to job {job_id}",
|
|
category=ErrorCategory.VALIDATION,
|
|
suggestion="Link the source to the same document as the job and retry.",
|
|
)
|
|
|
|
job.provider = provider or job.provider or self.settings.provider.value
|
|
job.model = model or job.model or _resolve_transcript_model(provider=self.provider, settings=self.settings)
|
|
metadata_payload = _validate_transcription_metadata(ai_metadata)
|
|
raw_response_payload = _validate_json_object(raw_api_response, field_name="raw_api_response")
|
|
timing_payload = _validate_json_object(timing_breakdown, field_name="timing_breakdown")
|
|
attempt_metadata = _merge_attempt_metadata(
|
|
metadata=metadata_payload,
|
|
quality_warnings=quality_warnings,
|
|
timing_breakdown=timing_payload,
|
|
)
|
|
|
|
existing_job_source = await _session.exec(
|
|
select(JobSource).where(JobSource.job_id == job_id).where(JobSource.source_id == source_id)
|
|
)
|
|
job_source = existing_job_source.first()
|
|
outcome = JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED
|
|
if job_source is None:
|
|
job_source = JobSource(job_id=job_id, source_id=source_id, status=outcome)
|
|
_session.add(job_source)
|
|
try:
|
|
await _session.flush()
|
|
except IntegrityError as exc:
|
|
raise self._job_source_conflict(job_id=job_id, source_id=source_id) from exc
|
|
else:
|
|
job_source.status = outcome
|
|
|
|
finish_time = finished_at or datetime.now(UTC)
|
|
start_time = started_at or finish_time
|
|
attempt_number = (
|
|
await _session.exec(
|
|
select(func.max(ExecutionAttempt.attempt_number))
|
|
.where(ExecutionAttempt.job_id == job_id)
|
|
.where(ExecutionAttempt.source_id == source_id)
|
|
)
|
|
).one()
|
|
transport = transport_evidence or TransportEvidence(response_received=False)
|
|
manifest_payload = request_manifest.model_dump(mode="json") if request_manifest is not None else None
|
|
software_payload = (
|
|
request_manifest.software.model_dump(mode="json") if request_manifest is not None else None
|
|
)
|
|
attempt = ExecutionAttempt(
|
|
job_source_id=job_source.id,
|
|
job_id=job_id,
|
|
source_id=source_id,
|
|
attempt_number=(attempt_number or 0) + 1,
|
|
status=outcome,
|
|
provider=provider or job.provider or self.settings.provider.value,
|
|
model=model or job.model,
|
|
request_manifest=manifest_payload,
|
|
request_manifest_sha256=request_manifest.digest() if request_manifest is not None else None,
|
|
request_manifest_schema_version=(
|
|
request_manifest.schema_version if request_manifest is not None else None
|
|
),
|
|
response_received=transport.response_received,
|
|
transport_status_code=transport.status_code,
|
|
transport_body=transport.body,
|
|
transport_content_type=transport.content_type,
|
|
transport_content_encoding=transport.content_encoding,
|
|
transport_safe_headers=transport.safe_headers or None,
|
|
router_request_id=transport.request_id,
|
|
router_generation_id=transport.generation_id,
|
|
sdk_response_snapshot=raw_response_payload,
|
|
normalized_metadata=attempt_metadata,
|
|
software_context=software_payload,
|
|
raw_transcription=text,
|
|
error_category=error_category,
|
|
error_detail=error_detail,
|
|
failure_phase=failure_phase,
|
|
started_at=start_time,
|
|
finished_at=finish_time,
|
|
duration_ms=duration_ms
|
|
if duration_ms is not None
|
|
else max(0, int((finish_time - start_time).total_seconds() * 1000)),
|
|
)
|
|
_session.add(attempt)
|
|
await _session.flush()
|
|
|
|
if text is not None and source.raw_transcription is None and source.preferred_execution_attempt_id is None:
|
|
source.raw_transcription = text
|
|
source.preferred_execution_attempt_id = attempt.id
|
|
|
|
await self._finalize(session=_session, caller_session=session, refresh=(job, source, job_source, attempt))
|
|
return job_source
|
|
|
|
@staticmethod
|
|
def _job_source_conflict(*, job_id: UUID, source_id: UUID) -> TranscriptionError:
|
|
return TranscriptionError(
|
|
f"Source {source_id} is already linked to Job {job_id}",
|
|
category=ErrorCategory.CONFLICT,
|
|
suggestion="Use the existing job-source link instead of creating a duplicate.",
|
|
)
|
|
|
|
async def upsert_revision_for_source(
|
|
self,
|
|
*,
|
|
source_id: UUID,
|
|
text: str,
|
|
session: AsyncSession | None = None,
|
|
) -> Source:
|
|
"""Persist a human revision on a source page."""
|
|
async with self._session_scope(session) as _session:
|
|
source = await self._read_source(session=_session, source_id=source_id)
|
|
|
|
source.revised_text = text
|
|
source.date_revised = datetime.now(UTC)
|
|
await self._finalize(session=_session, caller_session=session, refresh=(source,))
|
|
return source
|
|
|
|
async def read_revision_by_source(
|
|
self,
|
|
source_id: UUID,
|
|
*,
|
|
session: AsyncSession | None = None,
|
|
) -> Source | None:
|
|
"""Read the source record for a given page, including any revision text."""
|
|
async with self._session_scope(session) as _session:
|
|
return await _session.get(Source, source_id)
|
|
|
|
async def list_revisions_by_job(
|
|
self,
|
|
job_id: UUID,
|
|
*,
|
|
session: AsyncSession | None = None,
|
|
) -> Sequence[Source]:
|
|
"""List source pages for a job that carry revision text."""
|
|
async with self._session_scope(session) as _session:
|
|
query = (
|
|
select(Source)
|
|
.join(JobSource, col(JobSource.source_id) == col(Source.id))
|
|
.where(col(JobSource.job_id) == job_id)
|
|
.where(col(Source.revised_text).is_not(None))
|
|
.order_by(col(Source.date_revised))
|
|
)
|
|
result = await _session.exec(query)
|
|
return result.all()
|
|
|
|
|
|
def _resolve_transcript_model(*, provider: TranscriptionProvider, settings: Settings) -> str:
|
|
provider_model = provider.model
|
|
if provider_model and provider_model.strip():
|
|
return provider_model
|
|
|
|
if settings.provider_model and settings.provider_model.strip():
|
|
return settings.provider_model
|
|
|
|
return "unknown"
|
|
|
|
|
|
def _validate_transcription_metadata(
|
|
metadata: TranscriptionMetadata | dict[str, JsonValue] | None,
|
|
) -> dict[str, JsonValue] | None:
|
|
if metadata is None:
|
|
return None
|
|
try:
|
|
validated = (
|
|
metadata if isinstance(metadata, TranscriptionMetadata) else TranscriptionMetadata.model_validate(metadata)
|
|
)
|
|
except ValidationError as exc:
|
|
raise TranscriptionError(
|
|
"Transcription metadata failed validation",
|
|
category=ErrorCategory.VALIDATION,
|
|
suggestion="Persist only normalized provider execution metadata.",
|
|
) from exc
|
|
return validated.as_json_object()
|
|
|
|
|
|
def _merge_attempt_metadata(
|
|
metadata: dict[str, JsonValue] | None,
|
|
*,
|
|
quality_warnings: dict[str, JsonValue] | None,
|
|
timing_breakdown: dict[str, JsonValue] | None,
|
|
) -> dict[str, JsonValue] | None:
|
|
"""Attach app-computed metadata to provider-normalized metadata.
|
|
|
|
App-computed values (quality warnings and timing) are namespaced so provider
|
|
metadata remains semantically distinct.
|
|
"""
|
|
if quality_warnings is None and timing_breakdown is None:
|
|
return metadata
|
|
merged: dict[str, JsonValue] = dict(metadata or {})
|
|
if quality_warnings is not None:
|
|
merged["transcription_quality_warnings"] = quality_warnings
|
|
if timing_breakdown is not None:
|
|
merged["processing_timing"] = timing_breakdown
|
|
return merged
|
|
|
|
|
|
def _validate_json_object(
|
|
payload: dict[str, JsonValue] | None,
|
|
*,
|
|
field_name: str,
|
|
) -> dict[str, JsonValue] | None:
|
|
if payload is None:
|
|
return None
|
|
try:
|
|
return JSON_OBJECT_ADAPTER.validate_python(payload)
|
|
except ValidationError as exc:
|
|
raise TranscriptionError(
|
|
f"{field_name} must be a JSON-compatible object",
|
|
category=ErrorCategory.VALIDATION,
|
|
suggestion="Remove non-JSON values before persisting provider diagnostics.",
|
|
) from exc
|
|
|
|
|
|
def hash_prompt_text(prompt_text: str) -> str:
|
|
"""Return the canonical SHA-256 provenance hash for prompt text."""
|
|
return hashlib.sha256(prompt_text.encode("utf-8")).hexdigest()
|
|
|
|
|
|
async def transcribe_document_image(
|
|
image_path: str | Path,
|
|
*,
|
|
prompt_name: str | None = None,
|
|
prompt_text: str | None = None,
|
|
temperature: float | None = None,
|
|
top_p: float | None = None,
|
|
settings: Settings | None = None,
|
|
provider: TranscriptionProvider | None = None,
|
|
source_reference: SourceEvidenceReference | None = None,
|
|
requested_model: str | None = None,
|
|
) -> TranscriptionResult:
|
|
"""Transcribe a local image using the configured prompt and provider."""
|
|
runtime_settings = settings or get_settings()
|
|
if prompt_text is None:
|
|
prompt_execution = await asyncio.to_thread(
|
|
build_prompt_execution, prompt_name=prompt_name, settings=runtime_settings
|
|
)
|
|
else:
|
|
effective_prompt_name = (prompt_name or runtime_settings.default_prompt_name or DEFAULT_PROMPT_FILE).strip()
|
|
prompt_execution = PromptExecution(
|
|
prompt_name=effective_prompt_name,
|
|
prompt_hash=hash_prompt_text(prompt_text),
|
|
system_prompt=None,
|
|
user_prompt=prompt_text,
|
|
temperature=temperature if temperature is not None else runtime_settings.transcription_temperature,
|
|
top_p=top_p if top_p is not None else runtime_settings.transcription_top_p,
|
|
)
|
|
image_bytes, mime_type = await asyncio.to_thread(load_source_payload, image_path)
|
|
|
|
owns_adapter = provider is None
|
|
adapter = provider or get_transcription_provider(settings=runtime_settings)
|
|
logger.info("Starting transcription for image=%s mime_type=%s", image_path, mime_type)
|
|
|
|
try:
|
|
with handle_transcription_errors():
|
|
result = await adapter.transcribe(
|
|
prompt_text=prompt_execution.user_prompt,
|
|
image_bytes=image_bytes,
|
|
mime_type=mime_type,
|
|
temperature=prompt_execution.temperature,
|
|
top_p=prompt_execution.top_p,
|
|
source_reference=source_reference,
|
|
requested_model=requested_model,
|
|
)
|
|
finally:
|
|
if owns_adapter:
|
|
await adapter.aclose()
|
|
logger.info("Transcription completed for image=%s provider=%s", image_path, result.provider)
|
|
return TranscriptionResult(
|
|
text=result.text,
|
|
provider=result.provider,
|
|
prompt_name=prompt_execution.prompt_name,
|
|
prompt_hash=prompt_execution.prompt_hash,
|
|
system_prompt=prompt_execution.system_prompt,
|
|
user_prompt=result.user_prompt or prompt_execution.user_prompt,
|
|
temperature=result.temperature if result.temperature is not None else prompt_execution.temperature,
|
|
top_p=result.top_p if result.top_p is not None else prompt_execution.top_p,
|
|
model=result.model,
|
|
metadata=result.metadata,
|
|
raw_api_response=result.raw_api_response,
|
|
request_manifest=result.request_manifest,
|
|
transport_evidence=result.transport_evidence,
|
|
)
|
|
|
|
|
|
def build_prompt_execution(*, prompt_name: str | None = None, settings: Settings | None = None) -> PromptExecution:
|
|
"""Resolve the exact prompt payload and provenance for one execution."""
|
|
runtime_settings = settings or get_settings()
|
|
effective_prompt_name = (prompt_name or runtime_settings.default_prompt_name or DEFAULT_PROMPT_FILE).strip()
|
|
user_prompt = load_prompt_text(prompt_name=effective_prompt_name, settings=runtime_settings)
|
|
return PromptExecution(
|
|
prompt_name=effective_prompt_name,
|
|
prompt_hash=hash_prompt_text(user_prompt),
|
|
system_prompt=None,
|
|
user_prompt=user_prompt,
|
|
temperature=runtime_settings.transcription_temperature,
|
|
top_p=runtime_settings.transcription_top_p,
|
|
)
|
|
|
|
|
|
def load_prompt_text(*, prompt_name: str = DEFAULT_PROMPT_FILE, settings: Settings | None = None) -> str:
|
|
"""Load and validate prompt text from PROMPT_DIR."""
|
|
runtime_settings = settings or get_settings()
|
|
prompt_root = runtime_settings.prompt_dir.resolve()
|
|
prompt_path = (prompt_root / prompt_name).resolve()
|
|
if prompt_path.parent != prompt_root:
|
|
raise PromptLoadError(
|
|
f"Prompt file must be directly inside PROMPT_DIR: {prompt_name}",
|
|
category=ErrorCategory.VALIDATION,
|
|
suggestion="Configure a prompt filename without directory components.",
|
|
)
|
|
|
|
if not prompt_path.exists() or not prompt_path.is_file():
|
|
raise PromptLoadError(
|
|
f"Prompt file not found: {prompt_path}",
|
|
category=ErrorCategory.INFRA_PERSISTENT,
|
|
suggestion="Verify PROMPT_DIR and prompt file configuration, then retry.",
|
|
)
|
|
|
|
prompt_text = prompt_path.read_text(encoding="utf-8").strip()
|
|
if not prompt_text:
|
|
raise PromptLoadError(
|
|
f"Prompt file is empty: {prompt_path}",
|
|
category=ErrorCategory.VALIDATION,
|
|
suggestion="Populate the prompt file with valid instructions and retry.",
|
|
)
|
|
|
|
logger.info("Loaded prompt artifact: %s", prompt_path)
|
|
return prompt_text
|
|
|
|
|
|
def source_mime_type(filename: str | Path) -> str:
|
|
"""Return the canonical MIME type for a supported Source filename."""
|
|
mime_type = lookup_source_mime_type(filename)
|
|
if mime_type is None:
|
|
suffix = Path(filename).suffix.lower()
|
|
raise TranscriptionError(
|
|
f"Unsupported Source format: {suffix or '<none>'}",
|
|
category=ErrorCategory.USER_INPUT,
|
|
suggestion=f"Use one of the supported Source formats: {supported_source_formats()}.",
|
|
)
|
|
return mime_type
|
|
|
|
|
|
def validate_source_content(*, filename: str | Path, content: bytes) -> str:
|
|
"""Validate Source content and return its canonical MIME type."""
|
|
if not content:
|
|
raise TranscriptionError(
|
|
"Source content is empty",
|
|
category=ErrorCategory.VALIDATION,
|
|
suggestion="Select a non-empty Source file and try again.",
|
|
)
|
|
|
|
safe_name = Path(filename).name
|
|
if not safe_name:
|
|
raise TranscriptionError(
|
|
"Source filename is required",
|
|
category=ErrorCategory.VALIDATION,
|
|
suggestion="Choose a Source file with a valid filename and retry.",
|
|
)
|
|
return source_mime_type(safe_name)
|
|
|
|
|
|
def load_source_payload(source_path: str | Path) -> tuple[bytes, str]:
|
|
"""Read Source bytes and resolve MIME type from the canonical format policy.
|
|
|
|
Blocking. Async callers must dispatch this through ``asyncio.to_thread``.
|
|
"""
|
|
path = Path(source_path)
|
|
|
|
if not path.exists() or not path.is_file():
|
|
raise TranscriptionError(
|
|
f"Source file not found: {path}",
|
|
category=ErrorCategory.NOT_FOUND,
|
|
suggestion="Verify the Source file exists and retry from the jobs page.",
|
|
)
|
|
|
|
content = path.read_bytes()
|
|
return content, validate_source_content(filename=path.name, content=content)
|
|
|
|
|
|
@contextmanager
|
|
def handle_transcription_errors():
|
|
"""Context manager to handle transcription errors."""
|
|
try:
|
|
yield
|
|
except ProviderAuthError as exc:
|
|
raise TranscriptionError(
|
|
"Provider authentication failed",
|
|
category=ErrorCategory.INFRA_PERSISTENT,
|
|
suggestion="Verify provider API credentials and retry.",
|
|
) from exc
|
|
except ProviderResponseError as exc:
|
|
raise TranscriptionError(
|
|
"Provider returned an invalid response",
|
|
category=ErrorCategory.EXTERNAL_PROVIDER,
|
|
suggestion="Retry once. If it persists, try a different provider model or inspect provider status.",
|
|
retriable=True,
|
|
) from exc
|
|
except ProviderError as exc:
|
|
raise TranscriptionError(
|
|
f"Provider transcription failed: {exc}",
|
|
category=ErrorCategory.EXTERNAL_PROVIDER,
|
|
suggestion="Retry the transcription from jobs. If repeated, check provider availability.",
|
|
retriable=True,
|
|
) from exc
|