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

V4.7 Phase 1: ingest orientation normalization, ProcessingArtifact removal

Browse Source 
Move orientation normalization to the Source-ingest boundary and delete the
ProcessingArtifact subsystem it was built to serve.

Stored pages are now already upright, so nothing downstream derives a rotated
copy: every stored byte is the byte a provider is later sent. Rotation runs in
store_source_file ahead of hashing, so source.file_hash and file_size_bytes
describe exactly what is on disk. normalize_orientation becomes bytes-in /
bytes-out, and JPEG output reuses the source quantization tables and chroma
subsampling instead of re-quantizing at a fixed quality - measured at 50.3-56.1
dB PSNR at -6% size, against 50.0-53.5 dB at +38% for quality=95.

ProcessingArtifact held 2 rows against 77 successful transcriptions; the
subsystem effectively never ran. Deleting it removes the artifact cluster from
sources.py, the derivative resolution in workflows.py, the pre-provider commit
that only existed to make an artifact row durable, and the artifact evidence
dump from the Source detail page. The transcription_quality_warnings payload
folds into execution_attempt.normalized_metadata, so that feature keeps working
without the table.

tools/migrate_v46_to_v47.py carries steps 1 and 2: it rotated the 58 stored
images carrying EXIF orientation 3 in place, updated their recorded hash and
size, dropped processing_artifact and removed its one external file. It is
idempotent, keyed on state rather than a version marker.

tools/migrate_v45_to_v46.py is deleted. That migration is complete, and after
V4.7 it would restore a V4.5 backup into a schema that no longer matches.

Also fixes tests/test_config.py, which read the developer's local .env and
failed whenever WORKER_MAX_RETRIES was set.

Co-authored-by: Copilot App <[email protected]>
This commit is contained in:
zoltan57
2026-08-18 10:16:38 -05:00
co-authored by Copilot App
parent 246d7f9434
commit f86c0ff27b
15 changed files with 494 additions and 1240 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/transcription/config.py
-2
View File
@@ -101,9 +101,7 @@ class Settings(BaseSettings):
# --- filesystem paths ---
upload_dir: Path = Path("./uploads")
prompt_dir: Path = Path("./prompts")
artifact_dir: Path = Path("./data/artifacts")
homepage_dir: Path = Path("./data/homepage")
artifact_inline_threshold_bytes: int = Field(default=1_048_576, ge=1)
# --- worker reliability ---
worker_max_retries: int = Field(default=0, ge=0)
src/transcription/db/models.py
-46
View File
@@ -12,7 +12,6 @@ from uuid import uuid4
from pydantic import JsonValue
from sqlalchemy import JSON
from sqlalchemy import BigInteger
from sqlalchemy import CheckConstraint
from sqlalchemy import Column
from sqlalchemy import Enum as SAEnum
from sqlalchemy import ForeignKey
@@ -319,10 +318,6 @@ class Source(SQLModel, table=True):
back_populates="source",
sa_relationship_kwargs={"lazy": "raise"},
)
processing_artifacts: list["ProcessingArtifact"] = Relationship(
back_populates="source",
sa_relationship_kwargs={"lazy": "noload"},
)
@property
def latest_job_source(self) -> Optional["JobSource"]:
@@ -428,44 +423,3 @@ class ExecutionAttempt(SQLModel, table=True):
job_source: Optional["JobSource"] = Relationship(
back_populates="execution_attempts", sa_relationship_kwargs={"lazy": "raise"}
)
artifacts: list["ProcessingArtifact"] = Relationship(
back_populates="execution_attempt", sa_relationship_kwargs={"lazy": "noload"}
)
class ProcessingArtifact(SQLModel, table=True):
"""Provider-neutral, versioned output derived from a Source."""
__tablename__ = "processing_artifact"
__table_args__ = (
CheckConstraint(
"(inline_payload IS NOT NULL AND external_reference IS NULL) OR "
"(inline_payload IS NULL AND external_reference IS NOT NULL)",
name="ck_processing_artifact_one_content_location",
),
)
id: UUID = Field(default_factory=uuid4, primary_key=True)
source_id: UUID = Field(foreign_key="source.id", index=True)
execution_attempt_id: UUID | None = Field(default=None, foreign_key="execution_attempt.id", index=True)
artifact_type: str
media_type: str
schema_name: str
schema_version: str
producer: str
producer_version: str
inline_payload: dict[str, JsonValue] | None = Field(default=None, sa_column=Column(JSONBCompat(), nullable=True))
external_reference: str | None = None
payload_sha256: str = Field(index=True)
byte_size: int = Field(sa_column=Column(BigInteger(), nullable=False))
coordinate_metadata: dict[str, JsonValue] | None = Field(
default=None, sa_column=Column(JSONBCompat(), nullable=True)
)
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
execution_attempt: Optional["ExecutionAttempt"] = Relationship(
back_populates="artifacts", sa_relationship_kwargs={"lazy": "raise"}
)
source: Optional["Source"] = Relationship(
back_populates="processing_artifacts", sa_relationship_kwargs={"lazy": "raise"}
)
src/transcription/services/jobs.py
+3 -43
View File
@@ -2,7 +2,6 @@ import logging
from collections.abc import Sequence
from datetime import UTC
from datetime import datetime
from pathlib import Path
from uuid import UUID
from sqlalchemy import func
@@ -17,7 +16,6 @@ from ..db.models import Job
from ..db.models import JobSource
from ..db.models import JobSourceStatus
from ..db.models import JobStatus
from ..db.models import ProcessingArtifact
from ..db.models import Source
from ..errors import AppError
from ..errors import ErrorCategory
@@ -276,7 +274,6 @@ class JobService(ServiceBase):
async def delete_job_and_evidence(self, *, job_id: UUID) -> None:
"""Explicitly delete a terminal job and all evidence owned by its attempts."""
external_references: list[str] = []
async with self._session_scope() as session:
job = (
await session.exec(
@@ -302,28 +299,9 @@ class JobService(ServiceBase):
)
).all()
)
if attempts:
attempt_ids = [attempt.id for attempt in attempts]
artifacts = list(
(
await session.exec(
select(ProcessingArtifact).where(
col(ProcessingArtifact.execution_attempt_id).in_(attempt_ids)
)
)
).all()
)
external_references = [
artifact.external_reference
for artifact in artifacts
if artifact.external_reference is not None
]
for artifact in artifacts:
await session.delete(artifact)
await session.flush()
for attempt in attempts:
await session.delete(attempt)
await session.flush()
for attempt in attempts:
await session.delete(attempt)
await session.flush()
for job_source in list(job.job_sources):
await session.delete(job_source)
@@ -331,24 +309,6 @@ class JobService(ServiceBase):
await session.delete(job)
await self._finalize(session=session, caller_session=None)
for external_reference in external_references:
self._delete_external_artifact(external_reference)
def _delete_external_artifact(self, external_reference: str) -> None:
relative_path = Path(external_reference)
if relative_path.is_absolute() or ".." in relative_path.parts:
logger.warning("Skipped unsafe external artifact reference during job deletion: %s", external_reference)
return
artifact_root = self.settings.artifact_dir.resolve()
artifact_path = (artifact_root / relative_path).resolve()
if artifact_root not in artifact_path.parents:
logger.warning("Skipped external artifact outside configured root: %s", external_reference)
return
try:
artifact_path.unlink(missing_ok=True)
except OSError:
logger.warning("Failed to delete external artifact: %s", artifact_path)
async def cancel_job(self, *, job_id: UUID, session: AsyncSession | None = None) -> Job:
"""Cancel a queued/processing job and stop remaining source work."""
async with self._session_scope(session) as _session:
src/transcription/services/normalization.py
+48 -48
View File
@@ -1,25 +1,29 @@
"""Metadata-directed orientation normalization for provider image input."""
"""Metadata-directed orientation normalization applied to image bytes at ingest.
Uploaded pages are stored upright, so nothing downstream has to derive a
rotated copy: every stored byte is already the byte the provider is sent.
"""
from __future__ import annotations
import asyncio
import hashlib
import io
import logging
from dataclasses import dataclass
from pathlib import Path
from PIL import Image
from PIL import JpegImagePlugin
from PIL import UnidentifiedImageError
from PIL.TiffImagePlugin import TiffImageFile
from transcription.errors import AppError
from transcription.errors import ErrorCategory
logger = logging.getLogger(__name__)
ORIENTATION_TAG = 274
ORIENTATION_SCHEMA = "transcription.orientation-normalization"
ORIENTATION_SCHEMA_VERSION = "1"
ORIENTATION_PRODUCER = "transcription.orientation-normalizer"
ORIENTATION_PRODUCER_VERSION = "1"
NORMALIZED_MEDIA_TYPES = frozenset({"image/jpeg", "image/png", "image/tiff"})
_TRANSPOSE_BY_ORIENTATION = {
3: (Image.Transpose.ROTATE_180, 180),
@@ -34,46 +38,44 @@ class OrientationNormalizationError(AppError):
@dataclass(frozen=True, slots=True)
class OrientationNormalization:
"""Exact derivative bytes and transformation metadata."""
"""Upright image bytes and the rotation that produced them."""
content: bytes
media_type: str
suffix: str
original_orientation: int
applied_rotation_degrees: int
original_width: int
original_height: int
derivative_width: int
derivative_height: int
# Computed eagerly by `normalize_orientation`, which already runs off the
# event loop, so callers never hash multi-megabyte derivatives inline.
digest_sha256: str
def normalize_orientation(path: str | Path, *, media_type: str) -> OrientationNormalization | None:
def normalize_orientation(content: bytes, *, media_type: str) -> OrientationNormalization | None:
"""Physically apply supported EXIF rotation, returning None for a safe no-op.
JPEG output reuses the source quantization tables and chroma subsampling
rather than re-quantizing at a fixed quality. Measured across the corpus
that is better on both axes at once - 51.5-55.0 dB PSNR against 50.0-53.5,
and slightly smaller output against 38% larger - and it imposes no
constraint on image dimensions.
Blocking. Async callers must use :func:`normalize_orientation_async`.
"""
source_path = Path(path)
if media_type not in {"image/jpeg", "image/png", "image/tiff"}:
if media_type not in NORMALIZED_MEDIA_TYPES:
return None
try:
with Image.open(source_path) as image:
image_file = Image.open(io.BytesIO(content))
except (OSError, ValueError, UnidentifiedImageError):
# Undecodable content is not this function's business to reject. Ingest
# accepted such bytes before orientation moved here, and decision A
# forbids V4.7 changing what an upload does.
logger.info("Skipped orientation normalization for undecodable content (%s)", media_type)
return None
try:
with image_file as image:
orientation = int(image.getexif().get(ORIENTATION_TAG, 1))
transformation = _TRANSPOSE_BY_ORIENTATION.get(orientation)
if transformation is None:
return None
transpose, rotation = transformation
if isinstance(image, TiffImageFile):
original_width = int(image.tag_v2.get(256, image.width))
original_height = int(image.tag_v2.get(257, image.height))
# Pillow applies TIFF orientation while decoding; copying freezes those upright pixels.
normalized = image.copy()
else:
original_width, original_height = image.size
normalized = image.transpose(transpose)
# Pillow applies TIFF orientation while decoding; copying freezes those upright pixels.
normalized = image.copy() if isinstance(image, TiffImageFile) else image.transpose(transpose)
output = io.BytesIO()
exif = normalized.getexif()
if ORIENTATION_TAG in exif:
@@ -81,37 +83,35 @@ def normalize_orientation(path: str | Path, *, media_type: str) -> OrientationNo
save_kwargs: dict[str, object] = {"format": image.format}
if image.format in {"JPEG", "PNG"}:
save_kwargs["exif"] = exif.tobytes()
if image.format == "JPEG":
save_kwargs.update({"quality": 95, "subsampling": 0})
if isinstance(image, JpegImagePlugin.JpegImageFile):
# Reusing the source quantization tables and subsampling preserves fidelity
# at a smaller size than any re-encode quality setting.
save_kwargs.update(
{
"qtables": image.quantization,
"subsampling": JpegImagePlugin.get_sampling(image),
"optimize": True,
}
)
normalized.save(output, **save_kwargs)
except (OSError, ValueError, UnidentifiedImageError) as exc:
raise OrientationNormalizationError(
f"Source image orientation could not be normalized: {source_path.name}",
"Source image orientation could not be normalized",
category=ErrorCategory.VALIDATION,
suggestion="Verify that the curated Source is a valid supported raster image.",
suggestion="Verify that the uploaded Source is a valid supported raster image.",
) from exc
suffix = source_path.suffix.lower()
content = output.getvalue()
return OrientationNormalization(
content=content,
media_type=media_type,
suffix=suffix,
content=output.getvalue(),
original_orientation=orientation,
applied_rotation_degrees=rotation,
original_width=original_width,
original_height=original_height,
derivative_width=normalized.width,
derivative_height=normalized.height,
digest_sha256=hashlib.sha256(content).hexdigest(),
)
async def normalize_orientation_async(path: str | Path, *, media_type: str) -> OrientationNormalization | None:
async def normalize_orientation_async(content: bytes, *, media_type: str) -> OrientationNormalization | None:
"""Run :func:`normalize_orientation` off the event loop.
Pillow decode, transpose, and re-encode are CPU- and disk-bound and scale
with page size, so they must not run on the request or worker event loop
([MED-01]).
Pillow decode, transpose, and re-encode are CPU-bound and scale with page
size, so they must not run on the request or worker event loop ([MED-01]).
"""
return await asyncio.to_thread(normalize_orientation, path, media_type=media_type)
return await asyncio.to_thread(normalize_orientation, content, media_type=media_type)
src/transcription/services/sources.py
+39 -357
View File
@@ -6,7 +6,6 @@ import asyncio
import base64
import hashlib
import logging
import os
from collections.abc import Sequence
from contextlib import contextmanager
from dataclasses import dataclass
@@ -15,7 +14,6 @@ from datetime import datetime
from pathlib import Path
from typing import Any
from uuid import UUID
from uuid import uuid4
from pydantic import BaseModel
from pydantic import ConfigDict
@@ -38,7 +36,6 @@ 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 ProcessingArtifact
from transcription.db.models import Source
from transcription.errors import AppError
from transcription.errors import ErrorCategory
@@ -52,17 +49,11 @@ from transcription.providers import TranscriptionProvider
from transcription.providers import TranscriptionResult
from transcription.providers import TransportEvidence
from transcription.providers import get_transcription_provider
from transcription.providers.evidence import canonical_json_bytes
from ..db.loading import defer
from ..db.loading import orm_attribute
from ..db.loading import selectinload
from .base import ServiceBase
from .normalization import ORIENTATION_PRODUCER
from .normalization import ORIENTATION_PRODUCER_VERSION
from .normalization import ORIENTATION_SCHEMA
from .normalization import ORIENTATION_SCHEMA_VERSION
from .normalization import normalize_orientation_async
from .source_media import lookup_source_mime_type
from .source_media import supported_source_formats
@@ -121,10 +112,20 @@ class ProviderInput:
digest_sha256: str
byte_size: int
media_type: str
width: int | None = None
height: int | None = None
derivative_id: UUID | None = None
transformation: str | None = None
def build_provider_input(source: Source) -> 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=Path(source.file_path),
digest_sha256=source.file_hash.lower(),
byte_size=source.file_size_bytes,
media_type=source_mime_type(source.file_path),
)
@dataclass(frozen=True, slots=True)
@@ -291,13 +292,10 @@ class SourceService(ServiceBase):
source = await self._read_source(
session=_session,
source_id=source_id,
options=(
selectinload(Source.job_sources),
selectinload(Source.processing_artifacts),
),
options=(selectinload(Source.job_sources),),
)
if source.job_sources or source.processing_artifacts:
if source.job_sources:
raise SourceDeleteBlockedError(
"Source delete blocked because retained execution evidence exists",
category=ErrorCategory.VALIDATION,
@@ -445,10 +443,7 @@ class SourceService(ServiceBase):
source = await self._read_source(
session=_session,
source_id=source_id,
options=(
selectinload(Source.job_sources),
selectinload(Source.processing_artifacts),
),
options=(selectinload(Source.job_sources),),
)
linked_job_sources = list(source.job_sources)
@@ -459,7 +454,7 @@ class SourceService(ServiceBase):
.where(ExecutionAttempt.source_id == source_id)
)
).one()
if source.processing_artifacts or attempt_count:
if attempt_count:
raise SourceDeleteBlockedError(
"Source delete blocked because immutable evidence exists",
category=ErrorCategory.VALIDATION,
@@ -533,7 +528,7 @@ class SourceService(ServiceBase):
provider: str | None = None,
model: str | None = None,
request_manifest: RequestManifest | None = None,
model_input_artifact_id: UUID | None = None,
quality_warnings: dict[str, JsonValue] | None = None,
transport_evidence: TransportEvidence | None = None,
failure_phase: str | None = None,
error_category: str | None = None,
@@ -566,6 +561,7 @@ class SourceService(ServiceBase):
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")
attempt_metadata = _merge_quality_warnings(metadata_payload, quality_warnings)
existing_job_source = await _session.exec(
select(JobSource).where(JobSource.job_id == job_id).where(JobSource.source_id == source_id)
@@ -626,7 +622,7 @@ class SourceService(ServiceBase):
router_request_id=transport.request_id,
router_generation_id=transport.generation_id,
sdk_response_snapshot=raw_response_payload,
normalized_metadata=metadata_payload,
normalized_metadata=attempt_metadata,
software_context=software_payload,
raw_transcription=text,
error_category=error_category,
@@ -641,30 +637,6 @@ class SourceService(ServiceBase):
_session.add(attempt)
await _session.flush()
manifest_derivative_id = (
request_manifest.source.derivative_id if request_manifest is not None else None
)
if (
model_input_artifact_id is not None
and manifest_derivative_id is not None
and model_input_artifact_id != manifest_derivative_id
):
raise TranscriptionError(
"Provider-input artifact does not match the request manifest",
category=ErrorCategory.CONFLICT,
suggestion="Persist the exact normalized input consumed by this attempt.",
)
derivative_id = model_input_artifact_id or manifest_derivative_id
if derivative_id is not None:
derivative = await _session.get(ProcessingArtifact, derivative_id)
if derivative is None or derivative.source_id != source_id:
raise TranscriptionError(
"Provider-input derivative evidence is missing or belongs to another Source",
category=ErrorCategory.CONFLICT,
suggestion="Restore the normalized input artifact before persisting the attempt.",
)
derivative.execution_attempt_id = attempt.id
if (
text is not None
and source.raw_transcription is None
@@ -716,7 +688,7 @@ class SourceService(ServiceBase):
) -> Sequence[ExecutionAttempt]:
"""List immutable execution evidence in stable attempt order."""
async with self._session_scope(session) as _session:
query = select(ExecutionAttempt).options(selectinload(ExecutionAttempt.artifacts))
query = select(ExecutionAttempt)
if source_id is not None:
query = query.where(ExecutionAttempt.source_id == source_id)
if job_id is not None:
@@ -729,289 +701,6 @@ class SourceService(ServiceBase):
)
return (await _session.exec(query)).all()
async def create_processing_artifact(
self,
artifact: ProcessingArtifact,
*,
session: AsyncSession | None = None,
) -> ProcessingArtifact:
"""Persist a validated generic processing artifact."""
if (artifact.inline_payload is None) == (artifact.external_reference is None):
raise TranscriptionError(
"Processing artifact requires exactly one content location",
category=ErrorCategory.VALIDATION,
suggestion="Provide inline JSON or one stable external reference, but not both.",
)
self._verify_artifact_integrity(artifact)
async with self._session_scope(session) as _session:
_session.add(artifact)
await self._finalize(session=_session, caller_session=session)
return artifact
async def create_json_artifact(
self,
*,
source_id: UUID,
execution_attempt_id: UUID | None,
artifact_type: str,
schema_name: str,
schema_version: str,
producer: str,
producer_version: str,
payload: dict[str, JsonValue],
coordinate_metadata: dict[str, JsonValue] | None = None,
session: AsyncSession | None = None,
) -> ProcessingArtifact:
"""Store canonical JSON inline or atomically in the constrained artifact root."""
if coordinate_metadata is not None:
required = {"units", "origin", "width", "height", "transformations"}
missing = required.difference(coordinate_metadata)
if missing:
raise TranscriptionError(
f"Coordinate metadata is missing required fields: {', '.join(sorted(missing))}",
category=ErrorCategory.VALIDATION,
suggestion="Declare units, origin, dimensions, and transformations.",
)
payload_bytes = canonical_json_bytes(payload)
artifact_id = uuid4()
inline_payload: dict[str, JsonValue] | None = payload
external_reference: str | None = None
external_path: Path | None = None
if len(payload_bytes) > self.settings.artifact_inline_threshold_bytes:
relative_path = Path(str(source_id)) / f"{artifact_id}.json"
external_path = self.settings.artifact_dir / relative_path
await asyncio.to_thread(self._write_external_artifact, path=external_path, content=payload_bytes)
inline_payload = None
external_reference = relative_path.as_posix()
artifact = ProcessingArtifact(
id=artifact_id,
source_id=source_id,
execution_attempt_id=execution_attempt_id,
artifact_type=artifact_type,
media_type="application/json",
schema_name=schema_name,
schema_version=schema_version,
producer=producer,
producer_version=producer_version,
inline_payload=inline_payload,
external_reference=external_reference,
payload_sha256=hashlib.sha256(payload_bytes).hexdigest(),
byte_size=len(payload_bytes),
coordinate_metadata=coordinate_metadata,
)
try:
return await self.create_processing_artifact(artifact, session=session)
except Exception:
if external_path is not None:
external_path.unlink(missing_ok=True)
raise
async def create_binary_artifact(
self,
*,
source_id: UUID,
artifact_type: str,
media_type: str,
schema_name: str,
schema_version: str,
producer: str,
producer_version: str,
content: bytes,
suffix: str,
coordinate_metadata: dict[str, JsonValue] | None = None,
session: AsyncSession | None = None,
) -> ProcessingArtifact:
"""Persist exact binary derivative bytes in application-managed artifact storage."""
artifact_id = uuid4()
safe_suffix = suffix if suffix.startswith(".") and suffix[1:].isalnum() else ".bin"
relative_path = Path(str(source_id)) / f"{artifact_id}{safe_suffix.lower()}"
external_path = self.settings.artifact_dir / relative_path
payload_sha256 = await asyncio.to_thread(self._write_and_digest_artifact, path=external_path, content=content)
artifact = ProcessingArtifact(
id=artifact_id,
source_id=source_id,
artifact_type=artifact_type,
media_type=media_type,
schema_name=schema_name,
schema_version=schema_version,
producer=producer,
producer_version=producer_version,
external_reference=relative_path.as_posix(),
payload_sha256=payload_sha256,
byte_size=len(content),
coordinate_metadata=coordinate_metadata,
)
try:
return await self.create_processing_artifact(artifact, session=session)
except Exception:
external_path.unlink(missing_ok=True)
raise
async def resolve_provider_input(
self,
source: Source,
*,
session: AsyncSession | None = None,
) -> ProviderInput:
"""Resolve original or physically orientation-normalized provider input."""
media_type = source_mime_type(source.file_path)
normalized = await normalize_orientation_async(source.file_path, media_type=media_type)
if normalized is None:
return ProviderInput(
path=Path(source.file_path),
digest_sha256=source.file_hash.lower(),
byte_size=source.file_size_bytes,
media_type=media_type,
)
metadata: dict[str, JsonValue] = {
"units": "pixels",
"origin": "top-left",
"width": normalized.derivative_width,
"height": normalized.derivative_height,
"transformations": [f"rotate-{normalized.applied_rotation_degrees}-degrees-clockwise"],
"source_id": str(source.id),
"original_digest_sha256": source.file_hash.lower(),
"original_byte_size": source.file_size_bytes,
"original_orientation": normalized.original_orientation,
"applied_rotation_degrees": normalized.applied_rotation_degrees,
"original_width": normalized.original_width,
"original_height": normalized.original_height,
"derivative_width": normalized.derivative_width,
"derivative_height": normalized.derivative_height,
"derivative_digest_sha256": normalized.digest_sha256,
"derivative_byte_size": len(normalized.content),
"original_media_type": media_type,
"derivative_media_type": normalized.media_type,
}
artifact = await self.create_binary_artifact(
source_id=source.id,
artifact_type="orientation_normalized_model_input",
media_type=normalized.media_type,
schema_name=ORIENTATION_SCHEMA,
schema_version=ORIENTATION_SCHEMA_VERSION,
producer=ORIENTATION_PRODUCER,
producer_version=ORIENTATION_PRODUCER_VERSION,
content=normalized.content,
suffix=normalized.suffix,
coordinate_metadata=metadata,
session=session,
)
if artifact.external_reference is None:
raise RuntimeError("Orientation derivative did not receive an external reference")
return ProviderInput(
path=self.settings.artifact_dir / artifact.external_reference,
digest_sha256=artifact.payload_sha256,
byte_size=artifact.byte_size,
media_type=artifact.media_type,
width=normalized.derivative_width,
height=normalized.derivative_height,
derivative_id=artifact.id,
transformation=f"{ORIENTATION_SCHEMA}@{ORIENTATION_SCHEMA_VERSION}",
)
def _write_and_digest_artifact(self, *, path: Path, content: bytes) -> str:
"""Persist artifact bytes and return their digest in one off-loop hop.
Binary derivatives are page-sized, so hashing them belongs in the same
worker thread as the write rather than on the event loop ([MED-01]).
"""
self._write_external_artifact(path=path, content=content)
return hashlib.sha256(content).hexdigest()
def _write_external_artifact(self, *, path: Path, content: bytes) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
temporary_path = path.with_suffix(f"{path.suffix}.tmp")
try:
with temporary_path.open("wb") as stream:
stream.write(content)
stream.flush()
os.fsync(stream.fileno())
temporary_path.replace(path)
except OSError as exc:
temporary_path.unlink(missing_ok=True)
raise TranscriptionError(
"Failed to persist external processing artifact",
category=ErrorCategory.INFRA_PERSISTENT,
suggestion="Verify artifact storage permissions and available disk space.",
) from exc
def _verify_external_artifact(self, artifact: ProcessingArtifact) -> None:
relative_path = Path(artifact.external_reference or "")
if relative_path.is_absolute() or ".." in relative_path.parts:
raise TranscriptionError(
"External artifact reference must stay inside the artifact root",
category=ErrorCategory.VALIDATION,
suggestion="Use a relative application-managed artifact reference.",
)
artifact_root = self.settings.artifact_dir.resolve()
artifact_path = (artifact_root / relative_path).resolve()
if artifact_root not in artifact_path.parents or not artifact_path.is_file():
raise TranscriptionError(
"External processing artifact is unavailable",
category=ErrorCategory.NOT_FOUND,
suggestion="Restore the artifact file or remove its pending database record.",
)
content = artifact_path.read_bytes()
if len(content) != artifact.byte_size or hashlib.sha256(content).hexdigest() != artifact.payload_sha256:
raise TranscriptionError(
"External processing artifact failed integrity verification",
category=ErrorCategory.CONFLICT,
suggestion="Restore the expected artifact bytes before retrying.",
)
def _verify_artifacts_integrity(self, artifacts: Sequence[ProcessingArtifact]) -> None:
"""Verify a batch of artifacts; hashing and file reads run off the event loop."""
for artifact in artifacts:
self._verify_artifact_integrity(artifact)
def _verify_artifact_integrity(self, artifact: ProcessingArtifact) -> None:
if artifact.inline_payload is None:
self._verify_external_artifact(artifact)
return
content = canonical_json_bytes(artifact.inline_payload)
if len(content) != artifact.byte_size or hashlib.sha256(content).hexdigest() != artifact.payload_sha256:
raise TranscriptionError(
"Inline processing artifact failed integrity verification",
category=ErrorCategory.CONFLICT,
suggestion="Recreate the artifact with its canonical payload digest and byte size.",
)
async def list_processing_artifacts(
self,
*,
source_id: UUID,
limit: int = 100,
session: AsyncSession | None = None,
) -> Sequence[ProcessingArtifact]:
"""List generic artifacts associated with a Source."""
async with self._session_scope(session) as _session:
query = (
select(ProcessingArtifact)
.where(ProcessingArtifact.source_id == source_id)
.order_by(col(ProcessingArtifact.created_at), col(ProcessingArtifact.id))
.limit(limit)
)
return (await _session.exec(query)).all()
async def list_processing_artifact_summaries(
self,
*,
source_id: UUID,
limit: int = 100,
session: AsyncSession | None = None,
) -> Sequence[ProcessingArtifact]:
"""List artifact metadata without loading potentially large inline payloads."""
async with self._session_scope(session) as _session:
query = (
select(ProcessingArtifact)
.options(defer(ProcessingArtifact.inline_payload))
.where(ProcessingArtifact.source_id == source_id)
.order_by(col(ProcessingArtifact.created_at), col(ProcessingArtifact.id))
.limit(limit)
)
return (await _session.exec(query)).all()
async def build_evidence_export(
self,
*,
@@ -1022,30 +711,7 @@ class SourceService(ServiceBase):
async with self._session_scope(session) as _session:
source = await self._read_source(session=_session, source_id=source_id)
attempts = list(await self.list_execution_attempts(source_id=source_id, session=_session))
artifacts = list(await self.list_processing_artifacts(source_id=source_id, session=_session))
await asyncio.to_thread(self._verify_artifacts_integrity, artifacts)
artifact_payloads = [
{
"id": str(artifact.id),
"source_id": str(artifact.source_id),
"execution_attempt_id": (str(artifact.execution_attempt_id) if artifact.execution_attempt_id else None),
"artifact_type": artifact.artifact_type,
"media_type": artifact.media_type,
"schema_name": artifact.schema_name,
"schema_version": artifact.schema_version,
"producer": artifact.producer,
"producer_version": artifact.producer_version,
"inline_payload": artifact.inline_payload,
"external_reference": artifact.external_reference,
"payload_sha256": artifact.payload_sha256,
"byte_size": artifact.byte_size,
"coordinate_metadata": artifact.coordinate_metadata,
"created_at": artifact.created_at.isoformat(),
}
for artifact in artifacts
]
attempt_payloads = [
{
"id": str(attempt.id),
@@ -1101,7 +767,6 @@ class SourceService(ServiceBase):
"upload_name": source.upload_name,
},
"attempts": attempt_payloads,
"artifacts": artifact_payloads,
}
async def upsert_revision_for_source(
@@ -1178,6 +843,23 @@ def _validate_transcription_metadata(
return validated.as_json_object()
def _merge_quality_warnings(
metadata: dict[str, JsonValue] | None,
quality_warnings: dict[str, JsonValue] | None,
) -> dict[str, JsonValue] | None:
"""Attach app-computed quality warnings to provider-normalized metadata.
The warnings are derived from the transcription text rather than reported by
the provider, so they are namespaced under their own key instead of being
mixed into the provider's own fields.
"""
if quality_warnings is None:
return metadata
merged: dict[str, JsonValue] = dict(metadata or {})
merged["transcription_quality_warnings"] = quality_warnings
return merged
def _validate_json_object(
payload: dict[str, JsonValue] | None,
*,
src/transcription/services/store.py
+42 -19
View File
@@ -25,8 +25,10 @@ from ..db.session import SessionFactory
from ..db.session import session_scope
from .media_storage import build_stored_filename
from .media_storage import write_media_bytes
from .normalization import normalize_orientation_async
from .sources import TranscriptionError
from .sources import build_prompt_execution
from .sources import source_mime_type
from .sources import validate_source_content
logger = logging.getLogger(__name__)
@@ -55,6 +57,15 @@ class DocumentJobResult:
original_filename: str
@dataclass(frozen=True)
class StoredSourceFile:
"""A persisted Source file and the identity of the bytes actually stored."""
path: Path
file_hash: str
file_size_bytes: int
@dataclass(frozen=True)
class PendingStoredSource:
"""Pre-staged Source artifact tied to a Source id."""
@@ -83,14 +94,14 @@ async def create_document_job(
prompt_execution = build_prompt_execution(settings=runtime_settings)
document_id = uuid4()
source_id = uuid4()
stored_path = await store_source_file(
stored = await store_source_file(
filename=filename,
file_bytes=file_bytes,
settings=runtime_settings,
relative_directory=Path("documents") / str(document_id),
filename_stem=str(source_id),
)
file_hash, file_size_bytes = _compute_file_metadata(file_bytes)
stored_path = stored.path
try:
async with session_scope(
session_factory=session_factory,
@@ -103,8 +114,8 @@ async def create_document_job(
source_id=source_id,
original_filename=filename,
stored_path=stored_path,
file_hash=file_hash,
file_size_bytes=file_size_bytes,
file_hash=stored.file_hash,
file_size_bytes=stored.file_size_bytes,
prompt_execution=prompt_execution,
)
except Exception as exc:
@@ -153,7 +164,7 @@ async def create_job_for_document(
stored_sources: list[PendingStoredSource] = []
for filename, file_bytes in sorted_source_files:
source_id = uuid4()
stored_path = await store_source_file(
stored = await store_source_file(
filename=filename,
file_bytes=file_bytes,
settings=runtime_settings,
@@ -164,9 +175,9 @@ async def create_job_for_document(
PendingStoredSource(
source_id=source_id,
original_filename=filename,
stored_path=stored_path,
file_hash=_compute_file_hash(file_bytes),
file_size_bytes=len(file_bytes),
stored_path=stored.path,
file_hash=stored.file_hash,
file_size_bytes=stored.file_size_bytes,
)
)
@@ -330,14 +341,6 @@ def _best_effort_delete(path: Path) -> None:
logger.warning("Failed to clean up Source file after database error: %s", path)
def _compute_file_hash(file_bytes: bytes) -> str:
return hashlib.sha256(file_bytes).hexdigest()
def _compute_file_metadata(file_bytes: bytes) -> tuple[str, int]:
return _compute_file_hash(file_bytes), len(file_bytes)
async def store_source_file(
*,
filename: str,
@@ -345,8 +348,13 @@ async def store_source_file(
settings: Settings | None = None,
relative_directory: Path | None = None,
filename_stem: str | None = None,
) -> Path:
"""Validate and persist a Source file to configured media storage."""
) -> StoredSourceFile:
"""Validate, orient, and persist a Source file to configured media storage.
Orientation is applied here, at the ingest boundary, so the stored bytes are
already upright and the hash and byte size recorded on the ``Source`` row
describe exactly what is on disk and exactly what a provider is later sent.
"""
runtime_settings = settings or get_settings()
try:
validate_source_content(filename=filename, content=file_bytes)
@@ -358,8 +366,18 @@ async def store_source_file(
retriable=exc.retriable,
) from exc
normalized = await normalize_orientation_async(file_bytes, media_type=source_mime_type(filename))
if normalized is not None:
logger.info(
"Normalized Source orientation on ingest: %s (orientation=%s, rotation=%s)",
Path(filename).name,
normalized.original_orientation,
normalized.applied_rotation_degrees,
)
file_bytes = normalized.content
upload_dir = runtime_settings.upload_dir
return await write_media_bytes(
stored_path = await write_media_bytes(
target_dir=upload_dir if relative_directory is None else upload_dir / relative_directory,
stored_name=build_stored_filename(filename=filename, filename_stem=filename_stem),
file_bytes=file_bytes,
@@ -368,3 +386,8 @@ async def store_source_file(
failure_suggestion="Check upload directory permissions and available disk space, then retry.",
log_label="Source file",
)
return StoredSourceFile(
path=stored_path,
file_hash=hashlib.sha256(file_bytes).hexdigest(),
file_size_bytes=len(file_bytes),
)
src/transcription/services/workflows.py
+3 -46
View File
@@ -4,7 +4,6 @@ import logging
from dataclasses import dataclass
from datetime import UTC
from datetime import datetime
from uuid import UUID
from sqlmodel.ext.asyncio.session import AsyncSession
@@ -32,14 +31,11 @@ from . import ServiceBundle
from .documents import DocumentService
from .people import DocumentPersonInput
from .people import PeopleService
from .quality import QUALITY_ANALYSIS_PRODUCER
from .quality import QUALITY_ANALYSIS_PRODUCER_VERSION
from .quality import QUALITY_ANALYSIS_SCHEMA
from .quality import QUALITY_ANALYSIS_VERSION
from .quality import analyze_transcription_quality
from .quality import quality_warning_payload
from .sources import PromptExecution
from .sources import build_prompt_execution
from .sources import build_provider_input
from .sources import hash_prompt_text
from .sources import transcribe_document_image
@@ -121,7 +117,6 @@ class _SuccessfulPage:
started_at: datetime
finished_at: datetime
duration_ms: int
model_input_artifact_id: UUID | None = None
@dataclass(frozen=True)
@@ -138,7 +133,6 @@ class _FailedPage:
normalized_metadata: dict | None = None
provider: str | None = None
model: str | None = None
model_input_artifact_id: UUID | None = None
async def advance_job(
@@ -223,19 +217,13 @@ async def process_queued_job( # noqa: PLR0915
provider_input = None
page_outcome: _SuccessfulPage | _FailedPage
try:
provider_input = await services.sources.resolve_provider_input(source, session=session)
if session is not None:
await session.commit()
provider_input = build_provider_input(source)
source_reference = SourceEvidenceReference(
source_id=source.id,
digest_sha256=provider_input.digest_sha256,
byte_size=provider_input.byte_size,
media_type=provider_input.media_type,
page_number=source.page_number,
width=provider_input.width,
height=provider_input.height,
derivative_id=provider_input.derivative_id,
transformation=provider_input.transformation,
)
result = await asyncio.wait_for(
_call_transcriber(
@@ -276,7 +264,6 @@ async def process_queued_job( # noqa: PLR0915
started_at=started_at,
finished_at=finished_at,
duration_ms=max(0, int(elapsed_seconds * 1000)),
model_input_artifact_id=provider_input.derivative_id,
)
successful_pages.append(page_outcome)
except TimeoutError:
@@ -299,9 +286,6 @@ async def process_queued_job( # noqa: PLR0915
request_manifest=services.sources.provider.current_request_manifest,
transport_evidence=services.sources.provider.current_transport_evidence,
failure_phase="local_timeout",
model_input_artifact_id=(
provider_input.derivative_id if provider_input is not None else None
),
)
failed_pages.append(page_outcome)
logger.error(
@@ -356,9 +340,6 @@ async def process_queued_job( # noqa: PLR0915
normalized_metadata=result.metadata_payload() if result is not None else None,
provider=result.provider if result is not None else None,
model=result.model if result is not None else None,
model_input_artifact_id=(
provider_input.derivative_id if provider_input is not None else None
),
)
failed_pages.append(page_outcome)
logger.error(
@@ -539,36 +520,13 @@ async def _write_page_outcome(
provider=result.provider,
model=result.model,
request_manifest=result.request_manifest,
model_input_artifact_id=page.model_input_artifact_id,
quality_warnings=quality_warning_payload(analyze_transcription_quality(result.text)),
transport_evidence=result.transport_evidence,
started_at=page.started_at,
finished_at=page.finished_at,
duration_ms=page.duration_ms,
session=session,
)
job_source = await services.sources.read_job_source_for_job(
job_id=job.id,
source_id=source.id,
session=session,
)
attempt = await services.sources.read_latest_execution_attempt(
job_source_id=job_source.id,
session=session,
)
if attempt is None:
raise RuntimeError("Successful transcription did not create execution evidence")
warnings = analyze_transcription_quality(result.text)
await services.sources.create_json_artifact(
source_id=source.id,
execution_attempt_id=attempt.attempt.id,
artifact_type="transcription_quality_warnings",
schema_name=QUALITY_ANALYSIS_SCHEMA,
schema_version=QUALITY_ANALYSIS_VERSION,
producer=QUALITY_ANALYSIS_PRODUCER,
producer_version=QUALITY_ANALYSIS_PRODUCER_VERSION,
payload=quality_warning_payload(warnings),
session=session,
)
return
await services.sources.update_job_source_transcription(
@@ -581,7 +539,6 @@ async def _write_page_outcome(
provider=page.provider,
model=page.model,
request_manifest=page.request_manifest,
model_input_artifact_id=page.model_input_artifact_id,
transport_evidence=page.transport_evidence,
failure_phase=page.failure_phase,
error_category=page.error.category.value,
src/transcription/ui/pages/sources_page.py
+8 -40
View File
@@ -12,7 +12,6 @@ from nicegui import ui
from transcription.config import Settings
from transcription.db.models import ExecutionAttempt
from transcription.db.models import JobSource
from transcription.db.models import ProcessingArtifact
from transcription.db.models import Source
from transcription.services.sources import LatestExecutionAttempt
from transcription.services.sources import SourceDeleteBlockedError
@@ -127,9 +126,6 @@ def register_page() -> None: # noqa: PLR0915
if latest_job_source is not None
else None
)
source_artifacts = list(
await sources_service.list_processing_artifact_summaries(source_id=parsed_source_id)
)
attempts = list(await sources_service.list_execution_attempts(source_id=parsed_source_id))
except TranscriptionNotFoundError:
render_record_not_found("Source")
@@ -171,12 +167,6 @@ def register_page() -> None: # noqa: PLR0915
icon="delete",
extra_classes="text-xs",
)
if any(
artifact.artifact_type == "orientation_normalized_model_input"
for artifact in source_artifacts
):
archival_badge("Orientation normalized")
with ui.grid().classes("w-full grid-cols-12 gap-4"):
with ui.column().classes("col-span-12 lg:col-span-4 gap-2"):
_render_source_navigation(navigation.previous_id, navigation.next_id)
@@ -200,7 +190,6 @@ def register_page() -> None: # noqa: PLR0915
source=source,
latest_job_source=latest_job_source,
latest_attempt=latest_attempt,
source_artifacts=source_artifacts,
)
@ui.page("/sources/{source_id}/delete")
@@ -311,14 +300,12 @@ def _render_source_metadata_column(
source: Source,
latest_job_source: JobSource | None,
latest_attempt: LatestExecutionAttempt | None,
source_artifacts: list[ProcessingArtifact],
) -> None:
with ui.column().classes("col-span-12 lg:col-span-4 gap-4"):
_render_source_metadata_zone(source)
_render_source_job_metadata_zone(
latest_job_source,
latest_attempt=latest_attempt,
source_artifacts=source_artifacts,
)
_render_source_revision_logistics_zone(source)
@@ -337,7 +324,6 @@ def _render_source_job_metadata_zone(
latest_job_source: JobSource | None,
*,
latest_attempt: LatestExecutionAttempt | None,
source_artifacts: list[ProcessingArtifact],
) -> None:
with archival_card(title="SourceJob Metadata"):
if latest_job_source is None:
@@ -374,7 +360,6 @@ def _render_source_job_metadata_zone(
_render_provider_evidence(
latest_job_source,
latest_attempt=latest_attempt,
source_artifacts=source_artifacts,
)
@@ -382,7 +367,6 @@ def _render_provider_evidence(
job_source: JobSource,
*,
latest_attempt: LatestExecutionAttempt | None,
source_artifacts: list[ProcessingArtifact],
) -> None:
ui.label("Provider Evidence").classes("text-xs font-semibold ui-text-primary mt-3")
if latest_attempt is None:
@@ -392,7 +376,6 @@ def _render_provider_evidence(
"OpenRouter SDK Response Snapshot (Raw API Response compatibility field)",
job_source.raw_api_response,
)
_render_json_evidence("Derived Artifacts", _artifact_display(source_artifacts))
return
attempt = latest_attempt.attempt
@@ -403,22 +386,6 @@ def _render_provider_evidence(
_render_json_evidence("OpenRouter SDK Response Snapshot", attempt.sdk_response_snapshot)
_render_json_evidence("Normalized Metadata", attempt.normalized_metadata)
_render_json_evidence("Software Context", attempt.software_context)
_render_json_evidence("Derived Artifacts", _artifact_display(source_artifacts))
def _artifact_display(artifacts: list[ProcessingArtifact]) -> list[dict[str, object]] | None:
payload: list[dict[str, object]] = [
{
"id": str(artifact.id),
"type": artifact.artifact_type,
"format": artifact.media_type,
"schema": f"{artifact.schema_name}@{artifact.schema_version}",
"digest_sha256": artifact.payload_sha256,
"coordinate_metadata": artifact.coordinate_metadata,
}
for artifact in artifacts
]
return payload or None
def _transport_display(latest_attempt: LatestExecutionAttempt) -> dict[str, object]:
@@ -657,13 +624,14 @@ def _attempt_warning_count(attempt: ExecutionAttempt) -> int:
def _attempt_warnings(attempt: ExecutionAttempt) -> list[dict[str, object]]:
for artifact in attempt.artifacts:
if artifact.artifact_type != "transcription_quality_warnings" or artifact.inline_payload is None:
continue
warnings = artifact.inline_payload.get("warnings")
if isinstance(warnings, list):
return [warning for warning in warnings if isinstance(warning, dict)]
return []
metadata = attempt.normalized_metadata or {}
payload = metadata.get("transcription_quality_warnings")
if not isinstance(payload, dict):
return []
warnings = payload.get("warnings")
if not isinstance(warnings, list):
return []
return [warning for warning in warnings if isinstance(warning, dict)]
def _render_attempt_warnings(attempt: ExecutionAttempt, *, label: str) -> None: