326 lines
14 KiB
Python
326 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass
|
|
|
|
from app.normalization import meaningful_tokens, normalize_text, tokenize, word_tokens
|
|
from app.typo import character_trigrams, fuzzy_threshold, word_similarity
|
|
|
|
# Canonical product concepts are intentionally separate from SKU aliases. The
|
|
# list contains domain vocabulary only; it never maps a phrase directly to an
|
|
# individual product.
|
|
PRODUCT_ALIASES = [
|
|
{
|
|
"canonical": "ушм",
|
|
"aliases": [
|
|
"ушм",
|
|
"болгарка",
|
|
"углошлифовальная машина",
|
|
"угловая шлифмашина",
|
|
"угловая шлифовальная машина",
|
|
],
|
|
},
|
|
{
|
|
"canonical": "шуруповерт",
|
|
"aliases": [
|
|
"шуруповерт",
|
|
"шуруповёрт",
|
|
"шурик",
|
|
"дрель шуруповерт",
|
|
"дрель-шуруповерт",
|
|
],
|
|
},
|
|
{
|
|
"canonical": "перфоратор",
|
|
"aliases": ["перфоратор", "перф", "перфоратор sds"],
|
|
},
|
|
{"canonical": "саморез", "aliases": ["саморез", "саморезы", "саморезов"]},
|
|
{
|
|
"canonical": "труба",
|
|
"aliases": [
|
|
"труба",
|
|
"трубы",
|
|
"проф труба",
|
|
"профтруба",
|
|
"труба профильная",
|
|
"профильная труба",
|
|
],
|
|
},
|
|
{
|
|
"canonical": "кабель",
|
|
"aliases": [
|
|
"кабель",
|
|
"кабеля",
|
|
"кабелей",
|
|
"ввг",
|
|
"ввгнг",
|
|
"ввгнг ls",
|
|
"ввгнг лс",
|
|
"шввп",
|
|
"пвс",
|
|
],
|
|
},
|
|
{"canonical": "провод", "aliases": ["провод", "провода", "проводов", "пугв"]},
|
|
{"canonical": "бита", "aliases": ["бита", "биты", "биту", "биты для шуруповерта"]},
|
|
{
|
|
"canonical": "держатель бит",
|
|
"aliases": ["держатель бит", "битодержатель", "магнитный держатель"],
|
|
},
|
|
{"canonical": "адаптер бит", "aliases": ["адаптер для бит", "адаптер бит"]},
|
|
{"canonical": "сверло", "aliases": ["сверло", "сверла", "сверел", "сверлить"]},
|
|
{"canonical": "бур", "aliases": ["бур", "буры", "сдс бур", "sds бур"]},
|
|
{
|
|
"canonical": "диск",
|
|
"aliases": [
|
|
"диск",
|
|
"диски",
|
|
"дисков",
|
|
"круг отрезной",
|
|
"отрезной круг",
|
|
"круг зачистной",
|
|
"зачистной круг",
|
|
"пильный круг",
|
|
],
|
|
},
|
|
{
|
|
"canonical": "круг",
|
|
"aliases": ["круг", "круг шлифовальный", "шлифовальный круг", "круг на липучке"],
|
|
},
|
|
{"canonical": "дрель", "aliases": ["дрель", "дрели", "ударная дрель"]},
|
|
{"canonical": "лобзик", "aliases": ["лобзик", "электролобзик"]},
|
|
{"canonical": "фен", "aliases": ["строительный фен", "термофен", "фен"]},
|
|
{"canonical": "перчатки", "aliases": ["перчатки", "перчаток"]},
|
|
{"canonical": "изолента", "aliases": ["изолента", "изоляционная лента"]},
|
|
{"canonical": "скотч", "aliases": ["скотч", "малярный скотч"]},
|
|
{
|
|
"canonical": "стяжка",
|
|
"aliases": [
|
|
"стяжка",
|
|
"стяжки",
|
|
"кабельная стяжка",
|
|
"нейлоновая стяжка",
|
|
"хомут пластиковый",
|
|
"хомуты пластиковые",
|
|
"пластиковый хомут",
|
|
"пластиковые хомуты",
|
|
],
|
|
},
|
|
{"canonical": "пена", "aliases": ["пена", "монтажная пена"]},
|
|
{"canonical": "герметик", "aliases": ["герметик", "герметики"]},
|
|
{"canonical": "нож", "aliases": ["нож", "строительный нож"]},
|
|
{"canonical": "лезвия", "aliases": ["лезвие", "лезвия", "сменные лезвия"]},
|
|
{"canonical": "рулетка", "aliases": ["рулетка", "рулетки"]},
|
|
{"canonical": "карандаш", "aliases": ["карандаш", "строительный карандаш"]},
|
|
{"canonical": "маркер", "aliases": ["маркер", "маркеры"]},
|
|
{"canonical": "респиратор", "aliases": ["респиратор", "респираторы"]},
|
|
{"canonical": "очки", "aliases": ["защитные очки", "очки"]},
|
|
{"canonical": "мешки", "aliases": ["мешок", "мешки", "мешки для мусора"]},
|
|
{"canonical": "кисть", "aliases": ["кисть", "кисти", "малярная кисть"]},
|
|
{"canonical": "валик", "aliases": ["валик", "валики", "малярный валик"]},
|
|
{"canonical": "ванночка", "aliases": ["ванночка", "малярная ванночка"]},
|
|
{"canonical": "удлинитель", "aliases": ["удлинитель", "удлинители"]},
|
|
{
|
|
"canonical": "шкурка",
|
|
"aliases": [
|
|
"шкурка",
|
|
"шлифовальная шкурка",
|
|
"наждачка",
|
|
"наждачная бумага",
|
|
],
|
|
},
|
|
{"canonical": "лента фум", "aliases": ["лента фум", "фум лента", "фумка"]},
|
|
{
|
|
"canonical": "хомут",
|
|
"aliases": ["хомут", "хомуты", "червячный хомут", "металлический хомут"],
|
|
},
|
|
{
|
|
"canonical": "гипсокартон",
|
|
"aliases": [
|
|
"гкл",
|
|
"гипсокартон",
|
|
"гипсокартонный лист",
|
|
"лист гипсокартонный",
|
|
],
|
|
},
|
|
{"canonical": "профиль", "aliases": ["профиль", "профили", "профиль для гкл"]},
|
|
{"canonical": "подвес", "aliases": ["подвес", "прямой подвес"]},
|
|
{"canonical": "соединитель", "aliases": ["соединитель", "краб", "соединитель краб"]},
|
|
{"canonical": "уголок", "aliases": ["уголок", "перфорированный уголок"]},
|
|
{"canonical": "лента серпянка", "aliases": ["серпянка", "лента серпянка"]},
|
|
{"canonical": "лента демпферная", "aliases": ["демпферная лента", "лента демпферная"]},
|
|
{"canonical": "болт", "aliases": ["болт", "болты", "болтов"]},
|
|
{"canonical": "гайка", "aliases": ["гайка", "гайки", "гаек"]},
|
|
{"canonical": "шайба", "aliases": ["шайба", "шайбы", "шайб"]},
|
|
{"canonical": "шпилька", "aliases": ["шпилька", "шпильки", "резьбовая шпилька"]},
|
|
{"canonical": "анкер", "aliases": ["анкер", "анкеры", "анкеров"]},
|
|
{"canonical": "гвозди", "aliases": ["гвоздь", "гвозди", "гвоздей"]},
|
|
{
|
|
"canonical": "дюбель",
|
|
"aliases": [
|
|
"дюбель",
|
|
"дюбели",
|
|
"дюбелей",
|
|
"дюбель гвоздь",
|
|
"дюбель-гвоздь",
|
|
],
|
|
},
|
|
{"canonical": "ключ", "aliases": ["ключ", "ключи", "рожковый ключ", "разводной ключ"]},
|
|
{"canonical": "отвертка", "aliases": ["отвертка", "отвёртка", "отвертки"]},
|
|
{"canonical": "молоток", "aliases": ["молоток", "молотки"]},
|
|
{"canonical": "ножовка", "aliases": ["ножовка", "ножовки"]},
|
|
{"canonical": "плоскогубцы", "aliases": ["плоскогубцы"]},
|
|
{"canonical": "бокорезы", "aliases": ["бокорезы", "бокорез"]},
|
|
{"canonical": "пассатижи", "aliases": ["пассатижи", "клещи переставные"]},
|
|
{"canonical": "уровень", "aliases": ["уровень", "уровни"]},
|
|
{"canonical": "угольник", "aliases": ["угольник", "угольники"]},
|
|
{"canonical": "степлер", "aliases": ["степлер", "мебельный степлер"]},
|
|
{"canonical": "скобы", "aliases": ["скоба", "скобы", "скобы для степлера"]},
|
|
]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ProductTypeMatch:
|
|
canonical: str
|
|
score: float
|
|
exact: bool
|
|
alias: str
|
|
token_start: int
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _AliasRecord:
|
|
canonical: str
|
|
normalized: str
|
|
tokens: tuple[str, ...]
|
|
|
|
|
|
class ProductAliasIndex:
|
|
def __init__(self) -> None:
|
|
alias_to_canonical: dict[str, str] = {}
|
|
canonical_to_aliases: dict[str, list[str]] = defaultdict(list)
|
|
records: list[_AliasRecord] = []
|
|
single_token_records: list[_AliasRecord] = []
|
|
trigram_postings: dict[str, set[int]] = defaultdict(set)
|
|
|
|
for item in PRODUCT_ALIASES:
|
|
canonical = normalize_text(item["canonical"])
|
|
for raw_alias in item["aliases"]:
|
|
alias = normalize_text(raw_alias)
|
|
previous = alias_to_canonical.get(alias)
|
|
if previous is not None and previous != canonical:
|
|
raise ValueError(
|
|
f"Alias {raw_alias!r} points to both {previous!r} and {canonical!r}"
|
|
)
|
|
alias_to_canonical[alias] = canonical
|
|
canonical_to_aliases[canonical].append(alias)
|
|
|
|
record = _AliasRecord(canonical, alias, tokenize(alias, already_normalized=True))
|
|
records.append(record)
|
|
if len(record.tokens) == 1 and record.tokens[0].isalpha():
|
|
record_index = len(single_token_records)
|
|
single_token_records.append(record)
|
|
for trigram in character_trigrams(record.tokens[0]):
|
|
trigram_postings[trigram].add(record_index)
|
|
|
|
self.alias_to_canonical = alias_to_canonical
|
|
self.canonical_to_aliases = {
|
|
canonical: tuple(dict.fromkeys(aliases))
|
|
for canonical, aliases in canonical_to_aliases.items()
|
|
}
|
|
self._records = tuple(records)
|
|
self._single_token_records = tuple(single_token_records)
|
|
self._trigram_postings = {
|
|
trigram: frozenset(indexes) for trigram, indexes in trigram_postings.items()
|
|
}
|
|
|
|
def detect(self, text: str) -> ProductTypeMatch | None:
|
|
normalized = normalize_text(text)
|
|
tokens = tokenize(normalized, already_normalized=True)
|
|
exact = self._find_exact(tokens, prefix_only=False)
|
|
if exact is not None:
|
|
return exact
|
|
return self._find_fuzzy(tokens)
|
|
|
|
def detect_catalog_type(self, text: str) -> str:
|
|
normalized = normalize_text(text)
|
|
tokens = tokenize(normalized, already_normalized=True)
|
|
exact = self._find_exact(tokens, prefix_only=True)
|
|
if exact is None:
|
|
raise ValueError(f"No product type alias matches catalog name: {text!r}")
|
|
return exact.canonical
|
|
|
|
def aliases_for(self, canonical: str) -> tuple[str, ...]:
|
|
return self.canonical_to_aliases.get(canonical, ())
|
|
|
|
def _find_exact(
|
|
self, tokens: tuple[str, ...], *, prefix_only: bool
|
|
) -> ProductTypeMatch | None:
|
|
matches: list[ProductTypeMatch] = []
|
|
for record in self._records:
|
|
alias_length = len(record.tokens)
|
|
if alias_length == 0 or alias_length > len(tokens):
|
|
continue
|
|
starts = (0,) if prefix_only else range(len(tokens) - alias_length + 1)
|
|
for start in starts:
|
|
if tokens[start : start + alias_length] == record.tokens:
|
|
matches.append(
|
|
ProductTypeMatch(
|
|
canonical=record.canonical,
|
|
score=1.0,
|
|
exact=True,
|
|
alias=record.normalized,
|
|
token_start=start,
|
|
)
|
|
)
|
|
break
|
|
|
|
if not matches:
|
|
return None
|
|
matches.sort(
|
|
key=lambda match: (
|
|
-len(tokenize(match.alias, already_normalized=True)),
|
|
match.token_start,
|
|
-len(match.alias),
|
|
)
|
|
)
|
|
return matches[0]
|
|
|
|
def _find_fuzzy(self, tokens: tuple[str, ...]) -> ProductTypeMatch | None:
|
|
query_words = word_tokens(meaningful_tokens(tokens))
|
|
best: ProductTypeMatch | None = None
|
|
|
|
for token_start, query_token in enumerate(query_words):
|
|
if len(query_token) < 4:
|
|
continue
|
|
|
|
candidate_indexes: set[int] = set()
|
|
for trigram in character_trigrams(query_token):
|
|
candidate_indexes.update(self._trigram_postings.get(trigram, ()))
|
|
if not candidate_indexes:
|
|
continue
|
|
|
|
for candidate_index in candidate_indexes:
|
|
record = self._single_token_records[candidate_index]
|
|
alias_token = record.tokens[0]
|
|
if abs(len(query_token) - len(alias_token)) > max(4, len(alias_token) // 2):
|
|
continue
|
|
score = word_similarity(query_token, alias_token)
|
|
required = fuzzy_threshold(max(len(query_token), len(alias_token)))
|
|
if score < required:
|
|
continue
|
|
|
|
match = ProductTypeMatch(
|
|
canonical=record.canonical,
|
|
score=score,
|
|
exact=False,
|
|
alias=record.normalized,
|
|
token_start=token_start,
|
|
)
|
|
if best is None or (match.score, -match.token_start) > (
|
|
best.score,
|
|
-best.token_start,
|
|
):
|
|
best = match
|
|
|
|
return best
|