303 lines
11 KiB
Python
303 lines
11 KiB
Python
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.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)
|
|
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:
|
|
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=[])
|
|
|
|
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
|
|
):
|
|
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 _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)
|