From b5896427d462ff53871aa56367a1dce95f197e51 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 22 May 2026 09:01:46 -0400 Subject: [PATCH] fc5: ml_queue (fire tag_and_embed for unprocessed) + verify (row counts + sha256 sample) Co-Authored-By: Claude Opus 4.7 (1M context) --- backend/app/services/migrators/ml_queue.py | 21 ++++++ backend/app/services/migrators/verify.py | 77 ++++++++++++++++++++++ tests/test_migration_verify.py | 72 ++++++++++++++++++++ 3 files changed, 170 insertions(+) create mode 100644 backend/app/services/migrators/ml_queue.py create mode 100644 backend/app/services/migrators/verify.py create mode 100644 tests/test_migration_verify.py diff --git a/backend/app/services/migrators/ml_queue.py b/backend/app/services/migrators/ml_queue.py new file mode 100644 index 0000000..6d9e92f --- /dev/null +++ b/backend/app/services/migrators/ml_queue.py @@ -0,0 +1,21 @@ +"""Queue every migrated image_record with no embedding for ML re-processing.""" +from __future__ import annotations + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from ...models import ImageRecord + + +async def queue_all_unprocessed_async(db: AsyncSession) -> int: + """Find every ImageRecord with siglip_embedding IS NULL, fire + tag_and_embed.delay(id) for each. Returns count queued. + """ + from ...tasks.ml import tag_and_embed + + rows = (await db.execute( + select(ImageRecord.id).where(ImageRecord.siglip_embedding.is_(None)) + )).scalars().all() + for image_id in rows: + tag_and_embed.delay(image_id) + return len(rows) diff --git a/backend/app/services/migrators/verify.py b/backend/app/services/migrators/verify.py new file mode 100644 index 0000000..df8c951 --- /dev/null +++ b/backend/app/services/migrators/verify.py @@ -0,0 +1,77 @@ +"""Post-migration verification: row counts + sha256 sampling.""" +from __future__ import annotations + +import hashlib +from pathlib import Path + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from ...models import Artist, Credential, ImageRecord, Source, Tag + + +async def verify_async(db: AsyncSession, *, expected: dict | None = None) -> dict: + """Return per-check status dicts. `expected` is optional row-count + assertions; checks default to status='ok' when no expected provided.""" + expected = expected or {} + results: dict[str, dict] = {} + + checks = { + "artist_subscriptions": ( + select(func.count(Artist.id)).where(Artist.is_subscription == True) + ), + "source_count": select(func.count(Source.id)), + "credential_count": select(func.count(Credential.id)), + "tag_count": select(func.count(Tag.id)), + "image_record_imported_or_downloaded": ( + select(func.count(ImageRecord.id)) + .where(ImageRecord.origin.in_(["imported_filesystem", "downloaded"])) + ), + } + for name, stmt in checks.items(): + actual = (await db.execute(stmt)).scalar_one() + exp = expected.get(name) + status = "ok" if exp is None or exp == actual else "mismatch" + results[name] = {"status": status, "actual": int(actual), "expected": exp} + return results + + +async def verify_sha256_sample( + db: AsyncSession, *, sample_size: int = 20, +) -> dict: + """Sample N image_records; verify file exists + sha256 matches.""" + rows = (await db.execute( + select(ImageRecord.id, ImageRecord.path, ImageRecord.sha256) + .order_by(func.random()).limit(sample_size) + )).all() + + matched = 0 + mismatched = 0 + missing = 0 + samples: list[dict] = [] + for img_id, path, expected_sha in rows: + p = Path(path) + if not p.exists(): + missing += 1 + samples.append({"id": img_id, "path": path, "result": "missing"}) + continue + h = hashlib.sha256() + with p.open("rb") as f: + for chunk in iter(lambda: f.read(65536), b""): + h.update(chunk) + if h.hexdigest() == expected_sha: + matched += 1 + samples.append({"id": img_id, "result": "ok"}) + else: + mismatched += 1 + samples.append({ + "id": img_id, "path": path, "result": "mismatch", + "expected_sha": expected_sha, "actual_sha": h.hexdigest(), + }) + return { + "sample_size": len(rows), + "matched": matched, + "mismatched": mismatched, + "missing": missing, + "samples": samples, + } diff --git a/tests/test_migration_verify.py b/tests/test_migration_verify.py new file mode 100644 index 0000000..549cabb --- /dev/null +++ b/tests/test_migration_verify.py @@ -0,0 +1,72 @@ +"""FC-5: verify + ml_queue unit tests.""" +import hashlib + +import pytest + +from backend.app.models import Artist, ImageRecord +from backend.app.services.migrators import ml_queue, verify + +pytestmark = pytest.mark.integration + + +@pytest.mark.asyncio +async def test_verify_row_counts_match(db): + db.add(Artist(name="vAlpha", slug="valpha", is_subscription=True)) + db.add(Artist(name="vBeta", slug="vbeta", is_subscription=False)) + await db.commit() + result = await verify.verify_async( + db, expected={"artist_subscriptions": 1}, + ) + assert result["artist_subscriptions"]["status"] == "ok" + + +@pytest.mark.asyncio +async def test_verify_sha256_sample_matches_disk(db, tmp_path): + f = tmp_path / "sample.bin" + f.write_bytes(b"hello world") + sha = hashlib.sha256(b"hello world").hexdigest() + db.add(ImageRecord( + path=str(f), sha256=sha, size_bytes=11, mime="application/octet-stream", + origin="imported_filesystem", + )) + await db.commit() + result = await verify.verify_sha256_sample(db, sample_size=10) + assert result["matched"] >= 1 + assert result["mismatched"] == 0 + + +@pytest.mark.asyncio +async def test_verify_sha256_sample_detects_missing_file(db, tmp_path): + sha = "f" * 64 + db.add(ImageRecord( + path=str(tmp_path / "does-not-exist.bin"), + sha256=sha, size_bytes=1, mime="application/octet-stream", + origin="imported_filesystem", + )) + await db.commit() + result = await verify.verify_sha256_sample(db, sample_size=10) + assert result["missing"] >= 1 + + +@pytest.mark.asyncio +async def test_ml_queue_returns_count_for_unprocessed(db, tmp_path, monkeypatch): + queued: list[int] = [] + + class _FakeTask: + def delay(self, image_id): + queued.append(image_id) + + monkeypatch.setattr( + "backend.app.tasks.ml.tag_and_embed", _FakeTask(), + ) + + f = tmp_path / "q.bin" + f.write_bytes(b"x") + db.add(ImageRecord( + path=str(f), sha256="q" * 64, size_bytes=1, + mime="application/octet-stream", origin="imported_filesystem", + )) + await db.commit() + + count = await ml_queue.queue_all_unprocessed_async(db) + assert count >= 1