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, ParsedAttributes, 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.normalization import ZERO_WEIGHT_TOKENS, normalize_text, tokenize from app.search_index import LexicalIndex, LexicalScore @dataclass(frozen=True, slots=True) class _RankedCandidate: item: CatalogItem lexical: LexicalScore compatibility: Compatibility 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() 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) brand_item_index: dict[str, list[int]] = defaultdict(list) for index, item in enumerate(self.items): type_index[item.product_type].append(index) if item.attributes.brand is not None: brand_item_index[item.attributes.brand].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() } self.brand_item_index = { brand: tuple(indexes) for brand, indexes in brand_item_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 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 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( 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 _match_brand_only( self, message: str, query_attributes: ParsedAttributes ) -> MatchResult | None: if ( query_attributes.brand is None or not query_attributes.brand_required or query_attributes.model_codes or query_attributes.specificity != 1 ): return None candidate_indices = self.brand_item_index.get(query_attributes.brand, ()) brand_match = self.brand_index.detect(message) if not candidate_indices or brand_match is None: return None compatible = { index: Compatibility( compatible=True, matched_fields=1, total_fields=1, ) for index in candidate_indices } lexical_scores = self.lexical_index.rank(message, "", candidate_indices) ranked = self._apply_preferences( lexical_scores, compatible, cheap=query_attributes.cheap_preference, ) if not ranked: return None if not query_attributes.cheap_preference: # With no product type every item has the same brand evidence. # Document length must not arbitrarily favor short catalog names. ranked.sort(key=lambda candidate: candidate.item.sku) status = "matched" if len(ranked) == 1 else "ambiguous" brand_signal = ProductTypeMatch( canonical="", score=brand_match.score, exact=brand_match.score == 1.0, alias=brand_match.alias, token_start=brand_match.token_start, ) candidates = [ Candidate( sku=candidate.item.sku, confidence=self._confidence( candidate, product_match=brand_signal, 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=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() 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 = cheap_score else: ranking_score = lexical.score ranked.append( _RankedCandidate( item=item, lexical=lexical, compatibility=compatible[lexical.item_index], ranking_score=ranking_score, ) ) if cheap: ranked.sort( key=lambda candidate: ( candidate.item.price, -candidate.lexical.score, candidate.item.sku, ) ) else: ranked.sort( key=lambda candidate: ( -candidate.ranking_score, 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)