""" Matches extracted item names against a product catalogue (CSV with sku_code, product_name, category, unit) using fuzzy string matching. """ import pandas as pd from rapidfuzz import process, fuzz # Below this score (0-100), a match is flagged as low-confidence for human review. DEFAULT_CONFIDENCE_THRESHOLD = 70 def load_catalogue(csv_path_or_buffer) -> pd.DataFrame: """Load a catalogue CSV. Expects columns: sku_code, product_name, category, unit.""" df = pd.read_csv(csv_path_or_buffer) required_cols = {"sku_code", "product_name", "category", "unit"} missing = required_cols - set(c.strip() for c in df.columns) if missing: raise ValueError( f"Catalogue is missing required columns: {missing}. " f"Found columns: {list(df.columns)}" ) df["sku_code"] = df["sku_code"].astype(str) return df def match_item(item_name: str, catalogue: pd.DataFrame, threshold: int = DEFAULT_CONFIDENCE_THRESHOLD) -> dict: """ Find the best matching catalogue entry for a given item name. Returns a dict with sku_code, matched_name, category, unit, confidence, and needs_review (bool). """ if catalogue is None or catalogue.empty or not item_name: return { "sku_code": None, "matched_name": None, "category": None, "unit": None, "confidence": 0, "needs_review": True, } choices = catalogue["product_name"].tolist() result = process.extractOne(item_name, choices, scorer=fuzz.WRatio) if result is None: return { "sku_code": None, "matched_name": None, "category": None, "unit": None, "confidence": 0, "needs_review": True, } matched_name, score, idx = result row = catalogue.iloc[idx] return { "sku_code": row["sku_code"], "matched_name": row["product_name"], "category": row["category"], "unit": row["unit"], "confidence": round(score, 1), "needs_review": score < threshold, } def enrich_items(items: list[dict], catalogue: pd.DataFrame, threshold: int = DEFAULT_CONFIDENCE_THRESHOLD) -> list[dict]: """ Take extracted items (from extraction.py) and attach catalogue matches. Returns a new list of dicts combining original extraction with match info. """ enriched = [] for item in items: match = match_item(item.get("item_name", ""), catalogue, threshold) enriched.append({ "raw_text": item.get("raw_text"), "extracted_name": item.get("item_name"), "quantity": item.get("quantity"), "unit": item.get("unit"), "sku_code": match["sku_code"], "matched_product": match["matched_name"], "category": match["category"], "catalogue_unit": match["unit"], "match_confidence": match["confidence"], "needs_review": match["needs_review"], }) return enriched