fc5: ml_queue (fire tag_and_embed for unprocessed) + verify (row counts + sha256 sample)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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)

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,
+    }

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