diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 1cf8908..a2d3a56 100644 --- a/backend/app/celery_app.py +++ b/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", ) diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 1572406..be6c932 100644 --- a/backend/app/tasks/maintenance.py +++ b/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 diff --git a/tests/test_verify_integrity.py b/tests/test_verify_integrity.py new file mode 100644 index 0000000..ee4593a --- /dev/null +++ b/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"