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

V4.5 Complete - Enhanced trancription context, added option to restranscribe source under different models.

Browse Source
This commit is contained in:
Jim Lancaster
2026-08-16 09:06:56 -05:00
parent bdb1b31b0a
commit 7054cd8af9
28 changed files with 2883 additions and 1485 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/test_pipeline_flow.py
+17 -9
View File
@@ -1,9 +1,11 @@
"""Integration tests for end-to-end upload and worker pipeline behavior."""
import io
from pathlib import Path
from uuid import uuid4
import pytest
from PIL import Image
from transcription.config import Settings
from transcription.db.models import Document
@@ -18,6 +20,12 @@ from transcription.services.store import create_job_for_document
from transcription.services.workflows import advance_job
def _jpeg_bytes(color: str = "white") -> bytes:
output = io.BytesIO()
Image.new("RGB", (2, 2), color=color).save(output, format="JPEG")
return output.getvalue()
def _build_services(default_session_factory) -> ServiceBundle:
services = ServiceBundle()
object.__setattr__(
@@ -56,7 +64,7 @@ class TestPipelineSuccessFlow:
)
upload_result = await create_document_job(
filename="pipeline.jpg",
file_bytes=b"pipeline-bytes",
file_bytes=_jpeg_bytes(),
session=async_session,
settings=settings,
)
@@ -141,9 +149,9 @@ class TestPipelineSuccessFlow:
create_result = await create_job_for_document(
document_id=document.id,
source_files=[
("page-01.jpg", b"one"),
("page-02.jpg", b"two"),
("page-03.jpg", b"three"),
("page-01.jpg", _jpeg_bytes("white")),
("page-02.jpg", _jpeg_bytes("gray")),
("page-03.jpg", _jpeg_bytes("black")),
],
session=async_session,
settings=settings,
@@ -206,8 +214,8 @@ class TestPipelineSuccessFlow:
create_result = await create_job_for_document(
document_id=document.id,
source_files=[
("page-01.jpg", b"one"),
("page-02.jpg", b"two"),
("page-01.jpg", _jpeg_bytes("white")),
("page-02.jpg", _jpeg_bytes("gray")),
],
session=async_session,
settings=settings,
@@ -272,8 +280,8 @@ class TestPipelineSuccessFlow:
create_result = await create_job_for_document(
document_id=document.id,
source_files=[
("page-01.jpg", b"one"),
("page-02.jpg", b"two"),
("page-01.jpg", _jpeg_bytes("white")),
("page-02.jpg", _jpeg_bytes("gray")),
],
session=async_session,
settings=settings,
@@ -346,7 +354,7 @@ class TestPipelineFailureFlow:
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
upload_result = await create_document_job(
filename="pipeline.jpg",
file_bytes=b"pipeline-bytes",
file_bytes=_jpeg_bytes(),
session=async_session,
settings=settings,
)
tests/services/test_normalization.py
+228
View File
@@ -0,0 +1,228 @@
"""Tests for V4.5 metadata-directed orientation normalization."""
import hashlib
import io
from pathlib import Path
import pytest
from PIL import Image
from transcription.config import Settings
from transcription.db.models import Document
from transcription.db.models import Job
from transcription.db.models import JobSource
from transcription.db.models import Source
from transcription.providers import RequestManifest
from transcription.providers import TranscriptionResult
from transcription.providers.evidence import build_software_context
from transcription.services import ServiceBundle
from transcription.services.documents import DocumentService
from transcription.services.jobs import JobService
from transcription.services.normalization import normalize_orientation
from transcription.services.sources import SourceService
from transcription.services.workflows import process_queued_job
def _write_oriented_jpeg(path: Path, *, orientation: int) -> bytes:
image = Image.new("RGB", (2, 3))
image.putdata(
[
(255, 0, 0),
(255, 0, 0),
(0, 255, 0),
(0, 255, 0),
(0, 0, 255),
(0, 0, 255),
]
)
exif = Image.Exif()
exif[274] = orientation
image.save(path, format="JPEG", quality=100, subsampling=0, exif=exif)
return path.read_bytes()
def _write_oriented_image(path: Path, *, orientation: int, image_format: str) -> None:
image = Image.new("RGB", (2, 3), color="white")
exif = Image.Exif()
exif[274] = orientation
image.save(path, format=image_format, exif=exif)
@pytest.mark.unit
def test_orientation_three_is_physically_rotated_and_metadata_removed(tmp_path):
path = tmp_path / "upside-down.jpg"
original = _write_oriented_jpeg(path, orientation=3)
result = normalize_orientation(path, media_type="image/jpeg")
assert result is not None
assert result.applied_rotation_degrees == 180
assert path.read_bytes() == original
with Image.open(path) as source_image, Image.open(io.BytesIO(result.content)) as derivative:
assert source_image.getexif()[274] == 3
assert derivative.getexif().get(274, 1) == 1
assert derivative.getpixel((0, 0))[2] > derivative.getpixel((0, 0))[0]
@pytest.mark.unit
def test_orientation_one_is_noop(tmp_path):
path = tmp_path / "upright.jpg"
_write_oriented_jpeg(path, orientation=1)
assert normalize_orientation(path, media_type="image/jpeg") is None
@pytest.mark.parametrize(
("filename", "image_format", "media_type"),
[
("oriented.png", "PNG", "image/png"),
("oriented.tiff", "TIFF", "image/tiff"),
],
)
def test_supported_non_jpeg_orientation_is_normalized(
tmp_path,
filename,
image_format,
media_type,
):
path = tmp_path / filename
_write_oriented_image(path, orientation=6, image_format=image_format)
result = normalize_orientation(path, media_type=media_type)
assert result is not None
assert result.applied_rotation_degrees == 90
assert (result.derivative_width, result.derivative_height) == (3, 2)
with Image.open(io.BytesIO(result.content)) as derivative:
assert derivative.getexif().get(274, 1) == 1
@pytest.mark.integration
@pytest.mark.asyncio
async def test_resolve_provider_input_persists_exact_derivative(default_session_factory, tmp_path):
source_path = tmp_path / "source.jpg"
original = _write_oriented_jpeg(source_path, orientation=3)
settings = Settings(
openrouter_api_key="test-key",
artifact_dir=tmp_path / "artifacts",
provider_models=None,
)
documents = DocumentService(session_factory=default_session_factory, settings=settings)
sources = SourceService(session_factory=default_session_factory, settings=settings)
document = await documents.create_document(Document(name="Oriented"))
source = await sources.create_source(
Source(
document_id=document.id,
page_number=1,
upload_name="source.jpg",
filename="source.jpg",
file_path=str(source_path),
file_hash=hashlib.sha256(original).hexdigest(),
file_size_bytes=len(original),
)
)
provider_input = await sources.resolve_provider_input(source)
artifacts = await sources.list_processing_artifacts(source_id=source.id)
assert source_path.read_bytes() == original
assert provider_input.derivative_id == artifacts[0].id
assert provider_input.path.read_bytes() != original
assert hashlib.sha256(provider_input.path.read_bytes()).hexdigest() == provider_input.digest_sha256
assert artifacts[0].coordinate_metadata["original_orientation"] == 3
@pytest.mark.integration
@pytest.mark.asyncio
async def test_worker_sends_exact_derivative_and_links_attempt_evidence(
default_session_factory,
tmp_path,
monkeypatch,
):
source_path = tmp_path / "source.jpg"
original = _write_oriented_jpeg(source_path, orientation=3)
prompt_dir = tmp_path / "prompts"
prompt_dir.mkdir()
(prompt_dir / "transcribe_document.md").write_text("Transcribe verbatim.", encoding="utf-8")
settings = Settings(
openrouter_api_key="test-key",
artifact_dir=tmp_path / "artifacts",
prompt_dir=prompt_dir,
provider_models=None,
)
services = ServiceBundle(
documents=DocumentService(session_factory=default_session_factory, settings=settings),
jobs=JobService(session_factory=default_session_factory, settings=settings),
sources=SourceService(session_factory=default_session_factory, settings=settings),
)
document = await services.documents.create_document(Document(name="Pipeline"))
source = await services.sources.create_source(
Source(
document_id=document.id,
page_number=1,
upload_name="source.jpg",
filename="source.jpg",
file_path=str(source_path),
file_hash=hashlib.sha256(original).hexdigest(),
file_size_bytes=len(original),
)
)
job = await services.jobs.create_job(Job(document_id=document.id))
await services.sources.create_job_source(JobSource(job_id=job.id, source_id=source.id))
loaded = await services.jobs.read_job(job.id)
captured: dict[str, object] = {}
async def fake_transcribe(
image_path,
*,
prompt_name,
prompt_text,
temperature,
top_p,
settings,
provider,
source_reference,
requested_model,
):
_ = (prompt_name, temperature, top_p, settings, provider, requested_model)
image_bytes = Path(image_path).read_bytes()
captured["bytes"] = image_bytes
captured["source_reference"] = source_reference
manifest = RequestManifest(
provider="fixture",
requested_model="fixture/model",
request={"model": "fixture/model"},
source=source_reference,
optional_parameter_states={"temperature": "omitted", "top_p": "omitted"},
prompt_content=prompt_text,
prompt_sha256=hashlib.sha256(prompt_text.encode()).hexdigest(),
timeout_seconds=20,
retry_policy="none",
software=build_software_context(
adapter_name="fixture",
adapter_version="1",
client_library="transcription",
),
)
return TranscriptionResult(
text="[document body typewritten]\nDamaged \ufffd text",
provider="fixture",
model="fixture/model",
request_manifest=manifest,
)
monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", fake_transcribe)
await process_queued_job(job=loaded, services=services, settings=settings)
attempts = await services.sources.list_execution_attempts(source_id=source.id)
artifacts = await services.sources.list_processing_artifacts(source_id=source.id)
source_reference = captured["source_reference"]
assert source_path.read_bytes() == original
assert hashlib.sha256(captured["bytes"]).hexdigest() == source_reference.digest_sha256
assert source_reference.derivative_id is not None
assert {artifact.artifact_type for artifact in artifacts} == {
"orientation_normalized_model_input",
"transcription_quality_warnings",
}
assert {artifact.execution_attempt_id for artifact in artifacts} == {attempts[0].id}
tests/services/test_quality.py
+44
View File
@@ -0,0 +1,44 @@
"""Tests for deterministic V4.5 transcription warnings."""
import pytest
from transcription.services.quality import QualityWarningCode
from transcription.services.quality import analyze_transcription_quality
from transcription.services.quality import quality_warning_payload
@pytest.mark.unit
def test_quality_analysis_reports_each_v45_warning_without_mutating_text():
text = (
"[document body handwritten]\n"
"[document body typewritten]\n"
"[handwritten: first line]\n"
"[handwritten: second line]\n"
"Damaged \ufffd text &"
)
warnings = analyze_transcription_quality(text)
assert [warning.code for warning in warnings] == [
QualityWarningCode.REPLACEMENT_CHARACTER,
QualityWarningCode.MULTIPLE_BODY_MARKERS,
QualityWarningCode.REDUNDANT_HANDWRITING_WRAPPERS,
QualityWarningCode.UNRESOLVED_HTML_ENTITY,
]
assert text.endswith("&")
@pytest.mark.unit
def test_quality_analysis_accepts_clean_transcription():
assert analyze_transcription_quality("[document body typewritten]\nClean text.") == ()
@pytest.mark.unit
def test_quality_warning_payload_is_versioned():
payload = quality_warning_payload(
analyze_transcription_quality("[document body typeset]\nBroken \ufffd")
)
assert payload["schema_name"] == "transcription.quality-warnings"
assert payload["schema_version"] == "1"
assert payload["warnings"][0]["code"] == "replacement_character"
tests/services/test_v45_candidates.py
+130
View File
@@ -0,0 +1,130 @@
"""V4.5 retranscription candidate and promotion tests."""
from uuid import uuid4
import pytest
from transcription.config import Settings
from transcription.db.models import Document
from transcription.db.models import Job
from transcription.db.models import JobPurpose
from transcription.db.models import JobSource
from transcription.db.models import Source
from transcription.services import ServiceBundle
from transcription.services.documents import DocumentService
from transcription.services.jobs import JobService
from transcription.services.sources import CandidatePromotionError
from transcription.services.sources import SourceService
from transcription.services.workflows import create_source_retranscription_job
def _services(default_session_factory, settings: Settings) -> ServiceBundle:
return ServiceBundle(
documents=DocumentService(session_factory=default_session_factory, settings=settings),
jobs=JobService(session_factory=default_session_factory, settings=settings),
sources=SourceService(session_factory=default_session_factory, settings=settings),
)
async def _seed_source(services: ServiceBundle) -> Source:
document = await services.documents.create_document(Document(name="V4.5 source"))
source = Source(
document_id=document.id,
page_number=1,
upload_name="page.jpg",
filename="page.jpg",
file_path="page.jpg",
file_hash="a" * 64,
file_size_bytes=1,
revised_text="human revision",
)
return await services.sources.create_source(source)
@pytest.mark.integration
@pytest.mark.asyncio
async def test_first_success_is_preferred_and_later_success_remains_candidate(default_session_factory):
settings = Settings(openrouter_api_key="test-key", provider_models=None)
services = _services(default_session_factory, settings)
source = await _seed_source(services)
first_job = await services.jobs.create_job(Job(document_id=source.document_id))
second_job = await services.jobs.create_job(Job(document_id=source.document_id))
await services.sources.create_job_source(JobSource(job_id=first_job.id, source_id=source.id))
await services.sources.create_job_source(JobSource(job_id=second_job.id, source_id=source.id))
await services.sources.update_job_source_transcription(
job_id=first_job.id,
source_id=source.id,
text="first result",
provider="fixture",
model="model-a",
)
selected = await services.sources.read_source(source.id)
first_attempt_id = selected.preferred_execution_attempt_id
await services.sources.update_job_source_transcription(
job_id=second_job.id,
source_id=source.id,
text="candidate result",
provider="fixture",
model="model-b",
)
unchanged = await services.sources.read_source(source.id)
attempts = await services.sources.list_execution_attempts(source_id=source.id)
assert unchanged.raw_transcription == "first result"
assert unchanged.preferred_execution_attempt_id == first_attempt_id
assert {attempt.raw_transcription for attempt in attempts} == {"first result", "candidate result"}
candidate = next(attempt for attempt in attempts if attempt.raw_transcription == "candidate result")
promoted = await services.sources.promote_machine_attempt(
source_id=source.id,
execution_attempt_id=candidate.id,
)
assert promoted.raw_transcription == "candidate result"
assert promoted.preferred_execution_attempt_id == candidate.id
assert promoted.revised_text == "human revision"
@pytest.mark.integration
@pytest.mark.asyncio
async def test_promotion_rejects_unrelated_attempt(default_session_factory):
settings = Settings(openrouter_api_key="test-key", provider_models=None)
services = _services(default_session_factory, settings)
source = await _seed_source(services)
with pytest.raises(CandidatePromotionError):
await services.sources.promote_machine_attempt(
source_id=source.id,
execution_attempt_id=uuid4(),
)
@pytest.mark.integration
@pytest.mark.asyncio
async def test_retranscription_job_locks_source_and_frozen_model(default_session_factory, tmp_path):
settings = Settings(
openrouter_api_key="test-key",
prompt_dir=tmp_path,
provider_model="vendor/default",
provider_models=["vendor/alternate"],
)
(tmp_path / settings.default_prompt_name).write_text("Transcribe verbatim.", encoding="utf-8")
services = _services(default_session_factory, settings)
source = await _seed_source(services)
job = await create_source_retranscription_job(
source_id=source.id,
model="vendor/alternate",
services=services,
settings=settings,
)
loaded = await services.jobs.read_job(job.id)
assert loaded.purpose == JobPurpose.RETRANSCRIPTION
assert loaded.document_id == source.document_id
assert loaded.provider == "openrouter"
assert loaded.model == "vendor/alternate"
assert loaded.user_prompt == "Transcribe verbatim."
assert [link.source_id for link in loaded.job_sources] == [source.id]
tests/test_config.py
+26 -1
View File
@@ -12,7 +12,7 @@ from transcription.config import parse_cli_settings
def _make_settings(**overrides) -> Settings:
"""Build a Settings instance with a dummy API key unless overridden."""
defaults = {"openrouter_api_key": "test-key-abc123"}
defaults = {"openrouter_api_key": "test-key-abc123", "provider_models": None}
defaults.update(overrides)
return Settings(**defaults)
@@ -106,6 +106,31 @@ class TestProviderSettings:
settings = Settings(openrouter_api_key="test-key-abc123")
assert settings.provider_model == "google/gemini-2.5-flash"
def test_provider_models_defaults_to_default_model(self):
settings = _make_settings(provider_model="vendor/default")
assert settings.provider_models == ("vendor/default",)
def test_provider_models_are_default_first_trimmed_and_deduplicated(self):
settings = _make_settings(
provider_model=" vendor/default ",
provider_models=["vendor/alternate", "vendor/default", " vendor/other "],
)
assert settings.provider_models == ("vendor/default", "vendor/alternate", "vendor/other")
def test_provider_models_rejects_empty_list(self):
with pytest.raises(ValidationError):
_make_settings(provider_models=[])
def test_provider_models_loads_json_from_environment(self, monkeypatch):
monkeypatch.setenv("PROVIDER_MODEL", "vendor/default")
monkeypatch.setenv("PROVIDER_MODELS", '["vendor/alternate","vendor/default"]')
settings = Settings(openrouter_api_key="test-key-abc123")
assert settings.provider_models == ("vendor/default", "vendor/alternate")
class TestPathSettings:
"""Verify filesystem path field types."""
tests/test_db.py
+12
View File
@@ -172,11 +172,23 @@ async def test_v42_upgrade_adds_evidence_tables_without_rewriting_legacy_snapsho
await upgrade_schema(engine=runtime.engine)
async with runtime.engine.connect() as connection:
table_names = set(await connection.run_sync(lambda c: inspect(c).get_table_names()))
job_columns = set(
await connection.run_sync(
lambda c: tuple(column["name"] for column in inspect(c).get_columns("job"))
)
)
source_columns = set(
await connection.run_sync(
lambda c: tuple(column["name"] for column in inspect(c).get_columns("source"))
)
)
legacy_snapshot = (
await connection.execute(text("SELECT raw_api_response FROM job_source WHERE id = 'link-1'"))
).scalar_one()
assert {"execution_attempt", "processing_artifact"}.issubset(table_names)
assert "purpose" in job_columns
assert "preferred_execution_attempt_id" in source_columns
assert "legacy" in legacy_snapshot
finally:
await dispose_database_runtime()
tests/test_prompts.py
+17
View File
@@ -50,6 +50,23 @@ class TestPromptArtifact:
assert "[deleted:" in text
assert "[inserted:" in text
def test_prompt_defines_exactly_one_body_medium_marker(self):
text = _prompt_text().lower()
for marker in (
"[document body handwritten]",
"[document body typewritten]",
"[document body typeset]",
"[document body mixed]",
):
assert marker in text
assert "exactly one" in text
assert "typewriter defects are not handwriting" in text
def test_prompt_preserves_structured_layout_associations(self):
text = _prompt_text().lower()
for phrase in ("tables of contents", "dotted-leader", "page-reference", "tables and forms", "columns"):
assert phrase in text
class TestPromptConfiguration:
def test_builds_validated_immutable_prompt_provenance(self, tmp_path):