From 22cdf0f3347c64a917922096f03f12b32bd29dc7 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 10 Jun 2026 16:03:58 -0400 Subject: [PATCH] feat(ml): read suggestions + allowlist from image_prediction (#768 step 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Switch every prediction READER off the JSON column onto the normalized image_prediction table. Parity by construction: each reader loads the same {raw_name: {category, confidence}} dict it consumed before (via small _load_predictions helpers), so all downstream threshold/alias/merge/consensus logic is byte-identical — only the data source changed. - suggestions.SuggestionService.for_image (and for_selection via it) - ml.apply_allowlist_tags (iterates images that have prediction rows) - importer re-import reset deletes the image's prediction rows The tagger_predictions JSON column is still dual-written (step 1) so it stays valid during transition; the backfill task's NULL check still works. Removing the JSON write + DROP column + retiring the #764 prune is the cleanup follow-up (needs a quiesced-worker window for the DROP lock). Tests: shared tests/_prediction_helpers.seed_predictions seeds the table; read-path tests (suggestions, bulk consensus, allowlist apply, API) seed there instead of ImageRecord.tagger_predictions. Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/app/services/importer.py | 10 +++++ backend/app/services/ml/suggestions.py | 22 ++++++++++- backend/app/tasks/ml.py | 33 ++++++++++++++--- tests/_prediction_helpers.py | 21 +++++++++++ tests/test_api_suggestions.py | 5 ++- tests/test_ml_suggestions.py | 51 ++++++++++++++------------ tests/test_suggestions_bulk.py | 30 +++++++++------ tests/test_tasks_ml.py | 18 ++++++--- 8 files changed, 142 insertions(+), 48 deletions(-) create mode 100644 tests/_prediction_helpers.py diff --git a/backend/app/services/importer.py b/backend/app/services/importer.py index a79da39..a1cd9b5 100644 --- a/backend/app/services/importer.py +++ b/backend/app/services/importer.py @@ -1090,6 +1090,16 @@ class Importer: existing.siglip_embedding = None existing.siglip_model_version = None existing.centroid_scores = None + # #768: predictions also live in the normalized image_prediction table + # now — clear them so a re-imported file re-derives a fresh set. + from sqlalchemy import delete as _delete + + from ..models import ImagePrediction as _ImagePrediction + self.session.execute( + _delete(_ImagePrediction).where( + _ImagePrediction.image_record_id == existing.id + ) + ) # created_at intentionally preserved; updated_at auto-bumps. self.session.flush() self.session.commit() diff --git a/backend/app/services/ml/suggestions.py b/backend/app/services/ml/suggestions.py index b53e198..ca60c30 100644 --- a/backend/app/services/ml/suggestions.py +++ b/backend/app/services/ml/suggestions.py @@ -8,6 +8,7 @@ from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from ...models import ( + ImagePrediction, ImageRecord, MLSettings, Tag, @@ -48,6 +49,25 @@ class SuggestionService: await self.session.execute(select(MLSettings).where(MLSettings.id == 1)) ).scalar_one() + async def _load_predictions(self, image_id: int) -> dict: + """Predictions for one image from the normalized image_prediction + table (#768), in the {raw_name: {category, confidence}} shape the rest + of this service consumed from the old JSON column — so all downstream + threshold/alias/merge logic is unchanged.""" + rows = ( + await self.session.execute( + select( + ImagePrediction.raw_name, + ImagePrediction.category, + ImagePrediction.score, + ).where(ImagePrediction.image_record_id == image_id) + ) + ).all() + return { + r.raw_name: {"category": r.category, "confidence": r.score} + for r in rows + } + def _threshold_for( self, s: MLSettings, category: str, override: float | None = None, ) -> float: @@ -80,7 +100,7 @@ class SuggestionService: return SuggestionList() settings = await self._settings() - predictions: dict = img.tagger_predictions or {} + predictions: dict = await self._load_predictions(image_id) applied = set( ( diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index c79ed8d..3d62213 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -348,14 +348,16 @@ def apply_allowlist_tags(self, tag_id: int | None = None, if not allow: return 0 - img_query = sa_select( - ImageRecord.id, ImageRecord.tagger_predictions - ).where(ImageRecord.tagger_predictions.is_not(None)) + # Images that have any predictions (#768: from image_prediction, not + # the old JSON column), optionally narrowed to one image. + img_ids_query = sa_select(ImagePrediction.image_record_id).distinct() if image_id is not None: - img_query = img_query.where(ImageRecord.id == image_id) + img_ids_query = img_ids_query.where( + ImagePrediction.image_record_id == image_id + ) - for img_id, preds in session.execute(img_query).all(): - preds = preds or {} + for (img_id,) in session.execute(img_ids_query).all(): + preds = _load_predictions_sync(session, img_id) for a_tag_id, min_conf in allow.items(): exists = session.execute( sa_select(image_tag.c.tag_id).where( @@ -394,6 +396,25 @@ def apply_allowlist_tags(self, tag_id: int | None = None, return applied +def _load_predictions_sync(session, image_id: int) -> dict: + """Predictions for one image from image_prediction (#768), in the + {raw_name: {category, confidence}} shape _confidence_for_tag consumes — + keeps the allowlist resolution logic unchanged.""" + from sqlalchemy import select as sa_select + + rows = session.execute( + sa_select( + ImagePrediction.raw_name, + ImagePrediction.category, + ImagePrediction.score, + ).where(ImagePrediction.image_record_id == image_id) + ).all() + return { + r.raw_name: {"category": r.category, "confidence": r.score} + for r in rows + } + + 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 diff --git a/tests/_prediction_helpers.py b/tests/_prediction_helpers.py new file mode 100644 index 0000000..e689ccc --- /dev/null +++ b/tests/_prediction_helpers.py @@ -0,0 +1,21 @@ +"""#768 test helper: seed image_prediction rows. + +Read-path tests used to seed ImageRecord(tagger_predictions={...}); predictions +now live in the normalized image_prediction table, so seed there instead. +""" +from backend.app.models import ImagePrediction + + +async def seed_predictions(session, image_id: int, predictions: dict) -> None: + """Insert image_prediction rows from a {raw_name: {category, confidence}} + dict (the old JSON shape). Caller commits/flushes as needed; this flushes.""" + session.add_all([ + ImagePrediction( + image_record_id=image_id, + raw_name=name, + category=p.get("category", "general"), + score=float(p.get("confidence", 0.0)), + ) + for name, p in predictions.items() + ]) + await session.flush() diff --git a/tests/test_api_suggestions.py b/tests/test_api_suggestions.py index f8bd49c..b9fd041 100644 --- a/tests/test_api_suggestions.py +++ b/tests/test_api_suggestions.py @@ -15,14 +15,17 @@ def eager(): async def _img(db, preds): + from tests._prediction_helpers import seed_predictions + img = ImageRecord( path="/images/s.jpg", sha256="s" * 64, size_bytes=1, mime="image/jpeg", width=1, height=1, origin="imported_filesystem", integrity_status="unknown", - tagger_predictions=preds, ) db.add(img) await db.commit() + await seed_predictions(db, img.id, preds) + await db.commit() return img diff --git a/tests/test_ml_suggestions.py b/tests/test_ml_suggestions.py index 9486ce8..891b54f 100644 --- a/tests/test_ml_suggestions.py +++ b/tests/test_ml_suggestions.py @@ -9,7 +9,7 @@ from backend.app.services.tag_service import TagService pytestmark = pytest.mark.integration -def _img(sha: str, predictions: dict) -> ImageRecord: +def _img(sha: str) -> ImageRecord: return ImageRecord( path=f"/images/{sha}.jpg", sha256=sha, @@ -19,24 +19,34 @@ def _img(sha: str, predictions: dict) -> ImageRecord: height=1, origin="imported_filesystem", integrity_status="unknown", - tagger_predictions=predictions, ) +async def _seed_img(db, sha: str, predictions: dict) -> ImageRecord: + """#768: create an image + seed its predictions into image_prediction + (the read path's source), returning the flushed record.""" + from tests._prediction_helpers import seed_predictions + + img = _img(sha) + db.add(img) + await db.flush() + await seed_predictions(db, img.id, predictions) + return img + + @pytest.mark.asyncio async def test_threshold_filters_low_confidence_general(db): # Default general threshold is 0.50 (alembic 0029 lowered it from # 0.95). Use 0.30/0.60 to keep the test asserting threshold behavior # rather than the exact cutoff number. - img = _img( + img = await _seed_img( + db, "a" * 64, { "lowconf": {"category": "general", "confidence": 0.30}, "sword": {"category": "general", "confidence": 0.97}, }, ) - db.add(img) - await db.flush() sl = await SuggestionService(db).for_image(img.id) names = [s.display_name for s in sl.by_category.get("general", [])] # display_name is normalized (tag_name.normalize) before surfacing. @@ -49,36 +59,34 @@ async def test_threshold_override_surfaces_low_confidence(db): # The typed-dropdown "show everything the model saw" mode: threshold_override # surfaces stored predictions below the configured threshold (in canonical # formatting) so they can be picked instead of hand-typed (2026-06-09). - img = _img( + img = await _seed_img( + db, "d" * 64, { "lowconf": {"category": "general", "confidence": 0.30}, "sword": {"category": "general", "confidence": 0.97}, }, ) - db.add(img) - await db.flush() sl = await SuggestionService(db).for_image(img.id, threshold_override=0.0) names = [s.display_name for s in sl.by_category.get("general", [])] assert "Sword" in names assert "Lowconf" in names # below the configured threshold, surfaced anyway # Unsurfaced categories are still excluded even with the override. - img2 = _img("e" * 64, {"safe": {"category": "rating", "confidence": 0.99}}) - db.add(img2) - await db.flush() + img2 = await _seed_img( + db, "e" * 64, {"safe": {"category": "rating", "confidence": 0.99}} + ) sl2 = await SuggestionService(db).for_image(img2.id, threshold_override=0.0) assert "rating" not in sl2.by_category @pytest.mark.asyncio async def test_unsurfaced_category_dropped(db): - img = _img( + img = await _seed_img( + db, "b" * 64, {"safe": {"category": "rating", "confidence": 0.99}}, ) - db.add(img) - await db.flush() sl = await SuggestionService(db).for_image(img.id) assert "rating" not in sl.by_category @@ -88,12 +96,11 @@ async def test_alias_resolution(db): tags = TagService(db) canonical = await tags.find_or_create("Sasuke Uchiha", TagKind.character) await AliasService(db).create("uchiha_sasuke", "character", canonical.id) - img = _img( + img = await _seed_img( + db, "c" * 64, {"uchiha_sasuke": {"category": "character", "confidence": 0.96}}, ) - db.add(img) - await db.flush() sl = await SuggestionService(db).for_image(img.id) chars = sl.by_category["character"] assert len(chars) == 1 @@ -104,12 +111,11 @@ async def test_alias_resolution(db): @pytest.mark.asyncio async def test_raw_tag_creates_new(db): - img = _img( + img = await _seed_img( + db, "d" * 64, {"brand_new_tag": {"category": "character", "confidence": 0.96}}, ) - db.add(img) - await db.flush() sl = await SuggestionService(db).for_image(img.id) chars = sl.by_category["character"] # display_name is the normalized Camie name (underscores -> spaces, @@ -123,12 +129,11 @@ async def test_raw_tag_creates_new(db): async def test_applied_tag_not_suggested(db): tags = TagService(db) tag = await tags.find_or_create("alreadyhere", TagKind.character) - img = _img( + img = await _seed_img( + db, "e" * 64, {"alreadyhere": {"category": "character", "confidence": 0.96}}, ) - db.add(img) - await db.flush() await db.execute( image_tag.insert().values( image_record_id=img.id, tag_id=tag.id, source="manual" diff --git a/tests/test_suggestions_bulk.py b/tests/test_suggestions_bulk.py index 1bfe40d..d0d101a 100644 --- a/tests/test_suggestions_bulk.py +++ b/tests/test_suggestions_bulk.py @@ -5,16 +5,16 @@ from backend.app.models import ImageRecord, TagKind from backend.app.models.tag import image_tag from backend.app.services.ml.suggestions import SuggestionService from backend.app.services.tag_service import TagService +from tests._prediction_helpers import seed_predictions pytestmark = pytest.mark.integration -def _img(sha: str, predictions: dict) -> ImageRecord: +def _img(sha: str) -> ImageRecord: return ImageRecord( path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg", width=1, height=1, origin="imported_filesystem", integrity_status="unknown", - tagger_predictions=predictions, ) @@ -22,10 +22,12 @@ def _img(sha: str, predictions: dict) -> ImageRecord: async def test_consensus_includes_tag_over_threshold(db): tags = TagService(db) t = await tags.find_or_create("sword", TagKind.general) - a = _img("a" * 64, {"sword": {"category": "general", "confidence": 0.97}}) - b = _img("b" * 64, {"sword": {"category": "general", "confidence": 0.95}}) + a = _img("a" * 64) + b = _img("b" * 64) db.add_all([a, b]) await db.flush() + await seed_predictions(db, a.id, {"sword": {"category": "general", "confidence": 0.97}}) + await seed_predictions(db, b.id, {"sword": {"category": "general", "confidence": 0.95}}) res = await SuggestionService(db).for_selection([a.id, b.id], threshold=0.8) gen = res["general"] assert any(s["canonical_tag_id"] == t.id for s in gen) @@ -38,10 +40,11 @@ async def test_consensus_includes_tag_over_threshold(db): async def test_consensus_counts_already_applied_for_coverage(db): tags = TagService(db) t = await tags.find_or_create("sky", TagKind.general) - a = _img("c" * 64, {"sky": {"category": "general", "confidence": 0.96}}) - b = _img("d" * 64, {}) # no prediction + a = _img("c" * 64) + b = _img("d" * 64) # no prediction db.add_all([a, b]) await db.flush() + await seed_predictions(db, a.id, {"sky": {"category": "general", "confidence": 0.96}}) # b already has the tag applied -> counts toward coverage, not confidence await db.execute( image_tag.insert().values( @@ -58,10 +61,11 @@ async def test_consensus_counts_already_applied_for_coverage(db): async def test_consensus_excludes_below_threshold(db): tags = TagService(db) await tags.find_or_create("rare", TagKind.general) - a = _img("e" * 64, {"rare": {"category": "general", "confidence": 0.96}}) - b = _img("f" * 64, {}) + a = _img("e" * 64) + b = _img("f" * 64) db.add_all([a, b]) await db.flush() + await seed_predictions(db, a.id, {"rare": {"category": "general", "confidence": 0.96}}) res = await SuggestionService(db).for_selection([a.id, b.id], threshold=0.8) assert all( s["name"] != "rare" for s in res.get("general", []) @@ -70,10 +74,12 @@ async def test_consensus_excludes_below_threshold(db): @pytest.mark.asyncio async def test_consensus_skips_creates_new_tag(db): - a = _img("g" * 64, {"neverseen": {"category": "general", "confidence": 0.99}}) - b = _img("h" * 64, {"neverseen": {"category": "general", "confidence": 0.99}}) + a = _img("g" * 64) + b = _img("h" * 64) db.add_all([a, b]) await db.flush() + await seed_predictions(db, a.id, {"neverseen": {"category": "general", "confidence": 0.99}}) + await seed_predictions(db, b.id, {"neverseen": {"category": "general", "confidence": 0.99}}) res = await SuggestionService(db).for_selection([a.id, b.id], threshold=0.8) # 'neverseen' has no Tag row -> creates_new_tag -> excluded from consensus assert all(s["name"] != "neverseen" for s in res.get("general", [])) @@ -90,9 +96,11 @@ async def test_bulk_suggestions_route(db): tags = TagService(db) await tags.find_or_create("sword", TagKind.general) - a = _img("i" * 64, {"sword": {"category": "general", "confidence": 0.97}}) + a = _img("i" * 64) db.add(a) await db.commit() + await seed_predictions(db, a.id, {"sword": {"category": "general", "confidence": 0.97}}) + await db.commit() app = create_app() async with app.test_client() as c: resp = await c.post( diff --git a/tests/test_tasks_ml.py b/tests/test_tasks_ml.py index e759a2e..57afdab 100644 --- a/tests/test_tasks_ml.py +++ b/tests/test_tasks_ml.py @@ -64,18 +64,21 @@ async def test_apply_allowlist_applies_above_threshold(db): from backend.app.services.tag_service import TagService from backend.app.tasks import ml as ml_tasks + from tests._prediction_helpers import seed_predictions + 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() + await seed_predictions( + db, img.id, {"autohero": {"category": "character", "confidence": 0.97}} + ) + await db.commit() n = ml_tasks.apply_allowlist_tags(tag_id=tag.id) assert n >= 1 @@ -99,18 +102,21 @@ async def test_apply_allowlist_skips_below_threshold(db): from backend.app.services.tag_service import TagService from backend.app.tasks import ml as ml_tasks + from tests._prediction_helpers import seed_predictions + 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() + await seed_predictions( + db, img.id, {"lowconf": {"category": "character", "confidence": 0.40}} + ) + await db.commit() ml_tasks.apply_allowlist_tags(tag_id=tag.id) applied = ( await db.execute(