Public Access
136 lines
4.0 KiB
Python
136 lines
4.0 KiB
Python
"""Fuzzy ingredient<->grocery_item matcher.
|
|
|
|
Builds a candidate pool of (text, ingredient_id) tuples from the
|
|
canonical ingredient table (name + aliases), ranks each grocery_item
|
|
name against the pool with rapidfuzz, and writes the top N matches
|
|
above a confidence threshold to the ingredient_grocery_match table.
|
|
|
|
Manual matches (source='manual') are preserved across runs.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from decimal import Decimal
|
|
from typing import Iterable, List, Tuple
|
|
from uuid import UUID
|
|
|
|
from rapidfuzz import fuzz, process
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import (
|
|
GroceryItem,
|
|
Ingredient,
|
|
IngredientGroceryMatch,
|
|
IngredientMatchSource,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class MatchResult:
|
|
ingredient_id: UUID
|
|
grocery_item_id: UUID
|
|
confidence: float
|
|
|
|
|
|
def build_match_pool(ingredients: Iterable[dict]) -> List[Tuple[str, str]]:
|
|
"""Flatten (canonical name + aliases) into (text, ingredient_id) pairs."""
|
|
pool: List[Tuple[str, str]] = []
|
|
for ing in ingredients:
|
|
pool.append((ing["name"], ing["id"]))
|
|
for alias in ing.get("aliases") or []:
|
|
if alias:
|
|
pool.append((alias, ing["id"]))
|
|
return pool
|
|
|
|
|
|
def rank_candidates(
|
|
target: str,
|
|
pool: List[Tuple[str, str]],
|
|
top_n: int = 3,
|
|
threshold: float = 0.75,
|
|
) -> List[dict]:
|
|
"""Return up to top_n unique-by-ingredient_id matches above threshold."""
|
|
if not pool:
|
|
return []
|
|
texts = [t for t, _ in pool]
|
|
extracted = process.extract(target, texts, scorer=fuzz.WRatio, limit=20)
|
|
seen: set[str] = set()
|
|
out: List[dict] = []
|
|
for matched_text, score, idx in extracted:
|
|
confidence = score / 100.0
|
|
if confidence < threshold:
|
|
continue
|
|
ingredient_id = pool[idx][1]
|
|
if ingredient_id in seen:
|
|
continue
|
|
seen.add(ingredient_id)
|
|
out.append(
|
|
{
|
|
"ingredient_id": ingredient_id,
|
|
"matched_text": matched_text,
|
|
"confidence": confidence,
|
|
}
|
|
)
|
|
if len(out) >= top_n:
|
|
break
|
|
return out
|
|
|
|
|
|
def run_match_job(
|
|
db: Session,
|
|
*,
|
|
source_filter: str = "lucky_california",
|
|
top_n: int = 3,
|
|
threshold: float = 0.75,
|
|
) -> int:
|
|
"""Refresh ingredient_grocery_match for every grocery_item from `source_filter`.
|
|
|
|
Manual matches (source='manual') are NOT touched. Returns the number of
|
|
auto rows written/updated.
|
|
"""
|
|
ingredients = [
|
|
{"id": str(row.id), "name": row.name, "aliases": list(row.aliases or [])}
|
|
for row in db.query(Ingredient).all()
|
|
]
|
|
pool = build_match_pool(ingredients)
|
|
|
|
grocery_rows = (
|
|
db.query(GroceryItem)
|
|
.filter(GroceryItem.source == source_filter)
|
|
.all()
|
|
)
|
|
|
|
written = 0
|
|
for grocery in grocery_rows:
|
|
target = " ".join(filter(None, [grocery.name, grocery.brand or ""])).strip()
|
|
ranked = rank_candidates(target, pool, top_n=top_n, threshold=threshold)
|
|
for r in ranked:
|
|
ing_id = UUID(r["ingredient_id"])
|
|
existing = (
|
|
db.query(IngredientGroceryMatch)
|
|
.filter(
|
|
IngredientGroceryMatch.ingredient_id == ing_id,
|
|
IngredientGroceryMatch.grocery_item_id == grocery.id,
|
|
)
|
|
.first()
|
|
)
|
|
if existing and existing.source == IngredientMatchSource.MANUAL:
|
|
continue
|
|
confidence = Decimal(str(round(r["confidence"], 3)))
|
|
if existing is None:
|
|
db.add(
|
|
IngredientGroceryMatch(
|
|
ingredient_id=ing_id,
|
|
grocery_item_id=grocery.id,
|
|
confidence=confidence,
|
|
source=IngredientMatchSource.AUTO,
|
|
)
|
|
)
|
|
else:
|
|
existing.confidence = confidence
|
|
existing.source = IngredientMatchSource.AUTO
|
|
written += 1
|
|
|
|
db.commit()
|
|
return written
|