From 3e6cc8fffabf482ef86cc02737baddd1bc6f607b Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 15 May 2026 07:43:39 -0400 Subject: [PATCH] feat(fc2b): add allowlist-apply + centroid recompute tasks + beat apply_allowlist_tags: 4 modes (tag-only / image-only / both / full sweep), matches a tag to a prediction either by direct name or via alias (name, category) resolution, gates on per-tag min_confidence, skips applied/rejected, applies source='ml_auto'. recompute_centroid / recompute_centroids: async-bridged calls into CentroidService, delta-gated. Beat: daily backfill, daily centroid recompute, daily allowlist sweep. Co-Authored-By: Claude Opus 4.7 (1M context) --- backend/app/tasks/ml.py | 149 +++++++++++++++++++++++++++++++++++++++- tests/test_tasks_ml.py | 68 ++++++++++++++++++ 2 files changed, 214 insertions(+), 3 deletions(-) diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index 5305644..29a9816 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -189,9 +189,152 @@ def backfill(self) -> int: return enqueued -# --- Defined fully in Task 9/10. Stub so tag_and_embed's .delay() resolves -# and the module imports cleanly between commits. Replaced in Task 9. --- @celery.task(name="backend.app.tasks.ml.apply_allowlist_tags", bind=True) def apply_allowlist_tags(self, tag_id: int | None = None, image_id: int | None = None) -> int: - return 0 # replaced in Task 9 + """Retroactively apply allowlisted tags. + + Modes: + - tag_id only : scan all images for this tag. + - image_id only : scan all allowlisted tags for this image. + - both : just the (image, tag) pair. + - neither : full sweep (daily beat). + + Skips: already-applied, rejected (tag_suggestion_rejection), or + confidence below the tag's allowlist min_confidence. Applied with + source='ml_auto'. + """ + from sqlalchemy import and_ + from sqlalchemy import select as sa_select + from sqlalchemy.dialects.postgresql import insert as pg_insert + + from ..models import TagAllowlist, TagSuggestionRejection + from ..models.tag import image_tag + + SessionLocal = _sync_session_factory() + applied = 0 + with SessionLocal() as session: + allow_rows = session.execute( + sa_select(TagAllowlist.tag_id, TagAllowlist.min_confidence) + if tag_id is None + else sa_select( + TagAllowlist.tag_id, TagAllowlist.min_confidence + ).where(TagAllowlist.tag_id == tag_id) + ).all() + allow = {r[0]: r[1] for r in allow_rows} + if not allow: + return 0 + + img_query = sa_select( + ImageRecord.id, ImageRecord.tagger_predictions + ).where(ImageRecord.tagger_predictions.is_not(None)) + if image_id is not None: + img_query = img_query.where(ImageRecord.id == image_id) + + for img_id, preds in session.execute(img_query).all(): + preds = preds or {} + for a_tag_id, min_conf in allow.items(): + exists = session.execute( + sa_select(image_tag.c.tag_id).where( + and_( + image_tag.c.image_record_id == img_id, + image_tag.c.tag_id == a_tag_id, + ) + ) + ).scalar_one_or_none() + if exists is not None: + continue + rej = session.get( + TagSuggestionRejection, (img_id, a_tag_id) + ) + if rej is not None: + continue + from ..models import Tag + + tag = session.get(Tag, a_tag_id) + if tag is None: + continue + conf = _confidence_for_tag(session, tag, preds) + if conf is None or conf < min_conf: + continue + stmt = pg_insert(image_tag).values( + image_record_id=img_id, + tag_id=a_tag_id, + source="ml_auto", + ) + stmt = stmt.on_conflict_do_nothing( + index_elements=["image_record_id", "tag_id"] + ) + session.execute(stmt) + applied += 1 + session.commit() + return applied + + +def _confidence_for_tag(session, tag, preds: dict) -> float | None: + """Highest confidence among predictions that resolve to `tag` — + either the prediction name equals the tag name, or an alias maps + (prediction name, category) -> tag.id. + """ + from sqlalchemy import select as sa_select + + from ..models import TagAlias + + best: float | None = None + direct = preds.get(tag.name) + if direct is not None: + best = float(direct.get("confidence", 0.0)) + alias_rows = session.execute( + sa_select(TagAlias.alias_string, TagAlias.alias_category).where( + TagAlias.canonical_tag_id == tag.id + ) + ).all() + for alias_string, alias_category in alias_rows: + p = preds.get(alias_string) + if p is None: + continue + if p.get("category") != alias_category: + continue + c = float(p.get("confidence", 0.0)) + if best is None or c > best: + best = c + return best + + +@celery.task(name="backend.app.tasks.ml.recompute_centroid", bind=True) +def recompute_centroid(self, tag_id: int) -> bool: + import asyncio + + from ..extensions import make_engine, make_session_factory + from ..services.ml.centroids import CentroidService + + async def _run() -> bool: + engine = make_engine() + Session = make_session_factory(engine) + async with Session() as session: + svc = CentroidService(session) + result = await svc.recompute_for_tag(tag_id) + await session.commit() + return result + + return asyncio.run(_run()) + + +@celery.task(name="backend.app.tasks.ml.recompute_centroids", bind=True) +def recompute_centroids(self) -> int: + """Daily: find drifted centroids, enqueue recompute_centroid for each.""" + import asyncio + + from ..extensions import make_engine, make_session_factory + from ..services.ml.centroids import CentroidService + + async def _list() -> list[int]: + engine = make_engine() + Session = make_session_factory(engine) + async with Session() as session: + return await CentroidService(session).list_drifted() + + drifted = asyncio.run(_list()) + for tid in drifted: + recompute_centroid.delay(tid) + return len(drifted) diff --git a/tests/test_tasks_ml.py b/tests/test_tasks_ml.py index fe383df..e759a2e 100644 --- a/tests/test_tasks_ml.py +++ b/tests/test_tasks_ml.py @@ -52,3 +52,71 @@ async def test_backfill_enqueues_missing(db, monkeypatch): count = ml_tasks.backfill() assert count >= 1 assert img.id in calls + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_apply_allowlist_applies_above_threshold(db): + from sqlalchemy import select + + from backend.app.models import ImageRecord, TagAllowlist, TagKind + from backend.app.models.tag import image_tag + from backend.app.services.tag_service import TagService + from backend.app.tasks import ml as ml_tasks + + tag = await TagService(db).find_or_create("autohero", TagKind.character) + db.add(TagAllowlist(tag_id=tag.id, min_confidence=0.95)) + img = ImageRecord( + path="/images/al.jpg", sha256="al" + "0" * 62, size_bytes=1, + mime="image/jpeg", width=1, height=1, + origin="imported_filesystem", integrity_status="unknown", + tagger_predictions={ + "autohero": {"category": "character", "confidence": 0.97} + }, + ) + db.add(img) + await db.commit() + + n = ml_tasks.apply_allowlist_tags(tag_id=tag.id) + assert n >= 1 + src = ( + await db.execute( + select(image_tag.c.source) + .where(image_tag.c.image_record_id == img.id) + .where(image_tag.c.tag_id == tag.id) + ) + ).scalar_one() + assert src == "ml_auto" + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_apply_allowlist_skips_below_threshold(db): + from sqlalchemy import select + + from backend.app.models import ImageRecord, TagAllowlist, TagKind + from backend.app.models.tag import image_tag + from backend.app.services.tag_service import TagService + from backend.app.tasks import ml as ml_tasks + + tag = await TagService(db).find_or_create("lowconf", TagKind.character) + db.add(TagAllowlist(tag_id=tag.id, min_confidence=0.95)) + img = ImageRecord( + path="/images/lc.jpg", sha256="lc" + "0" * 62, size_bytes=1, + mime="image/jpeg", width=1, height=1, + origin="imported_filesystem", integrity_status="unknown", + tagger_predictions={ + "lowconf": {"category": "character", "confidence": 0.40} + }, + ) + db.add(img) + await db.commit() + ml_tasks.apply_allowlist_tags(tag_id=tag.id) + applied = ( + await db.execute( + select(image_tag.c.tag_id) + .where(image_tag.c.image_record_id == img.id) + .where(image_tag.c.tag_id == tag.id) + ) + ).scalar_one_or_none() + assert applied is None