feat(integrity): verify_integrity task (keyset, fail-soft) + weekly beat

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

backend/app/celery_app.py

@@ -66,6 +66,10 @@ def make_celery() -> Celery:
                 "task": "backend.app.tasks.ml.apply_allowlist_tags",
                 "schedule": 86400.0,
             },
+            "integrity-verify-weekly": {
+                "task": "backend.app.tasks.maintenance.verify_integrity",
+                "schedule": 604800.0,  # weekly
+            },
         },
         timezone="UTC",
     )

backend/app/tasks/maintenance.py

@@ -1,7 +1,9 @@
 """Periodic maintenance: recover stuck import tasks, garbage-collect old finished tasks."""
 
 import logging
+import subprocess
 from datetime import UTC, datetime, timedelta
+from pathlib import Path
 
 from PIL import Image
 from sqlalchemy import create_engine, delete, select, update
@@ -17,6 +19,8 @@ log = logging.getLogger(__name__)
 STUCK_THRESHOLD_MINUTES = 30
 OLD_TASK_DAYS = 7
 PHASH_PAGE = 500
+VERIFY_PAGE = 200
+FFPROBE_TIMEOUT_SECONDS = 10
 
 
 def _sync_session_factory():
@@ -115,3 +119,80 @@ def backfill_phash() -> int:
             session.commit()
             last_id = rows[-1].id
     return updated
+
+
+def _verify_one(path: Path, expected_sha: str, mime: str, sha_fn) -> str:
+    """Compute the integrity verdict for one file. Status precedence:
+    failed_verification (can't run) > corrupt (sha mismatch / decode
+    fails) > ok (passes both). Never raises."""
+    try:
+        if not path.is_file():
+            return "failed_verification"
+        try:
+            actual_sha = sha_fn(path)
+        except (OSError, PermissionError):
+            return "failed_verification"
+        if actual_sha != expected_sha:
+            return "corrupt"
+        if mime and mime.startswith("image/"):
+            try:
+                with Image.open(path) as im:
+                    im.verify()
+            except Exception:
+                return "corrupt"
+            return "ok"
+        if mime and mime.startswith("video/"):
+            try:
+                proc = subprocess.run(
+                    ["ffprobe", "-v", "error", "-i", str(path)],
+                    capture_output=True,
+                    timeout=FFPROBE_TIMEOUT_SECONDS,
+                )
+            except FileNotFoundError:
+                # ffprobe binary missing — environment problem, not file.
+                return "failed_verification"
+            except subprocess.TimeoutExpired:
+                return "corrupt"
+            return "ok" if proc.returncode == 0 else "corrupt"
+        # Unknown mime — sha matched already; trust that.
+        return "ok"
+    except Exception as exc:
+        log.warning("verify_integrity unexpected error for %s: %s", path, exc)
+        return "failed_verification"
+
+
+@celery.task(name="backend.app.tasks.maintenance.verify_integrity")
+def verify_integrity() -> int:
+    """Verify every ImageRecord file: sha256 recompute + decode/probe
+    (PIL for images; ffprobe for videos). Writes integrity_status
+    (always — the column reflects the most recent verdict).
+    Keyset-paginated, fail-soft per row, idempotent. Returns the total
+    count verified."""
+    from ..services.importer import _sha256_of  # reuse the importer's helper
+
+    SessionLocal = _sync_session_factory()
+    total = 0
+    counts = {"ok": 0, "corrupt": 0, "failed_verification": 0}
+    last_id = 0
+    with SessionLocal() as session:
+        while True:
+            rows = session.execute(
+                select(ImageRecord)
+                .where(ImageRecord.id > last_id)
+                .order_by(ImageRecord.id.asc())
+                .limit(VERIFY_PAGE)
+            ).scalars().all()
+            if not rows:
+                break
+            for rec in rows:
+                rec.integrity_status = _verify_one(
+                    Path(rec.path), rec.sha256, rec.mime, _sha256_of
+                )
+                counts[rec.integrity_status] = (
+                    counts.get(rec.integrity_status, 0) + 1
+                )
+                total += 1
+            session.commit()
+            last_id = rows[-1].id
+    log.info("verify_integrity verdicts: %s (total %d)", counts, total)
+    return total

tests/test_verify_integrity.py

@@ -0,0 +1,139 @@
+"""FC-2e: verify_integrity — sha + decode/probe per ImageRecord."""
+
+import subprocess
+
+import pytest
+from PIL import Image
+
+from backend.app.models import ImageRecord
+from backend.app.tasks.maintenance import verify_integrity
+
+pytestmark = pytest.mark.integration
+
+
+class _Ctx:
+    def __init__(self, s):
+        self.s = s
+
+    def __enter__(self):
+        return self.s
+
+    def __exit__(self, *a):
+        return False
+
+
+def _sf(db_sync):
+    class _SM:
+        def __call__(self):
+            return _Ctx(db_sync)
+
+    return _SM()
+
+
+def _good_jpeg(path):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    Image.new("RGB", (32, 32), (40, 80, 160)).save(path, "JPEG")
+
+
+def _rec(db_sync, path, *, sha, mime="image/jpeg"):
+    rec = ImageRecord(
+        path=str(path), sha256=sha, phash=None, size_bytes=1,
+        mime=mime, width=32, height=32, origin="imported_filesystem",
+        integrity_status="unknown",
+    )
+    db_sync.add(rec)
+    db_sync.flush()
+    return rec
+
+
+def _sha_of(path):
+    import hashlib
+
+    h = hashlib.sha256()
+    h.update(path.read_bytes())
+    return h.hexdigest()
+
+
+def test_ok_corrupt_failed_paths(db_sync, tmp_path, monkeypatch):
+    from backend.app.tasks import maintenance as m
+
+    good = tmp_path / "good.jpg"
+    _good_jpeg(good)
+    bad = tmp_path / "bad.jpg"
+    bad.write_bytes(b"not a real jpeg")
+    truncated = tmp_path / "trunc.jpg"
+    _good_jpeg(truncated)
+    truncated.write_bytes(truncated.read_bytes()[:30])  # break the bitstream
+
+    rec_ok = _rec(db_sync, good, sha=_sha_of(good))
+    rec_sha_mismatch = _rec(db_sync, good, sha="0" * 64)  # bytes intact, sha wrong
+    rec_decode_fail = _rec(db_sync, truncated, sha=_sha_of(truncated))
+    # 'bad' has not-jpeg bytes; sha matches the bad bytes → decode fails
+    rec_bad_decode = _rec(db_sync, bad, sha=_sha_of(bad))
+    rec_missing = _rec(
+        db_sync, tmp_path / "gone.jpg", sha="g" + "0" * 63,
+    )
+    db_sync.commit()
+
+    monkeypatch.setattr(m, "_sync_session_factory", lambda: _sf(db_sync))
+
+    total = verify_integrity()
+    db_sync.expire_all()
+    assert total == 5
+    assert db_sync.get(ImageRecord, rec_ok.id).integrity_status == "ok"
+    assert db_sync.get(ImageRecord, rec_sha_mismatch.id).integrity_status == "corrupt"
+    assert db_sync.get(ImageRecord, rec_decode_fail.id).integrity_status == "corrupt"
+    assert db_sync.get(ImageRecord, rec_bad_decode.id).integrity_status == "corrupt"
+    assert db_sync.get(ImageRecord, rec_missing.id).integrity_status == "failed_verification"
+
+    # Idempotent — second run re-derives the same verdict.
+    verify_integrity()
+    db_sync.expire_all()
+    assert db_sync.get(ImageRecord, rec_ok.id).integrity_status == "ok"
+
+
+def test_video_ffprobe_paths(db_sync, tmp_path, monkeypatch):
+    from backend.app.tasks import maintenance as m
+
+    vid = tmp_path / "v.mp4"
+    vid.write_bytes(b"fake but consistent video bytes")
+    rec = _rec(db_sync, vid, sha=_sha_of(vid), mime="video/mp4")
+    db_sync.commit()
+
+    monkeypatch.setattr(m, "_sync_session_factory", lambda: _sf(db_sync))
+
+    # Case A: ffprobe succeeds → ok
+    def _run_ok(args, **kw):
+        return subprocess.CompletedProcess(args, 0, b"", b"")
+
+    monkeypatch.setattr(m.subprocess, "run", _run_ok)
+    verify_integrity()
+    db_sync.expire_all()
+    assert db_sync.get(ImageRecord, rec.id).integrity_status == "ok"
+
+    # Case B: ffprobe non-zero → corrupt
+    def _run_bad(args, **kw):
+        return subprocess.CompletedProcess(args, 1, b"", b"err")
+
+    monkeypatch.setattr(m.subprocess, "run", _run_bad)
+    verify_integrity()
+    db_sync.expire_all()
+    assert db_sync.get(ImageRecord, rec.id).integrity_status == "corrupt"
+
+    # Case C: ffprobe times out → corrupt
+    def _run_timeout(args, **kw):
+        raise subprocess.TimeoutExpired(cmd=args, timeout=10)
+
+    monkeypatch.setattr(m.subprocess, "run", _run_timeout)
+    verify_integrity()
+    db_sync.expire_all()
+    assert db_sync.get(ImageRecord, rec.id).integrity_status == "corrupt"
+
+    # Case D: ffprobe binary missing → failed_verification
+    def _run_missing(args, **kw):
+        raise FileNotFoundError("ffprobe")
+
+    monkeypatch.setattr(m.subprocess, "run", _run_missing)
+    verify_integrity()
+    db_sync.expire_all()
+    assert db_sync.get(ImageRecord, rec.id).integrity_status == "failed_verification"