Validate product attributes and decide match status

app/matcher.py

@@ -0,0 +1,218 @@
+from __future__ import annotations
+
+from collections import defaultdict
+from dataclasses import dataclass
+from pathlib import Path
+
+from app.aliases import ProductAliasIndex, ProductTypeMatch
+from app.attributes import BrandIndex, Compatibility, compare_attributes, parse_attributes
+from app.catalog import CatalogItem, build_catalog, load_raw_catalog
+from app.intents import detect_service_intent
+from app.models import Candidate, MatchResult
+from app.search_index import LexicalIndex, LexicalScore
+
+
+@dataclass(frozen=True, slots=True)
+class _RankedCandidate:
+    item: CatalogItem
+    lexical: LexicalScore
+    compatibility: Compatibility
+    ranking_score: float
+
+
+class CatalogMatcher:
+    def __init__(self, catalog_path: Path) -> None:
+        self.alias_index = ProductAliasIndex()
+        self.brand_index = BrandIndex()
+        self.items = build_catalog(
+            load_raw_catalog(catalog_path), self.alias_index, self.brand_index
+        )
+        self.lexical_index = LexicalIndex(self.items)
+
+        type_index: dict[str, list[int]] = defaultdict(list)
+        model_index: dict[str, set[int]] = defaultdict(set)
+        for index, item in enumerate(self.items):
+            type_index[item.product_type].append(index)
+            for model_code in item.attributes.model_codes:
+                model_index[model_code].add(index)
+        self.type_index = {
+            product_type: tuple(indexes) for product_type, indexes in type_index.items()
+        }
+        self.model_index = {
+            model: frozenset(indexes) for model, indexes in model_index.items()
+        }
+
+    def match(self, message: str) -> MatchResult:
+        if not message or not message.strip():
+            return MatchResult(message=message, status="not_found", candidates=[])
+
+        product_match = self.alias_index.detect(message)
+        provisional_attributes = parse_attributes(
+            message,
+            product_match.canonical if product_match else "",
+            brand_index=self.brand_index,
+        )
+
+        if product_match is None:
+            model_candidates = self._model_candidates(provisional_attributes.model_codes)
+            if model_candidates:
+                product_types = {self.items[index].product_type for index in model_candidates}
+                if len(product_types) == 1:
+                    product_type = next(iter(product_types))
+                    product_match = ProductTypeMatch(
+                        canonical=product_type,
+                        score=1.0,
+                        exact=True,
+                        alias="model_code",
+                        token_start=0,
+                    )
+            if product_match is None:
+                # Product evidence always has priority. Service intent detection
+                # is only a diagnostic fallback; both unknown and service
+                # messages share the required not_found response.
+                detect_service_intent(message)
+                return MatchResult(message=message, status="not_found", candidates=[])
+
+        product_type = product_match.canonical
+        query_attributes = parse_attributes(
+            message,
+            product_type,
+            brand_index=self.brand_index,
+        )
+        candidate_indices = self.type_index.get(product_type, ())
+
+        model_candidates = self._model_candidates(query_attributes.model_codes)
+        if model_candidates:
+            candidate_indices = tuple(
+                index for index in candidate_indices if index in model_candidates
+            )
+
+        compatible: dict[int, Compatibility] = {}
+        for index in candidate_indices:
+            item = self.items[index]
+            comparison = compare_attributes(
+                product_type,
+                query_attributes,
+                item.attributes,
+                candidate_unit=item.unit,
+            )
+            if comparison.compatible:
+                compatible[index] = comparison
+
+        if not compatible:
+            return MatchResult(message=message, status="not_found", candidates=[])
+
+        lexical_scores = self.lexical_index.rank(
+            message, product_type, tuple(compatible.keys())
+        )
+        ranked = self._apply_preferences(
+            lexical_scores, compatible, cheap=query_attributes.cheap_preference
+        )
+        if not ranked:
+            return MatchResult(message=message, status="not_found", candidates=[])
+
+        status = "matched" if len(ranked) == 1 else "ambiguous"
+        response_candidates = [
+            Candidate(
+                sku=candidate.item.sku,
+                confidence=self._confidence(
+                    candidate,
+                    product_match=product_match,
+                    candidate_count=len(ranked),
+                    specificity=query_attributes.specificity,
+                    status=status,
+                    rank=rank,
+                ),
+            )
+            for rank, candidate in enumerate(ranked[:3])
+        ]
+        return MatchResult(
+            message=message,
+            status=status,
+            candidates=response_candidates,
+        )
+
+    def _model_candidates(self, model_codes: frozenset[str]) -> frozenset[int]:
+        if not model_codes:
+            return frozenset()
+        sets = [self.model_index.get(code, frozenset()) for code in model_codes]
+        if not sets or any(not indexes for indexes in sets):
+            return frozenset()
+        result = set(sets[0])
+        for indexes in sets[1:]:
+            result.intersection_update(indexes)
+        return frozenset(result)
+
+    def _apply_preferences(
+        self,
+        lexical_scores: list[LexicalScore],
+        compatible: dict[int, Compatibility],
+        *,
+        cheap: bool,
+    ) -> list[_RankedCandidate]:
+        if not lexical_scores:
+            return []
+
+        prices = [self.items[score.item_index].price for score in lexical_scores]
+        minimum_price = min(prices)
+        maximum_price = max(prices)
+        price_range = maximum_price - minimum_price
+
+        ranked: list[_RankedCandidate] = []
+        for lexical in lexical_scores:
+            item = self.items[lexical.item_index]
+            if cheap:
+                cheap_score = (
+                    1.0
+                    if price_range == 0
+                    else float((maximum_price - item.price) / price_range)
+                )
+                ranking_score = lexical.score * 0.68 + cheap_score * 0.32
+            else:
+                ranking_score = lexical.score
+            ranked.append(
+                _RankedCandidate(
+                    item=item,
+                    lexical=lexical,
+                    compatibility=compatible[lexical.item_index],
+                    ranking_score=ranking_score,
+                )
+            )
+
+        ranked.sort(
+            key=lambda candidate: (
+                -candidate.ranking_score,
+                candidate.item.price if cheap else candidate.item.sku,
+            )
+        )
+        return ranked
+
+    @staticmethod
+    def _confidence(
+        candidate: _RankedCandidate,
+        *,
+        product_match: ProductTypeMatch,
+        candidate_count: int,
+        specificity: int,
+        status: str,
+        rank: int,
+    ) -> float:
+        attribute_signal = (
+            candidate.compatibility.score
+            if candidate.compatibility.total_fields > 0
+            else 0.55
+        )
+        uniqueness_signal = 1.0 if candidate_count == 1 else max(0.35, 0.72 - rank * 0.10)
+        specificity_signal = min(1.0, 0.35 + specificity * 0.16)
+        value = (
+            product_match.score * 0.30
+            + candidate.lexical.score * 0.27
+            + attribute_signal * 0.23
+            + uniqueness_signal * 0.12
+            + specificity_signal * 0.08
+        )
+        if status == "matched":
+            value = max(value, 0.82)
+        else:
+            value = min(value, 0.84)
+        return round(max(0.01, min(0.99, value)), 4)