1388 lines
44 KiB
Python
1388 lines
44 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""跨電商商品身份比對工具。
|
||
|
||
這裡處理「是否為同一個商品」；價格只作為 sanity check，不能主導配對。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import unicodedata
|
||
from dataclasses import dataclass
|
||
from difflib import SequenceMatcher
|
||
from typing import Iterable, Optional
|
||
|
||
|
||
NOISE_PHRASES = (
|
||
"momo",
|
||
"pchome",
|
||
"24h",
|
||
"官方直營",
|
||
"官方",
|
||
"公司貨",
|
||
"台灣公司貨",
|
||
"專櫃公司貨",
|
||
"正貨",
|
||
"原廠",
|
||
"限時",
|
||
"特惠",
|
||
"優惠",
|
||
"超值",
|
||
"加贈",
|
||
"贈品",
|
||
"送禮",
|
||
"送",
|
||
"買一送一",
|
||
"買1送1",
|
||
"任選",
|
||
"即期品",
|
||
"福利品",
|
||
"預購",
|
||
"免運",
|
||
"熱銷",
|
||
"人氣",
|
||
"必買",
|
||
"推薦",
|
||
"新品",
|
||
"升級版",
|
||
"經典",
|
||
"獨家",
|
||
"囤貨組",
|
||
"超值組",
|
||
"優惠組",
|
||
"分享包",
|
||
"組合",
|
||
"多款可選",
|
||
"多款任選",
|
||
"任選多款",
|
||
"多色可選",
|
||
"色號可選",
|
||
"平行輸入",
|
||
"大容量",
|
||
)
|
||
|
||
GENERIC_TOKENS = {
|
||
"官方",
|
||
"直營",
|
||
"公司貨",
|
||
"專櫃",
|
||
"正貨",
|
||
"原廠",
|
||
"限時",
|
||
"特惠",
|
||
"優惠",
|
||
"超值",
|
||
"加贈",
|
||
"贈品",
|
||
"送禮",
|
||
"即期品",
|
||
"新品",
|
||
"升級版",
|
||
"經典",
|
||
"人氣",
|
||
"熱銷",
|
||
"必買",
|
||
"推薦",
|
||
"組",
|
||
"入",
|
||
"瓶",
|
||
"盒",
|
||
"包",
|
||
"片",
|
||
"支",
|
||
"條",
|
||
"件",
|
||
"ml",
|
||
"g",
|
||
"mg",
|
||
"la",
|
||
"paris",
|
||
"多款",
|
||
"可選",
|
||
"任選",
|
||
"平行輸入",
|
||
"大容量",
|
||
"日本",
|
||
"韓國",
|
||
"澳洲",
|
||
"法國",
|
||
"英國",
|
||
"美國",
|
||
}
|
||
|
||
SEARCH_NOISE_PHRASES = (
|
||
"新品上市",
|
||
"全新上市",
|
||
"任選一款",
|
||
"任選1款",
|
||
"任選一色",
|
||
"任選1色",
|
||
"多款任選",
|
||
"多款可選",
|
||
"色號可選",
|
||
"香味可選",
|
||
"口味可選",
|
||
"送精美紙袋",
|
||
"精美紙袋",
|
||
"交換禮物",
|
||
"聖誕禮物",
|
||
"母親節",
|
||
"父親節",
|
||
"情人節",
|
||
"外出清潔",
|
||
"卸除髒汙",
|
||
"卸除防曬",
|
||
"卸防曬",
|
||
"韓國彩妝",
|
||
"水光感",
|
||
"官方直營",
|
||
"官方",
|
||
"經典款",
|
||
)
|
||
|
||
SEARCH_NOISE_TOKENS = {
|
||
"一款",
|
||
"1款",
|
||
"一色",
|
||
"1色",
|
||
"上市",
|
||
"全新",
|
||
"新品",
|
||
"香味",
|
||
"口味",
|
||
"味道",
|
||
"顏色",
|
||
"色號",
|
||
"紙袋",
|
||
"禮物",
|
||
"清潔",
|
||
"髒汙",
|
||
"防曬",
|
||
"彩妝",
|
||
"水光感",
|
||
"保濕",
|
||
"抗老",
|
||
"超品日",
|
||
"經典款",
|
||
"款",
|
||
"pdrn",
|
||
}
|
||
|
||
SEARCH_IDENTITY_ANCHORS = (
|
||
"絕對完美永生玫瑰逆齡乳霜",
|
||
"永生玫瑰逆齡乳霜",
|
||
"永生玫瑰霜",
|
||
"玫瑰精露",
|
||
"玫瑰霜",
|
||
"青春敷面膜",
|
||
"長效潤膚霜",
|
||
"小黑瓶",
|
||
"免用水潔淨液",
|
||
"身體按摩精油",
|
||
"按摩精油",
|
||
"擴香補充瓶",
|
||
"擴香瓶",
|
||
"全面修復霜",
|
||
"修復霜",
|
||
"護膚膏",
|
||
"屁屁噴",
|
||
"身體乳",
|
||
"緊實乳",
|
||
"潔膚露",
|
||
"潔淨液",
|
||
"護甲油",
|
||
"指甲油",
|
||
"美甲片",
|
||
"唇凍",
|
||
"唇釉",
|
||
"唇膏",
|
||
"粉底棒",
|
||
"遮瑕棒",
|
||
"化妝水",
|
||
"精華液",
|
||
"精華",
|
||
"面膜",
|
||
"乳液",
|
||
"乳霜",
|
||
"面霜",
|
||
"精油",
|
||
"水氧機",
|
||
"香氛機",
|
||
)
|
||
|
||
SEARCH_BROAD_ANCHORS = {
|
||
"乳霜",
|
||
"面霜",
|
||
"面膜",
|
||
"精華",
|
||
"乳液",
|
||
"精油",
|
||
}
|
||
|
||
SEARCH_AMBIGUOUS_PRODUCT_TERMS = {
|
||
"保護膜",
|
||
"保護貼",
|
||
}
|
||
|
||
BRAND_ALIAS_OVERRIDES = {
|
||
"clarins": ("克蘭詩", "clarins"),
|
||
"nars": ("nars",),
|
||
"relove": ("relove",),
|
||
"stadler form": ("stadler form", "stadlerform"),
|
||
"cetaphil": ("舒特膚", "cetaphil"),
|
||
"sisley": ("希思黎", "sisley"),
|
||
"gennies": ("奇妮", "gennies"),
|
||
"uruhimemomoko": ("潤姬桃子", "uruhimemomoko", "uruhime momoko"),
|
||
"arau baby": ("arau baby", "arau", "愛樂寶", "saraya"),
|
||
"sebamed": ("sebamed", "施巴"),
|
||
"shu uemura": ("shu uemura", "shuuemura", "植村秀"),
|
||
"johnsons": ("johnsons", "johnson's", "johnson", "嬌生"),
|
||
"gillette": ("gillette", "吉列"),
|
||
"schick": ("schick", "舒適牌"),
|
||
"obge": ("obge",),
|
||
"vaseline": ("vaseline", "凡士林"),
|
||
"eaoron": ("eaoron",),
|
||
"kameria": ("kameria", "凱蜜菈"),
|
||
"cocodor": ("cocodor",),
|
||
"peripera": ("peripera",),
|
||
"solone": ("solone",),
|
||
"im meme": ("im meme", "i'm meme", "i’m meme"),
|
||
"febreze": ("febreze", "風倍清"),
|
||
"jo malone": ("jo malone",),
|
||
"prada": ("prada", "普拉達"),
|
||
}
|
||
|
||
PRODUCT_TYPES = {
|
||
"精華": ("精華", "精華液", "essence", "serum", "安瓶"),
|
||
"化妝水": ("化妝水", "機能水", "toner", "lotion"),
|
||
"乳液": ("乳液", "emulsion", "milk"),
|
||
"面霜": ("面霜", "乳霜", "霜", "cream"),
|
||
"防曬": ("防曬", "spf", "uv", "sunscreen"),
|
||
"洗面乳": ("洗面乳", "洗顏", "潔面", "cleanser", "foam"),
|
||
"面膜": ("面膜", "mask"),
|
||
"眼霜": ("眼霜", "眼部", "眼膜", "eye"),
|
||
"卸妝": ("卸妝", "cleansing", "remover"),
|
||
"粉底": ("粉底", "粉霜", "粉凝露", "foundation"),
|
||
"蜜粉": ("蜜粉", "powder"),
|
||
"精油": ("精油", "香氛", "擴香"),
|
||
"保健": ("錠", "膠囊", "粉", "飲", "包", "健康食品"),
|
||
}
|
||
|
||
COUNT_UNITS = {"入", "組", "瓶", "支", "條", "盒", "包", "袋", "片", "顆", "粒", "錠", "枚", "件", "罐", "杯", "本", "刀把", "刀片", "刀頭", "蕊"}
|
||
COUNT_UNIT_PATTERN = r"(?:刀把|刀片|刀頭|入|組|瓶|支|條|盒|包|袋|片|顆|粒|錠|枚|件|罐|杯|本|蕊)"
|
||
PIECE_UNITS = {"包", "袋", "片", "顆", "粒", "錠", "枚"}
|
||
CONTAINER_UNITS = {"入", "組", "盒", "罐", "杯", "本", "瓶", "支", "條", "件"}
|
||
COUNT_UNIT_FAMILIES = {
|
||
"刀片": "blade",
|
||
"刀頭": "blade",
|
||
"蕊": "refill",
|
||
}
|
||
ENGLISH_COUNT_UNIT_RE = r"(?:pcs?|pieces?|capsules?|caps?|tablets?|tabs?|packs?|sachets?|bottles?|boxes?)"
|
||
BUNDLE_OFFER_PHRASES = (
|
||
"囤貨組",
|
||
"超值組",
|
||
"特惠組",
|
||
"優惠組",
|
||
"優惠套組",
|
||
"禮盒組",
|
||
"加大組",
|
||
"加量組",
|
||
"分享組",
|
||
"明星組",
|
||
"套組",
|
||
"組合",
|
||
"組合包",
|
||
"雙件組",
|
||
"二件組",
|
||
"2件組",
|
||
"家庭組",
|
||
"多入組",
|
||
)
|
||
NON_BRAND_BRACKET_PHRASES = (
|
||
"保濕組",
|
||
"熱銷款",
|
||
"限定",
|
||
"特惠",
|
||
"優惠",
|
||
"超值",
|
||
"囤貨",
|
||
"組合",
|
||
"套組",
|
||
"禮盒",
|
||
"分享",
|
||
"雙件",
|
||
"二件",
|
||
"2件",
|
||
"家庭",
|
||
"多入",
|
||
"任選",
|
||
"買",
|
||
"母親節",
|
||
)
|
||
CHINESE_COUNT = {
|
||
"一": 1,
|
||
"二": 2,
|
||
"兩": 2,
|
||
"雙": 2,
|
||
"三": 3,
|
||
"四": 4,
|
||
"五": 5,
|
||
"六": 6,
|
||
"七": 7,
|
||
"八": 8,
|
||
"九": 9,
|
||
"十": 10,
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ProductIdentity:
|
||
original_name: str
|
||
normalized_name: str
|
||
searchable_name: str
|
||
brand_tokens: frozenset[str]
|
||
product_type: Optional[str]
|
||
tokens: frozenset[str]
|
||
core_tokens: frozenset[str]
|
||
volumes_ml: tuple[float, ...]
|
||
weights_g: tuple[float, ...]
|
||
dosages_mg: tuple[float, ...]
|
||
counts: tuple[tuple[int, str], ...]
|
||
total_piece_count: Optional[int]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class MatchDiagnostics:
|
||
score: float
|
||
brand_score: float
|
||
token_score: float
|
||
spec_score: float
|
||
sequence_score: float
|
||
type_score: float
|
||
price_penalty: float
|
||
hard_veto: bool
|
||
reasons: tuple[str, ...]
|
||
comparison_mode: str = "exact_identity"
|
||
|
||
@property
|
||
def tags(self) -> list[str]:
|
||
tags: list[str] = ["identity_v2"]
|
||
if self.comparison_mode:
|
||
tags.append(f"comparison_{self.comparison_mode}")
|
||
if self.brand_score >= 0.95:
|
||
tags.append("brand_match")
|
||
if self.spec_score >= 0.85:
|
||
tags.append("spec_match")
|
||
if self.hard_veto:
|
||
tags.append("identity_veto")
|
||
return tags
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class UnitPriceComparison:
|
||
comparable: bool
|
||
reason: str
|
||
unit_label: str = ""
|
||
momo_total_quantity: Optional[float] = None
|
||
competitor_total_quantity: Optional[float] = None
|
||
momo_unit_price: Optional[float] = None
|
||
competitor_unit_price: Optional[float] = None
|
||
unit_gap_amount: Optional[float] = None
|
||
unit_gap_pct: Optional[float] = None
|
||
summary: str = ""
|
||
|
||
def as_dict(self) -> dict:
|
||
return {
|
||
"comparable": self.comparable,
|
||
"reason": self.reason,
|
||
"unit_label": self.unit_label,
|
||
"momo_total_quantity": self.momo_total_quantity,
|
||
"competitor_total_quantity": self.competitor_total_quantity,
|
||
"momo_unit_price": self.momo_unit_price,
|
||
"competitor_unit_price": self.competitor_unit_price,
|
||
"unit_gap_amount": self.unit_gap_amount,
|
||
"unit_gap_pct": self.unit_gap_pct,
|
||
"summary": self.summary,
|
||
}
|
||
|
||
|
||
def normalize_product_text(value: str) -> str:
|
||
text = unicodedata.normalize("NFKC", value or "")
|
||
text = "".join(
|
||
char for char in unicodedata.normalize("NFKD", text)
|
||
if not unicodedata.combining(char)
|
||
)
|
||
text = text.replace("×", "x").replace("＊", "x").replace("*", "x")
|
||
text = text.replace("／", "/").replace("＆", "&")
|
||
text = re.sub(r"[\u3000\r\n\t]+", " ", text)
|
||
text = text.lower()
|
||
text = re.sub(r"[?？]+", " ", text)
|
||
text = re.sub(r"[【】\[\]{}「」『』]", " ", text)
|
||
text = re.sub(r"[()（）]", " ", text)
|
||
text = re.sub(r"\s+", " ", text).strip()
|
||
return text
|
||
|
||
|
||
def _strip_noise(value: str) -> str:
|
||
text = value
|
||
for phrase in sorted(NOISE_PHRASES, key=len, reverse=True):
|
||
text = text.replace(phrase.lower(), " ")
|
||
text = re.sub(r"\s+", " ", text).strip()
|
||
return text
|
||
|
||
|
||
def _tokenize(value: str) -> list[str]:
|
||
raw_tokens = re.findall(r"[a-z0-9]+|[\u4e00-\u9fff]+", value)
|
||
tokens: list[str] = []
|
||
for token in raw_tokens:
|
||
if len(token) <= 1 and not token.isdigit():
|
||
continue
|
||
tokens.append(token)
|
||
return tokens
|
||
|
||
|
||
def _known_brand_tokens(text: str) -> set[str]:
|
||
tokens: set[str] = set()
|
||
try:
|
||
from services.price_comparison import BRAND_ALIASES, BRAND_NORMALIZE_MAP
|
||
except Exception:
|
||
BRAND_ALIASES = {}
|
||
BRAND_NORMALIZE_MAP = {}
|
||
|
||
alias_map = dict(BRAND_NORMALIZE_MAP)
|
||
alias_groups = {canonical: list(aliases) for canonical, aliases in BRAND_ALIASES.items()}
|
||
for canonical, aliases in BRAND_ALIAS_OVERRIDES.items():
|
||
alias_groups.setdefault(canonical, [])
|
||
alias_groups[canonical].extend(aliases)
|
||
alias_map[canonical.lower()] = canonical
|
||
for alias in aliases:
|
||
alias_map[alias.lower()] = canonical
|
||
|
||
for alias, canonical in alias_map.items():
|
||
alias_norm = normalize_product_text(alias)
|
||
if alias_norm and alias_norm in text:
|
||
tokens.add(canonical)
|
||
tokens.update(
|
||
token for token in _tokenize(alias_norm)
|
||
if not re.fullmatch(r"[a-z]{1,2}", token)
|
||
)
|
||
for related in alias_groups.get(canonical, []):
|
||
tokens.update(
|
||
token for token in _tokenize(normalize_product_text(related))
|
||
if not re.fullmatch(r"[a-z]{1,2}", token)
|
||
)
|
||
|
||
return {token for token in tokens if token and token not in GENERIC_TOKENS}
|
||
|
||
|
||
def _leading_brand_tokens(original: str, normalized: str) -> set[str]:
|
||
tokens: set[str] = set()
|
||
bracket_match = re.match(r"\s*[【\[]([^】\]]{2,40})[】\]]", original or "")
|
||
if bracket_match:
|
||
content = normalize_product_text(bracket_match.group(1))
|
||
if not any(phrase in content for phrase in NON_BRAND_BRACKET_PHRASES):
|
||
for token in _tokenize(_strip_noise(content)):
|
||
if token not in GENERIC_TOKENS:
|
||
tokens.add(token)
|
||
|
||
leading = normalized[:48]
|
||
for token in _tokenize(leading):
|
||
if re.fullmatch(r"[a-z][a-z0-9\-']{2,}", token):
|
||
tokens.add(token)
|
||
return tokens
|
||
|
||
|
||
def _extract_product_type(text: str) -> Optional[str]:
|
||
for product_type, aliases in PRODUCT_TYPES.items():
|
||
if any(alias.lower() in text for alias in aliases):
|
||
return product_type
|
||
return None
|
||
|
||
|
||
def _convert_volume(value: str, unit: str) -> Optional[tuple[str, float]]:
|
||
try:
|
||
number = float(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
unit = unit.lower()
|
||
if unit in {"ml", "毫升"}:
|
||
return ("ml", number)
|
||
if unit == "l":
|
||
return ("ml", number * 1000)
|
||
if unit in {"g", "公克"}:
|
||
return ("g", number)
|
||
if unit == "kg":
|
||
return ("g", number * 1000)
|
||
if unit in {"mg", "毫克"}:
|
||
return ("mg", number)
|
||
if unit in {"mcg", "μg", "ug", "微克"}:
|
||
return ("mg", number / 1000)
|
||
return None
|
||
|
||
|
||
def _count_unit_family(unit: str) -> str:
|
||
return COUNT_UNIT_FAMILIES.get(unit, unit)
|
||
|
||
|
||
def _extract_specs(
|
||
text: str,
|
||
) -> tuple[tuple[float, ...], tuple[float, ...], tuple[float, ...], tuple[tuple[int, str], ...], Optional[int]]:
|
||
volumes_ml: list[float] = []
|
||
weights_g: list[float] = []
|
||
dosages_mg: list[float] = []
|
||
for match in re.finditer(r"(\d+(?:\.\d+)?)\s*(ml|毫升|l|g|公克|kg|mg|毫克|mcg|μg|ug|微克)", text, re.I):
|
||
converted = _convert_volume(match.group(1), match.group(2))
|
||
if not converted:
|
||
continue
|
||
unit, number = converted
|
||
if unit == "ml":
|
||
volumes_ml.append(number)
|
||
elif unit == "g":
|
||
weights_g.append(number)
|
||
else:
|
||
dosages_mg.append(number)
|
||
|
||
counts: list[tuple[int, str]] = []
|
||
for match in re.finditer(rf"(\d+)\s*({COUNT_UNIT_PATTERN})", text):
|
||
counts.append((int(match.group(1)), match.group(2)))
|
||
for match in re.finditer(rf"([一二兩雙三四五六七八九十])\s*({COUNT_UNIT_PATTERN})", text):
|
||
counts.append((CHINESE_COUNT[match.group(1)], match.group(2)))
|
||
for match in re.finditer(rf"(?:x|乘)\s*(\d+)\s*({COUNT_UNIT_PATTERN})?", text, re.I):
|
||
unit = match.group(2) or "入"
|
||
counts.append((int(match.group(1)), unit))
|
||
for match in re.finditer(rf"(\d+)\s*{ENGLISH_COUNT_UNIT_RE}", text, re.I):
|
||
counts.append((int(match.group(1)), "入"))
|
||
buy_get = re.search(r"買\s*(\d+|[一二兩雙三四五六七八九十])\s*送\s*(\d+|[一二兩雙三四五六七八九十])", text)
|
||
if buy_get:
|
||
total_count = (_count_text_value(buy_get.group(1)) or 0) + (_count_text_value(buy_get.group(2)) or 0)
|
||
if total_count > 1:
|
||
counts.append((total_count, "入"))
|
||
if "買一送一" in text or "買1送1" in text:
|
||
counts.append((2, "入"))
|
||
|
||
total_piece_count = None
|
||
explicit_total = re.search(r"共\s*(\d+)\s*([包袋片顆粒錠枚])", text)
|
||
if explicit_total:
|
||
total_piece_count = int(explicit_total.group(1))
|
||
else:
|
||
piece_counts = [count for count, unit in counts if unit in PIECE_UNITS]
|
||
container_counts = [count for count, unit in counts if unit in CONTAINER_UNITS]
|
||
if piece_counts and container_counts:
|
||
total_piece_count = max(piece_counts) * max(container_counts)
|
||
elif piece_counts:
|
||
total_piece_count = max(piece_counts)
|
||
|
||
unique_counts = tuple(sorted(set(counts)))
|
||
return (
|
||
tuple(sorted(set(volumes_ml))),
|
||
tuple(sorted(set(weights_g))),
|
||
tuple(sorted(set(dosages_mg))),
|
||
unique_counts,
|
||
total_piece_count,
|
||
)
|
||
|
||
|
||
def parse_product_identity(name: str) -> ProductIdentity:
|
||
normalized = normalize_product_text(name)
|
||
searchable = _strip_noise(normalized)
|
||
tokens = set(_tokenize(searchable))
|
||
product_type = _extract_product_type(searchable)
|
||
known_brand_tokens = _known_brand_tokens(searchable)
|
||
brand_tokens = known_brand_tokens or _leading_brand_tokens(name, normalized)
|
||
|
||
core_tokens = {
|
||
token
|
||
for token in tokens
|
||
if token not in GENERIC_TOKENS
|
||
and not token.isdigit()
|
||
and not re.fullmatch(r"\d+(ml|g|kg|l|mg|mcg|ug)?", token)
|
||
}
|
||
core_tokens -= brand_tokens
|
||
|
||
volumes_ml, weights_g, dosages_mg, counts, total_piece_count = _extract_specs(normalized)
|
||
return ProductIdentity(
|
||
original_name=name or "",
|
||
normalized_name=normalized,
|
||
searchable_name=searchable,
|
||
brand_tokens=frozenset(brand_tokens),
|
||
product_type=product_type,
|
||
tokens=frozenset(tokens),
|
||
core_tokens=frozenset(core_tokens),
|
||
volumes_ml=volumes_ml,
|
||
weights_g=weights_g,
|
||
dosages_mg=dosages_mg,
|
||
counts=counts,
|
||
total_piece_count=total_piece_count,
|
||
)
|
||
|
||
|
||
def _weighted_token_score(left: ProductIdentity, right: ProductIdentity) -> float:
|
||
def expand_tokens(identity: ProductIdentity) -> set[str]:
|
||
tokens = set(identity.brand_tokens | identity.core_tokens)
|
||
for token in identity.core_tokens:
|
||
chinese = "".join(char for char in token if "\u4e00" <= char <= "\u9fff")
|
||
if len(chinese) >= 3:
|
||
tokens.update(f"zh:{chinese[i:i + 2]}" for i in range(len(chinese) - 1))
|
||
return tokens
|
||
|
||
left_tokens = expand_tokens(left)
|
||
right_tokens = expand_tokens(right)
|
||
if not left_tokens or not right_tokens:
|
||
return SequenceMatcher(None, left.searchable_name, right.searchable_name).ratio() * 0.6
|
||
|
||
def weight(token: str) -> float:
|
||
if token in left.brand_tokens or token in right.brand_tokens:
|
||
return 1.4
|
||
if token.startswith("zh:"):
|
||
return 0.55
|
||
if re.search(r"\d", token):
|
||
return 1.2
|
||
if len(token) >= 4:
|
||
return 1.25
|
||
return 1.0
|
||
|
||
overlap = left_tokens & right_tokens
|
||
overlap_weight = sum(weight(token) for token in overlap)
|
||
total_weight = sum(weight(token) for token in left_tokens) + sum(weight(token) for token in right_tokens)
|
||
dice = (2 * overlap_weight / total_weight) if total_weight else 0
|
||
sequence = SequenceMatcher(None, " ".join(sorted(left_tokens)), " ".join(sorted(right_tokens))).ratio()
|
||
return min(1.0, dice * 0.72 + sequence * 0.28)
|
||
|
||
|
||
def _brand_score(left: ProductIdentity, right: ProductIdentity) -> tuple[float, bool, str | None]:
|
||
if not left.brand_tokens or not right.brand_tokens:
|
||
return 0.55, False, None
|
||
if left.brand_tokens & right.brand_tokens:
|
||
return 1.0, False, None
|
||
return 0.0, True, "brand_conflict"
|
||
|
||
|
||
def _close_number(left: float, right: float, tolerance: float = 0.04) -> bool:
|
||
denominator = max(abs(left), abs(right), 1.0)
|
||
return abs(left - right) / denominator <= tolerance
|
||
|
||
|
||
def _spec_component(left_values: Iterable[float], right_values: Iterable[float]) -> tuple[float, bool]:
|
||
left_tuple = tuple(sorted(set(left_values)))
|
||
right_tuple = tuple(sorted(set(right_values)))
|
||
if not left_tuple and not right_tuple:
|
||
return 0.55, False
|
||
if not left_tuple or not right_tuple:
|
||
return 0.45, False
|
||
if len(left_tuple) > 1 or len(right_tuple) > 1:
|
||
if len(left_tuple) != len(right_tuple):
|
||
return 0.0, True
|
||
unmatched = list(right_tuple)
|
||
for left_value in left_tuple:
|
||
match_index = next(
|
||
(
|
||
index
|
||
for index, right_value in enumerate(unmatched)
|
||
if _close_number(left_value, right_value)
|
||
),
|
||
None,
|
||
)
|
||
if match_index is None:
|
||
return 0.0, True
|
||
unmatched.pop(match_index)
|
||
return 1.0, False
|
||
for left_value in left_tuple:
|
||
if any(_close_number(left_value, right_value) for right_value in right_tuple):
|
||
return 1.0, False
|
||
return 0.0, True
|
||
|
||
|
||
def _has_hard_count_unit_conflict(left: ProductIdentity, right: ProductIdentity) -> bool:
|
||
if not left.counts or not right.counts:
|
||
return False
|
||
left_by_count: dict[int, set[str]] = {}
|
||
right_by_count: dict[int, set[str]] = {}
|
||
for count, unit in left.counts:
|
||
left_by_count.setdefault(count, set()).add(_count_unit_family(unit))
|
||
for count, unit in right.counts:
|
||
right_by_count.setdefault(count, set()).add(_count_unit_family(unit))
|
||
|
||
for count in set(left_by_count) & set(right_by_count):
|
||
left_units = left_by_count[count]
|
||
right_units = right_by_count[count]
|
||
if left_units & right_units:
|
||
continue
|
||
if (
|
||
(left_units & PIECE_UNITS and right_units & CONTAINER_UNITS)
|
||
or (right_units & PIECE_UNITS and left_units & CONTAINER_UNITS)
|
||
):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _count_score(left: ProductIdentity, right: ProductIdentity) -> tuple[float, bool]:
|
||
left_counts = [count for count, _unit in left.counts]
|
||
right_counts = [count for count, _unit in right.counts]
|
||
left_by_unit: dict[str, set[int]] = {}
|
||
right_by_unit: dict[str, set[int]] = {}
|
||
for count, unit in left.counts:
|
||
left_by_unit.setdefault(_count_unit_family(unit), set()).add(count)
|
||
for count, unit in right.counts:
|
||
right_by_unit.setdefault(_count_unit_family(unit), set()).add(count)
|
||
|
||
if left.total_piece_count and right.total_piece_count:
|
||
if left.total_piece_count == right.total_piece_count:
|
||
return 1.0, False
|
||
ratio = max(left.total_piece_count, right.total_piece_count) / max(min(left.total_piece_count, right.total_piece_count), 1)
|
||
return (0.0, True) if ratio >= 1.5 else (0.45, False)
|
||
|
||
for unit in set(left_by_unit) & set(right_by_unit):
|
||
if left_by_unit[unit] != right_by_unit[unit]:
|
||
return 0.0, True
|
||
if left.counts and right.counts:
|
||
if set(left.counts) & set(right.counts):
|
||
return 0.85, False
|
||
if _has_hard_count_unit_conflict(left, right):
|
||
return 0.0, True
|
||
if left_counts and right_counts:
|
||
ratio = max(max(left_counts), max(right_counts)) / max(min(max(left_counts), max(right_counts)), 1)
|
||
if ratio >= 1.5:
|
||
return 0.0, True
|
||
return 0.35, False
|
||
if (left_counts and max(left_counts) > 1) or (right_counts and max(right_counts) > 1):
|
||
return 0.0, True
|
||
return 0.5, False
|
||
|
||
|
||
def _spec_score(left: ProductIdentity, right: ProductIdentity) -> tuple[float, bool, tuple[str, ...]]:
|
||
volume_score, volume_conflict = _spec_component(left.volumes_ml, right.volumes_ml)
|
||
weight_score, weight_conflict = _spec_component(left.weights_g, right.weights_g)
|
||
dosage_score, dosage_conflict = _spec_component(left.dosages_mg, right.dosages_mg)
|
||
count_score, count_conflict = _count_score(left, right)
|
||
|
||
available = []
|
||
if left.volumes_ml or right.volumes_ml:
|
||
available.append(volume_score)
|
||
if left.weights_g or right.weights_g:
|
||
available.append(weight_score)
|
||
if left.dosages_mg or right.dosages_mg:
|
||
available.append(dosage_score)
|
||
if left.counts or right.counts:
|
||
available.append(count_score)
|
||
if not available:
|
||
return 0.55, False, ()
|
||
|
||
score = sum(available) / len(available)
|
||
conflicts = []
|
||
if volume_conflict:
|
||
conflicts.append("volume_conflict")
|
||
if weight_conflict:
|
||
conflicts.append("weight_conflict")
|
||
if dosage_conflict:
|
||
conflicts.append("dosage_conflict")
|
||
if count_conflict:
|
||
conflicts.append("count_conflict")
|
||
return score, bool(conflicts), tuple(conflicts)
|
||
|
||
|
||
def _has_bundle_offer(identity: ProductIdentity) -> bool:
|
||
text = identity.normalized_name
|
||
return bool(
|
||
re.search(r"買\s*\d+\s*送\s*\d+", text)
|
||
or re.search(r"買\s*[一二兩雙三四五六七八九十]\s*送\s*[一二兩雙三四五六七八九十]", text)
|
||
or "買一送一" in text
|
||
or any(phrase in text for phrase in BUNDLE_OFFER_PHRASES)
|
||
)
|
||
|
||
|
||
def _has_multi_component(identity: ProductIdentity) -> bool:
|
||
text = identity.normalized_name
|
||
return bool(
|
||
"+" in text
|
||
or "＋" in text
|
||
or re.search(r"\d+\s*(?:ml|g|mg|毫升|公克|毫克)\s*x\s*\d+", text, re.I)
|
||
)
|
||
|
||
|
||
def _has_refill_pack(identity: ProductIdentity) -> bool:
|
||
text = identity.normalized_name
|
||
return bool(
|
||
"補充瓶" in text
|
||
or "補充包" in text
|
||
or "替換蕊" in text
|
||
or "替換芯" in text
|
||
or "refill" in text
|
||
)
|
||
|
||
|
||
def _spec_mention_count(identity: ProductIdentity) -> int:
|
||
return len(
|
||
re.findall(
|
||
r"\d+(?:\.\d+)?\s*(?:ml|毫升|l|g|公克|kg|mg|毫克|mcg|μg|ug|微克)",
|
||
identity.normalized_name,
|
||
re.I,
|
||
)
|
||
)
|
||
|
||
|
||
def _count_text_value(value: str) -> Optional[int]:
|
||
if value.isdigit():
|
||
return int(value)
|
||
return CHINESE_COUNT.get(value)
|
||
|
||
|
||
def _pack_multiplier(identity: ProductIdentity) -> int:
|
||
text = identity.normalized_name
|
||
buy_get = re.search(r"買\s*(\d+|[一二兩雙三四五六七八九十])\s*送\s*(\d+|[一二兩雙三四五六七八九十])", text)
|
||
if buy_get:
|
||
left = _count_text_value(buy_get.group(1)) or 0
|
||
right = _count_text_value(buy_get.group(2)) or 0
|
||
if left + right > 1:
|
||
return left + right
|
||
if "買一送一" in text or "買1送1" in text:
|
||
return 2
|
||
piece_pack = re.search(r"(\d+|[一二兩雙三四五六七八九十])\s*件\s*組", text)
|
||
if piece_pack:
|
||
count = _count_text_value(piece_pack.group(1)) or 0
|
||
if count > 1:
|
||
return count
|
||
|
||
multipliers = [count for count, unit in identity.counts if unit in COUNT_UNITS and count > 1]
|
||
if multipliers:
|
||
return max(multipliers)
|
||
return 1
|
||
|
||
|
||
def _has_overlapping_base_spec(left: ProductIdentity, right: ProductIdentity) -> bool:
|
||
left_volumes = tuple(sorted(set(left.volumes_ml)))
|
||
right_volumes = tuple(sorted(set(right.volumes_ml)))
|
||
if left_volumes or right_volumes:
|
||
if not left_volumes or not right_volumes:
|
||
return False
|
||
if len(left_volumes) > 1 or len(right_volumes) > 1:
|
||
return False
|
||
return _close_number(left_volumes[0], right_volumes[0])
|
||
|
||
left_weights = tuple(sorted(set(left.weights_g)))
|
||
right_weights = tuple(sorted(set(right.weights_g)))
|
||
if left_weights or right_weights:
|
||
if not left_weights or not right_weights:
|
||
return False
|
||
if len(left_weights) > 1 or len(right_weights) > 1:
|
||
return False
|
||
return _close_number(left_weights[0], right_weights[0])
|
||
|
||
return False
|
||
|
||
|
||
def _single_unit_total(identity: ProductIdentity) -> tuple[Optional[str], Optional[float], str]:
|
||
volumes = tuple(sorted(set(identity.volumes_ml)))
|
||
weights = tuple(sorted(set(identity.weights_g)))
|
||
if volumes and weights:
|
||
return None, None, "mixed_volume_weight"
|
||
if len(volumes) > 1 or len(weights) > 1:
|
||
return None, None, "multi_spec_component"
|
||
if volumes:
|
||
return "ml", volumes[0] * _pack_multiplier(identity), "ok"
|
||
if weights:
|
||
multiplier = identity.total_piece_count or _pack_multiplier(identity)
|
||
return "g", weights[0] * multiplier, "ok"
|
||
if identity.total_piece_count:
|
||
return "入", float(identity.total_piece_count), "ok"
|
||
return None, None, "missing_single_unit"
|
||
|
||
|
||
def build_unit_price_comparison(
|
||
momo_name: str,
|
||
competitor_name: str,
|
||
momo_price: Optional[float],
|
||
competitor_price: Optional[float],
|
||
) -> dict:
|
||
"""Build deterministic unit-price evidence for unit-comparable candidates."""
|
||
diagnostics = score_marketplace_match(
|
||
momo_name,
|
||
competitor_name,
|
||
momo_price=momo_price,
|
||
competitor_price=competitor_price,
|
||
)
|
||
if diagnostics.comparison_mode != "unit_comparable":
|
||
return UnitPriceComparison(False, diagnostics.comparison_mode).as_dict()
|
||
|
||
left = parse_product_identity(momo_name)
|
||
right = parse_product_identity(competitor_name)
|
||
left_unit, left_total, left_reason = _single_unit_total(left)
|
||
right_unit, right_total, right_reason = _single_unit_total(right)
|
||
if left_reason != "ok" or right_reason != "ok":
|
||
return UnitPriceComparison(False, f"{left_reason}:{right_reason}").as_dict()
|
||
if left_unit != right_unit or not left_total or not right_total:
|
||
return UnitPriceComparison(False, "unit_mismatch").as_dict()
|
||
|
||
try:
|
||
momo_price_num = float(momo_price or 0)
|
||
competitor_price_num = float(competitor_price or 0)
|
||
except (TypeError, ValueError):
|
||
return UnitPriceComparison(False, "invalid_price").as_dict()
|
||
if momo_price_num <= 0 or competitor_price_num <= 0:
|
||
return UnitPriceComparison(False, "invalid_price").as_dict()
|
||
|
||
momo_unit_price = momo_price_num / left_total
|
||
competitor_unit_price = competitor_price_num / right_total
|
||
unit_gap_amount = momo_unit_price - competitor_unit_price
|
||
unit_gap_pct = unit_gap_amount / competitor_unit_price * 100 if competitor_unit_price else 0
|
||
summary = (
|
||
f"MOMO ${momo_unit_price:.2f}/{left_unit} vs "
|
||
f"PChome ${competitor_unit_price:.2f}/{left_unit} "
|
||
f"({unit_gap_pct:+.1f}%)"
|
||
)
|
||
return UnitPriceComparison(
|
||
comparable=True,
|
||
reason="unit_comparable",
|
||
unit_label=left_unit,
|
||
momo_total_quantity=round(left_total, 3),
|
||
competitor_total_quantity=round(right_total, 3),
|
||
momo_unit_price=round(momo_unit_price, 4),
|
||
competitor_unit_price=round(competitor_unit_price, 4),
|
||
unit_gap_amount=round(unit_gap_amount, 4),
|
||
unit_gap_pct=round(unit_gap_pct, 2),
|
||
summary=summary,
|
||
).as_dict()
|
||
|
||
|
||
def _is_unit_comparable_candidate(
|
||
left: ProductIdentity,
|
||
right: ProductIdentity,
|
||
token_score: float,
|
||
chinese_name_score: float,
|
||
brand_conflict: bool,
|
||
type_score: float,
|
||
reasons: Iterable[str],
|
||
) -> bool:
|
||
"""Identify same core product sold in different packs.
|
||
|
||
These are not safe exact matches. They can only enter a normalized unit-price
|
||
review lane, otherwise a bundle price may be incorrectly compared with a
|
||
single-item price.
|
||
"""
|
||
reason_set = set(reasons)
|
||
pack_difference = bool(reason_set & {
|
||
"bundle_offer_conflict",
|
||
"multi_component_conflict",
|
||
"count_conflict",
|
||
"component_count_conflict",
|
||
})
|
||
if not pack_difference:
|
||
return False
|
||
if brand_conflict or "brand_conflict" in reason_set:
|
||
return False
|
||
if "refill_pack_conflict" in reason_set:
|
||
return False
|
||
if type_score == 0.0 or "type_conflict" in reason_set:
|
||
return False
|
||
if not _has_overlapping_base_spec(left, right):
|
||
return False
|
||
if token_score < 0.45 and chinese_name_score < 0.28:
|
||
return False
|
||
if "product_line_conflict" in reason_set and token_score < 0.72:
|
||
return False
|
||
return True
|
||
|
||
|
||
def _chinese_bigram_score(left: ProductIdentity, right: ProductIdentity) -> float:
|
||
def signature(identity: ProductIdentity) -> set[str]:
|
||
text = identity.searchable_name
|
||
for token in sorted(identity.brand_tokens, key=len, reverse=True):
|
||
text = text.replace(token, " ")
|
||
text = re.sub(r"[a-z0-9]+", " ", text)
|
||
text = "".join(char for char in text if "\u4e00" <= char <= "\u9fff")
|
||
for phrase in (
|
||
"官方", "直營", "公司貨", "專櫃", "正貨", "原廠", "限定", "獨家",
|
||
"期間", "超值", "特惠", "優惠", "新品", "經典", "人氣", "熱銷",
|
||
"必買", "推薦", "任選", "禮盒", "母親節", "超品日", "多款",
|
||
"組", "入", "瓶", "盒", "包", "片", "支", "條",
|
||
):
|
||
text = text.replace(phrase, "")
|
||
return {text[i:i + 2] for i in range(max(0, len(text) - 1))}
|
||
|
||
left_signature = signature(left)
|
||
right_signature = signature(right)
|
||
if not left_signature or not right_signature:
|
||
return 0.55
|
||
return 2 * len(left_signature & right_signature) / (len(left_signature) + len(right_signature))
|
||
|
||
|
||
def _has_strong_product_line_signal(
|
||
left: ProductIdentity,
|
||
right: ProductIdentity,
|
||
token_score: float,
|
||
chinese_name_score: float,
|
||
) -> bool:
|
||
shared_core = left.core_tokens & right.core_tokens
|
||
shared_latin_or_model = {
|
||
token for token in shared_core
|
||
if re.fullmatch(r"[a-z][a-z0-9-]{3,}", token)
|
||
or re.fullmatch(r"[a-z]{2,}-?\d+[a-z0-9-]*", token)
|
||
}
|
||
if shared_latin_or_model and token_score >= 0.50:
|
||
return True
|
||
return token_score >= 0.56 and chinese_name_score >= 0.45
|
||
|
||
|
||
def _has_safe_exact_spec_signal(
|
||
left: ProductIdentity,
|
||
right: ProductIdentity,
|
||
token_score: float,
|
||
sequence_score: float,
|
||
type_score: float,
|
||
) -> bool:
|
||
if type_score < 0.55:
|
||
return False
|
||
if _spec_mention_count(left) > 1 or _spec_mention_count(right) > 1:
|
||
return False
|
||
if not _has_overlapping_base_spec(left, right):
|
||
return False
|
||
return token_score >= 0.42 or sequence_score >= 0.50
|
||
|
||
|
||
def _model_line_tokens(identity: ProductIdentity) -> set[str]:
|
||
tokens: set[str] = set()
|
||
for token in identity.core_tokens:
|
||
if token in GENERIC_TOKENS:
|
||
continue
|
||
if re.fullmatch(r"[a-z][a-z0-9-]{2,}", token):
|
||
tokens.add(token)
|
||
for match in re.finditer(r"([\u4e00-\u9fff]{2,})(?:系列)", token):
|
||
value = match.group(1)
|
||
if value not in GENERIC_TOKENS:
|
||
tokens.add(value)
|
||
return tokens
|
||
|
||
|
||
def _has_model_line_conflict(left: ProductIdentity, right: ProductIdentity) -> bool:
|
||
left_tokens = _model_line_tokens(left)
|
||
right_tokens = _model_line_tokens(right)
|
||
if not left_tokens or not right_tokens:
|
||
return False
|
||
return not bool(left_tokens & right_tokens)
|
||
|
||
|
||
def score_marketplace_match(
|
||
momo_name: str,
|
||
competitor_name: str,
|
||
momo_price: Optional[float] = None,
|
||
competitor_price: Optional[float] = None,
|
||
) -> MatchDiagnostics:
|
||
left = parse_product_identity(momo_name)
|
||
right = parse_product_identity(competitor_name)
|
||
|
||
brand_score, brand_conflict, brand_reason = _brand_score(left, right)
|
||
token_score = _weighted_token_score(left, right)
|
||
spec_score, spec_conflict, spec_reasons = _spec_score(left, right)
|
||
sequence_score = SequenceMatcher(None, left.searchable_name, right.searchable_name).ratio()
|
||
chinese_name_score = _chinese_bigram_score(left, right)
|
||
if left.product_type and right.product_type:
|
||
type_score = 1.0 if left.product_type == right.product_type else 0.0
|
||
else:
|
||
type_score = 0.55
|
||
|
||
reasons = []
|
||
if brand_reason:
|
||
reasons.append(brand_reason)
|
||
reasons.extend(spec_reasons)
|
||
if left.product_type and right.product_type and left.product_type != right.product_type:
|
||
reasons.append("type_conflict")
|
||
model_line_conflict = _has_model_line_conflict(left, right)
|
||
if model_line_conflict:
|
||
reasons.append("model_line_conflict")
|
||
bundle_offer_conflict = (
|
||
_has_bundle_offer(left) != _has_bundle_offer(right)
|
||
and not (
|
||
left.total_piece_count
|
||
and right.total_piece_count
|
||
and left.total_piece_count == right.total_piece_count
|
||
)
|
||
)
|
||
if bundle_offer_conflict:
|
||
reasons.append("bundle_offer_conflict")
|
||
if _has_multi_component(left) != _has_multi_component(right):
|
||
reasons.append("multi_component_conflict")
|
||
if _has_refill_pack(left) != _has_refill_pack(right):
|
||
reasons.append("refill_pack_conflict")
|
||
left_spec_mentions = _spec_mention_count(left)
|
||
right_spec_mentions = _spec_mention_count(right)
|
||
if left_spec_mentions and right_spec_mentions and left_spec_mentions != right_spec_mentions:
|
||
reasons.append("component_count_conflict")
|
||
if chinese_name_score < 0.16:
|
||
reasons.append("product_line_conflict")
|
||
|
||
hard_veto = brand_conflict or spec_conflict
|
||
if bundle_offer_conflict:
|
||
hard_veto = True
|
||
if _has_multi_component(left) != _has_multi_component(right):
|
||
hard_veto = True
|
||
if _has_refill_pack(left) != _has_refill_pack(right):
|
||
hard_veto = True
|
||
if model_line_conflict:
|
||
hard_veto = True
|
||
if left_spec_mentions and right_spec_mentions and left_spec_mentions != right_spec_mentions:
|
||
hard_veto = True
|
||
if chinese_name_score < 0.16 and token_score < 0.72:
|
||
hard_veto = True
|
||
if left.product_type and right.product_type and left.product_type != right.product_type:
|
||
hard_veto = True
|
||
|
||
comparison_mode = "exact_identity"
|
||
if _is_unit_comparable_candidate(
|
||
left,
|
||
right,
|
||
token_score,
|
||
chinese_name_score,
|
||
brand_conflict,
|
||
type_score,
|
||
reasons,
|
||
):
|
||
comparison_mode = "unit_comparable"
|
||
reasons.append("unit_comparable")
|
||
elif hard_veto:
|
||
comparison_mode = "not_comparable"
|
||
|
||
price_penalty = 0.0
|
||
try:
|
||
if momo_price and competitor_price:
|
||
ratio = float(competitor_price) / max(float(momo_price), 1.0)
|
||
if (ratio < 0.3 or ratio > 3.2) and token_score < 0.78:
|
||
price_penalty = 0.12
|
||
reasons.append("price_ratio_extreme")
|
||
elif (ratio < 0.48 or ratio > 2.2) and token_score < 0.68:
|
||
price_penalty = 0.06
|
||
reasons.append("price_ratio_wide")
|
||
except (TypeError, ValueError, ZeroDivisionError):
|
||
price_penalty = 0.0
|
||
|
||
score = (
|
||
brand_score * 0.20
|
||
+ token_score * 0.36
|
||
+ spec_score * 0.25
|
||
+ sequence_score * 0.12
|
||
+ type_score * 0.07
|
||
- price_penalty
|
||
)
|
||
|
||
if token_score >= 0.72 and spec_score >= 0.82 and not brand_conflict:
|
||
score += 0.08
|
||
if (
|
||
brand_score >= 0.95
|
||
and not hard_veto
|
||
and price_penalty == 0
|
||
and type_score >= 0.55
|
||
and spec_score >= 0.55
|
||
and _has_strong_product_line_signal(left, right, token_score, chinese_name_score)
|
||
):
|
||
score += 0.07
|
||
reasons.append("strong_product_line_match")
|
||
if (
|
||
brand_score >= 0.95
|
||
and not hard_veto
|
||
and price_penalty == 0
|
||
and _has_safe_exact_spec_signal(left, right, token_score, sequence_score, type_score)
|
||
):
|
||
score += 0.025
|
||
reasons.append("strong_exact_spec_match")
|
||
if (
|
||
brand_score >= 0.95
|
||
and not hard_veto
|
||
and not reasons
|
||
and price_penalty == 0
|
||
and type_score >= 0.95
|
||
and token_score >= 0.82
|
||
and spec_score >= 0.40
|
||
and chinese_name_score >= 0.65
|
||
):
|
||
score += 0.04
|
||
reasons.append("strong_component_line_match")
|
||
if hard_veto:
|
||
score = min(score, 0.74 if comparison_mode == "unit_comparable" else 0.32)
|
||
score = max(0.0, min(1.0, score))
|
||
|
||
return MatchDiagnostics(
|
||
score=round(score, 3),
|
||
brand_score=round(brand_score, 3),
|
||
token_score=round(token_score, 3),
|
||
spec_score=round(spec_score, 3),
|
||
sequence_score=round(sequence_score, 3),
|
||
type_score=round(type_score, 3),
|
||
price_penalty=round(price_penalty, 3),
|
||
hard_veto=hard_veto,
|
||
reasons=tuple(reasons),
|
||
comparison_mode=comparison_mode,
|
||
)
|
||
|
||
|
||
def _clean_search_phrase(value: str) -> str:
|
||
text = normalize_product_text(value)
|
||
for phrase in sorted(SEARCH_NOISE_PHRASES, key=len, reverse=True):
|
||
text = text.replace(phrase.lower(), " ")
|
||
text = re.sub(r"[^\w\u4e00-\u9fff]+", " ", text)
|
||
text = " ".join(
|
||
token for token in text.split()
|
||
if token not in SEARCH_NOISE_TOKENS and token not in GENERIC_TOKENS
|
||
)
|
||
text = re.sub(r"\s+", " ", text).strip()
|
||
return text
|
||
|
||
|
||
def _search_spec_terms(identity: ProductIdentity) -> list[str]:
|
||
specs: list[str] = []
|
||
if identity.volumes_ml:
|
||
volume = identity.volumes_ml[0]
|
||
specs.append(f"{volume:g}ml")
|
||
if identity.weights_g:
|
||
weight = identity.weights_g[0]
|
||
specs.append(f"{weight:g}g")
|
||
if identity.dosages_mg:
|
||
dosage = identity.dosages_mg[0]
|
||
specs.append(f"{dosage:g}mg")
|
||
if identity.total_piece_count:
|
||
specs.append(f"{identity.total_piece_count}包")
|
||
return specs
|
||
|
||
|
||
def _extract_anchor_phrases(token: str) -> list[str]:
|
||
cleaned = _clean_search_phrase(token)
|
||
if not cleaned:
|
||
return []
|
||
|
||
phrases: list[str] = []
|
||
for anchor in SEARCH_IDENTITY_ANCHORS:
|
||
if anchor not in cleaned:
|
||
continue
|
||
if re.search(r"[\u4e00-\u9fff]", anchor):
|
||
prefix_width = 0 if len(anchor) >= 5 else (4 if len(anchor) >= 3 else 6)
|
||
match = re.search(rf"([\u4e00-\u9fff]{{0,{prefix_width}}}{re.escape(anchor)})", cleaned)
|
||
phrase = match.group(1) if match else anchor
|
||
else:
|
||
phrase = anchor
|
||
phrase = _clean_search_phrase(phrase)
|
||
if phrase.startswith("款") and len(phrase) > 2:
|
||
phrase = phrase[1:]
|
||
if any(existing in phrase and existing != phrase for existing in phrases):
|
||
continue
|
||
if len(phrase) >= 2 and phrase not in phrases:
|
||
phrases.append(phrase)
|
||
return phrases
|
||
|
||
|
||
def _search_core_score(token: str, all_tokens: set[str]) -> tuple[int, int, str]:
|
||
cleaned = _clean_search_phrase(token)
|
||
if not cleaned:
|
||
return (-999, 0, cleaned)
|
||
compact = cleaned.replace(" ", "")
|
||
if compact in SEARCH_NOISE_TOKENS or compact in GENERIC_TOKENS:
|
||
return (-900, 0, cleaned)
|
||
if re.fullmatch(r"\d+(?:\.\d+)?(?:ml|g|mg|kg|l)x\d+", compact, re.I):
|
||
return (-900, 0, cleaned)
|
||
|
||
score = 0
|
||
if re.search(r"[a-z][a-z0-9-]{2,}", cleaned):
|
||
score += 30
|
||
if re.search(r"\d", cleaned):
|
||
score += 12
|
||
|
||
anchors = _extract_anchor_phrases(cleaned)
|
||
if anchors:
|
||
score += 90
|
||
score += min(24, len(anchors[0]) * 3)
|
||
if anchors[0] == compact:
|
||
score += 8
|
||
if compact in SEARCH_BROAD_ANCHORS:
|
||
score -= 28
|
||
else:
|
||
score += max(0, 24 - len(compact))
|
||
|
||
if len(compact) <= 8:
|
||
score += 14
|
||
elif len(compact) >= 12:
|
||
score -= 12
|
||
|
||
has_better_anchor = any(
|
||
other != token and _extract_anchor_phrases(other)
|
||
for other in all_tokens
|
||
)
|
||
if has_better_anchor and any(term in compact for term in SEARCH_AMBIGUOUS_PRODUCT_TERMS):
|
||
score -= 80
|
||
if any(noise in compact for noise in SEARCH_NOISE_TOKENS):
|
||
score -= 18
|
||
|
||
return (score, -len(compact), cleaned)
|
||
|
||
|
||
def _ranked_search_core_phrases(identity: ProductIdentity, limit: int = 4) -> list[str]:
|
||
tokens = {token for token in identity.core_tokens if token not in GENERIC_TOKENS}
|
||
ranked_tokens = sorted(
|
||
tokens,
|
||
key=lambda token: _search_core_score(token, tokens),
|
||
reverse=True,
|
||
)
|
||
|
||
phrases: list[str] = []
|
||
for token in ranked_tokens:
|
||
if _search_core_score(token, tokens)[0] < -100:
|
||
continue
|
||
candidates = _extract_anchor_phrases(token) or [_clean_search_phrase(token)]
|
||
for phrase in candidates:
|
||
compact = phrase.replace(" ", "")
|
||
if len(compact) < 2 or compact in SEARCH_NOISE_TOKENS:
|
||
continue
|
||
if any(term in compact for term in SEARCH_AMBIGUOUS_PRODUCT_TERMS) and len(phrases) > 0:
|
||
continue
|
||
if phrase not in phrases:
|
||
phrases.append(phrase)
|
||
if len(phrases) >= limit:
|
||
return phrases
|
||
return phrases
|
||
|
||
|
||
def build_search_terms(name: str, max_terms: int = 3) -> list[str]:
|
||
identity = parse_product_identity(name)
|
||
terms: list[str] = []
|
||
|
||
def primary_brand_phrase() -> str:
|
||
chinese = sorted(
|
||
(token for token in identity.brand_tokens if re.search(r"[\u4e00-\u9fff]", token)),
|
||
key=lambda token: (-len(token), token),
|
||
)
|
||
if chinese:
|
||
return chinese[0]
|
||
latin = sorted(
|
||
(
|
||
token for token in identity.brand_tokens
|
||
if re.search(r"[a-z]", token) and len(token) >= 3 and token not in GENERIC_TOKENS
|
||
),
|
||
key=lambda token: (" " not in token and "-" not in token, -len(token), token),
|
||
)
|
||
return latin[0] if latin else ""
|
||
|
||
brand_part = primary_brand_phrase()
|
||
spec_part = " ".join(_search_spec_terms(identity))
|
||
core_phrases = _ranked_search_core_phrases(identity, limit=4)
|
||
core_short = " ".join(core_phrases[:2])
|
||
core_primary = core_phrases[0] if core_phrases else ""
|
||
model_phrases = [
|
||
phrase
|
||
for phrase in core_phrases[1:]
|
||
if re.fullmatch(r"[a-z]*\d+[a-z0-9-]*", phrase)
|
||
or re.fullmatch(r"[a-z][a-z0-9-]{2,}", phrase)
|
||
]
|
||
primary_with_model = " ".join(
|
||
part for part in (core_primary, model_phrases[0] if model_phrases else "") if part
|
||
)
|
||
for value in (
|
||
" ".join(part for part in (brand_part, primary_with_model, spec_part) if part),
|
||
" ".join(part for part in (brand_part, core_short, spec_part) if part),
|
||
" ".join(part for part in (brand_part, core_short) if part),
|
||
" ".join(part for part in (core_primary, spec_part) if part),
|
||
identity.searchable_name,
|
||
):
|
||
cleaned = _clean_search_phrase(value)
|
||
if cleaned and cleaned not in terms:
|
||
terms.append(cleaned[:42])
|
||
if len(terms) >= max_terms:
|
||
break
|
||
|
||
return terms
|