feat(deep-scan): keyset-paginated backfill_phash task
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -1,16 +1,22 @@
|
||||
"""Periodic maintenance: recover stuck import tasks, garbage-collect old finished tasks."""
|
||||
|
||||
import logging
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
from PIL import Image
|
||||
from sqlalchemy import create_engine, delete, select, update
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from ..celery_app import celery
|
||||
from ..config import get_config
|
||||
from ..models import ImportTask
|
||||
from ..models import ImageRecord, ImportTask
|
||||
from ..utils.phash import compute_phash
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
STUCK_THRESHOLD_MINUTES = 30
|
||||
OLD_TASK_DAYS = 7
|
||||
PHASH_PAGE = 500
|
||||
|
||||
|
||||
def _sync_session_factory():
|
||||
@@ -71,3 +77,41 @@ def cleanup_old_tasks() -> int:
|
||||
)
|
||||
session.commit()
|
||||
return result.rowcount or 0
|
||||
|
||||
|
||||
@celery.task(name="backend.app.tasks.maintenance.backfill_phash")
|
||||
def backfill_phash() -> int:
|
||||
"""Recompute phash for stored images that have none (imported before
|
||||
FC-2d-i+ii). Keyset-paginated by id (restart-safe), NULL-only fill,
|
||||
idempotent. Videos legitimately keep phash NULL. A missing/unreadable
|
||||
file is logged and left NULL — never fails the task."""
|
||||
SessionLocal = _sync_session_factory()
|
||||
updated = 0
|
||||
last_id = 0
|
||||
with SessionLocal() as session:
|
||||
while True:
|
||||
rows = session.execute(
|
||||
select(ImageRecord)
|
||||
.where(ImageRecord.id > last_id)
|
||||
.where(ImageRecord.phash.is_(None))
|
||||
.where(ImageRecord.mime.like("image/%"))
|
||||
.order_by(ImageRecord.id.asc())
|
||||
.limit(PHASH_PAGE)
|
||||
).scalars().all()
|
||||
if not rows:
|
||||
break
|
||||
for rec in rows:
|
||||
try:
|
||||
with Image.open(rec.path) as im:
|
||||
ph = compute_phash(im)
|
||||
except Exception as exc:
|
||||
log.warning(
|
||||
"backfill_phash: unreadable %s: %s", rec.path, exc
|
||||
)
|
||||
ph = None
|
||||
if ph is not None and rec.phash is None:
|
||||
rec.phash = ph
|
||||
updated += 1
|
||||
session.commit()
|
||||
last_id = rows[-1].id
|
||||
return updated
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
"""FC-2d-vi: library-wide phash backfill (keyset, idempotent)."""
|
||||
|
||||
import pytest
|
||||
from PIL import Image
|
||||
|
||||
from backend.app.models import ImageRecord
|
||||
from backend.app.tasks.maintenance import backfill_phash
|
||||
|
||||
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):
|
||||
"""sessionmaker-like returning the test's bound session, so the task
|
||||
runs against the same transaction / sees the seeded rows."""
|
||||
class _SM:
|
||||
def __call__(self):
|
||||
return _Ctx(db_sync)
|
||||
|
||||
return _SM()
|
||||
|
||||
|
||||
def _img(path, color):
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
Image.new("RGB", (64, 64), color).save(path, "JPEG")
|
||||
|
||||
|
||||
def _rec(db_sync, path, *, sha, mime="image/jpeg", phash=None):
|
||||
rec = ImageRecord(
|
||||
path=str(path), sha256=sha, phash=phash, size_bytes=1,
|
||||
mime=mime, width=64, height=64, origin="imported_filesystem",
|
||||
integrity_status="unknown",
|
||||
)
|
||||
db_sync.add(rec)
|
||||
db_sync.flush()
|
||||
return rec
|
||||
|
||||
|
||||
def test_backfill_fills_null_image_phash(db_sync, tmp_path, monkeypatch):
|
||||
from backend.app.tasks import maintenance as m
|
||||
|
||||
p = tmp_path / "a.jpg"
|
||||
_img(p, (10, 120, 200))
|
||||
pre = tmp_path / "preset.jpg"
|
||||
_img(pre, (5, 5, 5))
|
||||
|
||||
rec = _rec(db_sync, p, sha="a" + "0" * 63, phash=None)
|
||||
vid = _rec(
|
||||
db_sync, tmp_path / "v.mp4", sha="v" + "0" * 63,
|
||||
mime="video/mp4", phash=None,
|
||||
)
|
||||
missing = _rec(
|
||||
db_sync, tmp_path / "gone.jpg", sha="g" + "0" * 63, phash=None,
|
||||
)
|
||||
preset = _rec(
|
||||
db_sync, pre, sha="p" + "0" * 63, phash="deadbeefdeadbeef",
|
||||
)
|
||||
db_sync.commit()
|
||||
|
||||
monkeypatch.setattr(m, "_sync_session_factory", lambda: _sf(db_sync))
|
||||
|
||||
updated = backfill_phash()
|
||||
db_sync.expire_all()
|
||||
assert updated == 1
|
||||
assert db_sync.get(ImageRecord, rec.id).phash is not None
|
||||
assert db_sync.get(ImageRecord, vid.id).phash is None
|
||||
assert db_sync.get(ImageRecord, missing.id).phash is None
|
||||
assert db_sync.get(ImageRecord, preset.id).phash == "deadbeefdeadbeef"
|
||||
|
||||
# idempotent: a second run touches nothing
|
||||
assert backfill_phash() == 0
|
||||
Reference in New Issue
Block a user