Improve sparse and structured catalog matching

app/matcher.py

@@ -15,6 +15,7 @@ from app.attributes import (
 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.normalization import ZERO_WEIGHT_TOKENS, normalize_text, tokenize
 from app.search_index import LexicalIndex, LexicalScore
 
 
@@ -26,6 +27,43 @@ class _RankedCandidate:
     ranking_score: float
 
 
+_STRUCTURED_QUERY_WORDS = frozenset(
+    {
+        "x",
+        "max",
+        "plus",
+        "sds",
+        "t",
+        "z",
+        "v",
+        "w",
+        "в",
+        "ватт",
+        "ватта",
+        "ваттов",
+        "вольт",
+        "вольта",
+        "вольтов",
+        "зуб",
+        "зуба",
+        "зубов",
+        "зубьев",
+        "м",
+        "метр",
+        "метра",
+        "метров",
+        "миллиметр",
+        "миллиметра",
+        "миллиметров",
+        "мм",
+        "размер",
+        "размера",
+        "размером",
+        "вт",
+    }
+)
+
+
 class CatalogMatcher:
     def __init__(self, catalog_path: Path) -> None:
         self.alias_index = ProductAliasIndex()
@@ -78,15 +116,46 @@ class CatalogMatcher:
                         alias="model_code",
                         token_start=0,
                     )
+            if product_match is None and provisional_attributes.teeth is not None:
+                # A tooth count is type-specific evidence in this domain: only
+                # saw blades carry it. This safely supports requests such as
+                # "210 мм 24 зуба" without requiring the word "диск".
+                product_match = ProductTypeMatch(
+                    canonical="диск",
+                    score=1.0,
+                    exact=True,
+                    alias="tooth_count",
+                    token_start=0,
+                )
             if product_match is None:
                 brand_result = self._match_brand_only(message, provisional_attributes)
                 if brand_result is not None:
                     return brand_result
-                # 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=[])
+                structured_result = self._match_structured_only(
+                    message, provisional_attributes
+                )
+                if structured_result is not None:
+                    return structured_result
+                # Trigrams used to rank only inside a type selected by aliases.
+                # Let them infer that missing type as well, but only when the
+                # leading catalog rows agree and share strong word-level evidence.
+                # This handles "удар" -> "ударная дрель" while
+                # keeping numeric coincidences and arbitrary text out.
+                if detect_service_intent(message) is None:
+                    lexical_inference = self.lexical_index.infer_product_type(message)
+                    if lexical_inference is not None:
+                        product_match = ProductTypeMatch(
+                            canonical=lexical_inference.product_type,
+                            score=lexical_inference.score,
+                            exact=lexical_inference.query_word
+                            == lexical_inference.catalog_word,
+                            alias=lexical_inference.catalog_word,
+                            token_start=0,
+                        )
+                if product_match is None:
+                    return MatchResult(
+                        message=message, status="not_found", candidates=[]
+                    )
 
         product_type = product_match.canonical
         query_attributes = parse_attributes(
@@ -154,6 +223,7 @@ class CatalogMatcher:
             query_attributes.brand is None
             or not query_attributes.brand_required
             or query_attributes.model_codes
+            or query_attributes.specificity != 1
         ):
             return None
 
@@ -207,6 +277,119 @@ class CatalogMatcher:
         ]
         return MatchResult(message=message, status=status, candidates=candidates)
 
+    def _match_structured_only(
+        self, message: str, query_attributes: ParsedAttributes
+    ) -> MatchResult | None:
+        if not self._is_structured_only_query(message, query_attributes):
+            return None
+
+        compatible: dict[int, Compatibility] = {}
+        for index, item in enumerate(self.items):
+            comparison = compare_attributes(
+                item.product_type,
+                query_attributes,
+                item.attributes,
+                candidate_unit=item.unit,
+            )
+            if comparison.compatible and comparison.matched_fields > 0:
+                compatible[index] = comparison
+
+        if not compatible:
+            return None
+
+        lexical_scores = self.lexical_index.rank(
+            message, "", tuple(compatible.keys())
+        )
+        ranked = self._apply_preferences(
+            lexical_scores,
+            compatible,
+            cheap=query_attributes.cheap_preference,
+        )
+        if not ranked:
+            return None
+
+        status = "matched" if len(ranked) == 1 else "ambiguous"
+        selected = self._select_diverse_product_types(ranked, limit=3)
+        structured_signal = ProductTypeMatch(
+            canonical="",
+            score=0.90,
+            exact=True,
+            alias="structured_attributes",
+            token_start=0,
+        )
+        candidates = [
+            Candidate(
+                sku=candidate.item.sku,
+                confidence=self._confidence(
+                    candidate,
+                    product_match=structured_signal,
+                    candidate_count=len(ranked),
+                    specificity=query_attributes.specificity,
+                    status=status,
+                    rank=rank,
+                ),
+            )
+            for rank, candidate in enumerate(selected)
+        ]
+        return MatchResult(message=message, status=status, candidates=candidates)
+
+    def _is_structured_only_query(
+        self, message: str, query_attributes: ParsedAttributes
+    ) -> bool:
+        has_strong_evidence = any(
+            (
+                bool(query_attributes.dimensions),
+                bool(query_attributes.mm_values),
+                bool(query_attributes.meter_values),
+                query_attributes.thread_size is not None,
+                query_attributes.profile is not None,
+                query_attributes.grit is not None,
+                query_attributes.voltage_v is not None,
+                query_attributes.power_w is not None,
+                query_attributes.teeth is not None,
+                query_attributes.brand is not None
+                and query_attributes.brand_required,
+                bool(query_attributes.model_codes),
+                query_attributes.sds_type is not None,
+            )
+        )
+        if not has_strong_evidence:
+            return False
+
+        allowed_words = set(ZERO_WEIGHT_TOKENS) | set(_STRUCTURED_QUERY_WORDS)
+        brand_match = self.brand_index.detect(message)
+        if brand_match is not None:
+            allowed_words.update(tokenize(brand_match.alias))
+
+        for token in tokenize(normalize_text(message), already_normalized=True):
+            if any(character.isdigit() for character in token):
+                continue
+            if token in allowed_words:
+                continue
+            if token.startswith(("бюджет", "дешев", "недорог")):
+                continue
+            return False
+        return True
+
+    @staticmethod
+    def _select_diverse_product_types(
+        ranked: list[_RankedCandidate], *, limit: int
+    ) -> list[_RankedCandidate]:
+        if len({candidate.item.product_type for candidate in ranked}) == 1:
+            return ranked[:limit]
+
+        selected: list[_RankedCandidate] = []
+        product_types: set[str] = set()
+
+        for candidate in ranked:
+            if candidate.item.product_type in product_types:
+                continue
+            selected.append(candidate)
+            product_types.add(candidate.item.product_type)
+            if len(selected) == limit:
+                break
+        return selected
+
     def _model_candidates(self, model_codes: frozenset[str]) -> frozenset[int]:
         if not model_codes:
             return frozenset()