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

Scrub references to older versions
Quality Gate / gate (push) Successful in 2m39s
Details

Browse Source
This commit is contained in:
Jim Lancaster
2026-09-02 17:02:48 -05:00
parent eeb1888aa3
commit 065acad125
20 changed files with 70 additions and 545 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/migrate_v46_to_v47.py
-337
View File
@@ -1,337 +0,0 @@
"""One-time migration of a V4.6 database into the V4.7 schema.
V4.7 is an architectural cleanup: no new user-facing behaviour, but three
structural changes plus a one-time image backfill. This tool carries all of
them, and is built up phase by phase so the live database stays usable at
every phase boundary.
Steps, in execution order:
1. Rotate every stored Source image that still carries a supported EXIF
orientation, in place, and update ``source.file_hash`` and
``source.file_size_bytes`` to describe the rewritten file.
2. Drop the ``processing_artifact`` table and delete its external files.
3. Rewrite ``execution_attempt.status`` from enum *names* to enum *values*, so
it compares equal to ``job_source.status`` (defect [45]).
4. Drop the five evidence columns from ``job_source``, leaving it a pure work
queue of ``id``, ``job_id``, ``source_id`` and ``status``.
Design notes:
- The image rewrite reuses the application's own
:func:`~transcription.services.normalization.normalize_orientation`, so the
backfilled bytes are byte-identical to what ingest would now produce. It
reuses the source quantization tables and subsampling rather than
re-quantizing, which is both smaller and higher fidelity than a fixed
quality setting.
- The hash and size are rewritten alongside the file. After V4.7 the evidence
digest is derived straight from ``source.file_hash``, so leaving it
describing the pre-rotation bytes would silently invalidate every future
export.
- The database is read and written through SQLAlchemy Core against the live
metadata, so the same script works against PostgreSQL when that cutover
happens. Raw DDL is used only for the table drop, which has no Core
equivalent that is safe to express against deleted metadata.
- The script is idempotent, keyed on state rather than on a version marker:
an image with no supported orientation tag is skipped, and a table that is
already absent is skipped. It is never invoked from application startup and
never runs in the test suite.
- **The application must not be running.** The image rewrite is not atomic
with the row update, and SQLite will refuse the schema change while another
connection holds the database.
Usage::
python tools/migrate_v46_to_v47.py --dry-run
python tools/migrate_v46_to_v47.py
"""
from __future__ import annotations
import argparse
import hashlib
import sys
from collections.abc import Sequence
from pathlib import Path
from sqlalchemy import bindparam
from sqlalchemy import create_engine
from sqlalchemy import inspect as sqlalchemy_inspect
from sqlalchemy import select
from sqlalchemy import text
from sqlalchemy import update
from sqlalchemy.engine import Connection
from sqlmodel import SQLModel
from transcription.config import Settings
from transcription.config import get_settings
from transcription.db import models as _models # noqa: F401 (registers every table)
from transcription.db.engine import get_database_url
from transcription.db.models import JobSourceStatus
from transcription.services.normalization import normalize_orientation
from transcription.services.sources import source_mime_type
#: Row counts the V4.6 database is expected to carry, used as a pre-flight
#: guard so the script cannot silently run against the wrong file.
EXPECTED_ROW_COUNTS = {
"document": 8,
"document_person": 11,
"document_type": 7,
"execution_attempt": 80,
"job": 11,
"job_source": 79,
"person": 5,
"person_role": 3,
"source": 76,
}
ARTIFACT_TABLE = "processing_artifact"
#: Evidence columns removed from ``job_source`` in V4.7. Every one of them is
#: duplicated byte-for-byte by ``execution_attempt`` across all 77 rows that
#: carry evidence, so no information is lost by dropping them.
JOB_SOURCE_DROPPED_COLUMNS = (
"raw_transcription",
"ai_metadata",
"raw_api_response",
"error_detail",
"executed_at",
)
#: The V4.6 default for the deleted ``Settings.artifact_dir``. The setting no
#: longer exists, so the historical location is recorded here instead.
DEFAULT_ARTIFACT_DIR = Path("data/artifacts")
def _sync_url(settings: Settings) -> str:
"""Return the target database URL with any async driver stripped."""
url = get_database_url(settings)
return url.replace("+aiosqlite", "").replace("+asyncpg", "").replace("+psycopg", "")
def _preflight(connection: Connection, *, strict: bool) -> None:
inspector = sqlalchemy_inspect(connection)
present = set(inspector.get_table_names())
mismatched: dict[str, tuple[object, int]] = {}
for name, expected in EXPECTED_ROW_COUNTS.items():
if name not in present:
mismatched[name] = ("missing", expected)
continue
actual = connection.execute(text(f'select count(*) from "{name}"')).scalar_one()
if actual != expected:
mismatched[name] = (actual, expected)
if not mismatched:
return
detail = ", ".join(f"{name}: found {found}, expected {want}" for name, (found, want) in sorted(mismatched.items()))
message = f"Database row counts do not match the recorded V4.6 snapshot ({detail})"
if strict:
raise RuntimeError(message)
print(f"WARNING: {message}", file=sys.stderr)
def rotate_stored_images(connection: Connection, *, dry_run: bool) -> int:
"""Step 1: rewrite every mis-oriented stored image and its recorded digest."""
source = SQLModel.metadata.tables["source"]
rows = connection.execute(select(source.c.id, source.c.file_path, source.c.filename)).all()
rotated = 0
missing = 0
for source_id, file_path, filename in rows:
path = Path(str(file_path))
if not path.is_file():
print(f" WARNING: source file not found, skipped: {path}", file=sys.stderr)
missing += 1
continue
content = path.read_bytes()
normalized = normalize_orientation(content, media_type=source_mime_type(str(filename)))
if normalized is None:
continue
rotated += 1
print(
f" {path.name} orientation={normalized.original_orientation} "
f"rotation={normalized.applied_rotation_degrees} "
f"{len(content)} -> {len(normalized.content)} bytes"
)
if dry_run:
continue
path.write_bytes(normalized.content)
connection.execute(
update(source)
.where(source.c.id == source_id)
.values(
file_hash=hashlib.sha256(normalized.content).hexdigest(),
file_size_bytes=len(normalized.content),
)
)
print(f" rotated={rotated} upright={len(rows) - rotated - missing} missing={missing}")
return rotated
def drop_processing_artifacts(connection: Connection, artifact_dir: Path, *, dry_run: bool) -> int:
"""Step 2: drop the artifact table and delete the files it referenced."""
inspector = sqlalchemy_inspect(connection)
if ARTIFACT_TABLE not in set(inspector.get_table_names()):
print(f" {ARTIFACT_TABLE} already absent")
return 0
references = [
str(row[0])
for row in connection.execute(
text(f'select external_reference from "{ARTIFACT_TABLE}" where external_reference is not null')
)
]
count = connection.execute(text(f'select count(*) from "{ARTIFACT_TABLE}"')).scalar_one()
print(f" dropping {ARTIFACT_TABLE} ({count} row(s), {len(references)} external file(s))")
if dry_run:
return count
connection.execute(text(f'drop table "{ARTIFACT_TABLE}"'))
artifact_root = artifact_dir.resolve()
for reference in references:
relative = Path(reference)
if relative.is_absolute() or ".." in relative.parts:
print(f" WARNING: skipped unsafe artifact reference: {reference}", file=sys.stderr)
continue
artifact_path = (artifact_root / relative).resolve()
if artifact_root not in artifact_path.parents:
print(f" WARNING: skipped artifact outside root: {reference}", file=sys.stderr)
continue
artifact_path.unlink(missing_ok=True)
parent = artifact_path.parent
if parent != artifact_root and parent.is_dir() and not any(parent.iterdir()):
parent.rmdir()
return count
def normalize_attempt_status(connection: Connection, *, dry_run: bool) -> int:
"""Step 3: rewrite ``execution_attempt.status`` from enum names to values.
Defect [45]: ``execution_attempt.status`` was declared without
``values_callable``, so SQLAlchemy persisted enum *names* ('TRANSCRIBED')
while ``job_source.status`` persisted *values* ('transcribed'). The two
columns never compared equal on a single one of the 79 rows. The model
declaration is fixed in V4.7; the stored rows are fixed here.
"""
name_to_value = {member.name: member.value for member in JobSourceStatus}
recognised = sorted(set(name_to_value) | set(name_to_value.values()))
unknown = (
connection.execute(
text("select distinct status from execution_attempt where status not in :values").bindparams(
bindparam("values", recognised, expanding=True)
)
)
.scalars()
.all()
)
if unknown:
raise RuntimeError(f"execution_attempt.status carries unrecognised spellings: {sorted(unknown)}")
rewritten = 0
for name, value in sorted(name_to_value.items()):
if name == value:
continue
count = connection.execute(
text("select count(*) from execution_attempt where status = :name"),
{"name": name},
).scalar_one()
if not count:
continue
print(f" {name} -> {value}: {count} row(s)")
rewritten += count
if dry_run:
continue
connection.execute(
text("update execution_attempt set status = :value where status = :name"),
{"name": name, "value": value},
)
print(f" rewritten={rewritten}")
return rewritten
def strip_job_source_columns(connection: Connection, *, dry_run: bool) -> int:
"""Step 4: drop the evidence columns from ``job_source``.
Uses ``ALTER TABLE ... DROP COLUMN``, supported by SQLite 3.35+ and by
PostgreSQL. Idempotent: a column that is already gone is skipped.
"""
inspector = sqlalchemy_inspect(connection)
present = {column["name"] for column in inspector.get_columns("job_source")}
targets = [name for name in JOB_SOURCE_DROPPED_COLUMNS if name in present]
if not targets:
print(" all evidence columns already dropped")
return 0
print(f" dropping {len(targets)} column(s): {', '.join(targets)}")
if dry_run:
return len(targets)
for name in targets:
connection.execute(text(f'alter table "job_source" drop column "{name}"'))
return len(targets)
def migrate(*, settings: Settings, artifact_dir: Path, dry_run: bool, strict_counts: bool) -> None:
"""Apply every V4.7 migration step in order."""
engine = create_engine(_sync_url(settings))
try:
with engine.begin() as connection:
_preflight(connection, strict=strict_counts)
print("\nStep 1: rotate stored images")
rotate_stored_images(connection, dry_run=dry_run)
print(f"\nStep 2: drop {ARTIFACT_TABLE}")
drop_processing_artifacts(connection, artifact_dir, dry_run=dry_run)
print("\nStep 3: normalize execution_attempt.status spelling")
normalize_attempt_status(connection, dry_run=dry_run)
print("\nStep 4: strip evidence columns from job_source")
strip_job_source_columns(connection, dry_run=dry_run)
finally:
engine.dispose()
if dry_run:
print("\nDry run: nothing was written.")
else:
print("\nDone.")
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("--dry-run", action="store_true", help="Report what would change without writing")
parser.add_argument(
"--allow-count-mismatch",
action="store_true",
help="Warn instead of aborting when row counts differ from the recorded V4.6 snapshot",
)
parser.add_argument(
"--artifact-dir",
type=Path,
default=DEFAULT_ARTIFACT_DIR,
help="Directory that held external artifact files before V4.7",
)
args = parser.parse_args(argv)
settings = get_settings()
print(f"Target: {_sync_url(settings)}")
migrate(
settings=settings,
artifact_dir=args.artifact_dir,
dry_run=args.dry_run,
strict_counts=not args.allow_count_mismatch,
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
tools/migrate_v47_to_v48.py
-185
View File
@@ -1,185 +0,0 @@
"""One-time migration for V4.8 queue-membership uniqueness.
This migration enforces the V4 requirement that each ``(job_id, source_id)``
pair appears at most once in ``job_source``.
Steps:
1. Detect duplicate ``job_source`` rows per ``(job_id, source_id)``.
2. Keep one row per pair (prefer the row with the latest attempt evidence).
3. Re-point ``execution_attempt.job_source_id`` from removed duplicates to the
kept row.
4. Delete duplicate ``job_source`` rows.
5. Add uniqueness enforcement for ``(job_id, source_id)``.
Usage::
python tools/migrate_v47_to_v48.py --dry-run
python tools/migrate_v47_to_v48.py
"""
from __future__ import annotations
import argparse
from collections import defaultdict
from collections.abc import Sequence
from datetime import UTC
from datetime import datetime
from uuid import UUID
from sqlalchemy import bindparam
from sqlalchemy import create_engine
from sqlalchemy import func
from sqlalchemy import inspect as sqlalchemy_inspect
from sqlalchemy import select
from sqlalchemy import text
from sqlalchemy import update
from sqlalchemy.engine import Connection
from sqlmodel import SQLModel
from transcription.config import Settings
from transcription.config import get_settings
from transcription.db import models as _models # noqa: F401 (registers metadata tables)
from transcription.db.engine import get_database_url
UNIQUE_NAME = "uq_job_source_job_source"
def _sync_url(settings: Settings) -> str:
"""Return a sync URL for direct SQLAlchemy Core access."""
return get_database_url(settings).replace("+aiosqlite", "").replace("+asyncpg", "").replace("+psycopg", "")
def _uniqueness_already_enforced(connection: Connection) -> bool:
inspector = sqlalchemy_inspect(connection)
unique_constraints = inspector.get_unique_constraints("job_source")
if any(constraint.get("name") == UNIQUE_NAME for constraint in unique_constraints):
return True
indexes = inspector.get_indexes("job_source")
return any(index.get("name") == UNIQUE_NAME and index.get("unique") is True for index in indexes)
def _choose_keeper(
candidates: list[dict[str, object]],
) -> dict[str, object]:
"""Keep the row with latest attempt activity, then lexicographically greatest UUID."""
def key(item: dict[str, object]) -> tuple[datetime, str]:
latest_attempt = item["latest_attempt_at"]
attempt_key = latest_attempt if isinstance(latest_attempt, datetime) else datetime.min.replace(tzinfo=UTC)
return (attempt_key, str(item["job_source_id"]))
return max(candidates, key=key)
def deduplicate_job_source_membership(connection: Connection, *, dry_run: bool) -> tuple[int, int]:
"""Return ``(pairs_deduplicated, rows_deleted)``."""
job_source = SQLModel.metadata.tables["job_source"]
execution_attempt = SQLModel.metadata.tables["execution_attempt"]
duplicate_pairs = (
connection.execute(
select(job_source.c.job_id, job_source.c.source_id)
.group_by(job_source.c.job_id, job_source.c.source_id)
.having(func.count(job_source.c.id) > 1)
)
.mappings()
.all()
)
if not duplicate_pairs:
return (0, 0)
rows_by_pair: dict[tuple[UUID, UUID], list[dict[str, object]]] = defaultdict(list)
duplicate_memberships = (
connection.execute(
select(
job_source.c.id.label("job_source_id"),
job_source.c.job_id,
job_source.c.source_id,
func.max(execution_attempt.c.created_at).label("latest_attempt_at"),
)
.select_from(job_source.outerjoin(execution_attempt, execution_attempt.c.job_source_id == job_source.c.id))
.group_by(job_source.c.id, job_source.c.job_id, job_source.c.source_id)
)
.mappings()
.all()
)
duplicate_key_set = {(row["job_id"], row["source_id"]) for row in duplicate_pairs}
for row in duplicate_memberships:
key = (row["job_id"], row["source_id"])
if key in duplicate_key_set:
rows_by_pair[key].append(dict(row))
rows_deleted = 0
for key, rows in sorted(rows_by_pair.items(), key=lambda item: (str(item[0][0]), str(item[0][1]))):
keeper = _choose_keeper(rows)
keeper_id = keeper["job_source_id"]
duplicate_ids = [row["job_source_id"] for row in rows if row["job_source_id"] != keeper_id]
rows_deleted += len(duplicate_ids)
print(
f" dedupe pair job_id={key[0]} source_id={key[1]} "
f"keep={keeper_id} drop={','.join(str(item) for item in duplicate_ids)}"
)
if dry_run or not duplicate_ids:
continue
connection.execute(
update(execution_attempt)
.where(execution_attempt.c.job_source_id.in_(bindparam("duplicate_ids", expanding=True)))
.values(job_source_id=keeper_id),
{"duplicate_ids": duplicate_ids},
)
connection.execute(
job_source.delete().where(job_source.c.id.in_(bindparam("duplicate_ids", expanding=True))),
{"duplicate_ids": duplicate_ids},
)
return (len(duplicate_pairs), rows_deleted)
def add_job_source_uniqueness(connection: Connection, *, dry_run: bool) -> None:
"""Create uniqueness enforcement for ``(job_id, source_id)``."""
if _uniqueness_already_enforced(connection):
print(f" uniqueness already enforced ({UNIQUE_NAME})")
return
dialect = connection.dialect.name
if dialect == "postgresql":
statement = text(f'alter table "job_source" add constraint "{UNIQUE_NAME}" unique ("job_id", "source_id")')
else:
statement = text(f'create unique index "{UNIQUE_NAME}" on "job_source" ("job_id", "source_id")')
print(f" applying {UNIQUE_NAME}")
if not dry_run:
connection.execute(statement)
def migrate(*, settings: Settings, dry_run: bool) -> None:
engine = create_engine(_sync_url(settings))
try:
with engine.begin() as connection:
print("Step 1: deduplicate job_source membership")
pair_count, rows_deleted = deduplicate_job_source_membership(connection, dry_run=dry_run)
print(f" duplicate pairs={pair_count} rows_deleted={rows_deleted}")
print("Step 2: enforce unique (job_id, source_id)")
add_job_source_uniqueness(connection, dry_run=dry_run)
finally:
engine.dispose()
if dry_run:
print("\nDry run: nothing was written.")
else:
print("\nDone.")
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--dry-run", action="store_true", help="Report planned changes without writing")
args = parser.parse_args(argv)
settings = get_settings()
print(f"Target: {_sync_url(settings)}")
migrate(settings=settings, dry_run=args.dry_run)
return 0
if __name__ == "__main__":
raise SystemExit(main())