Files
ewoooc/services/marketplace_product_matcher.py
OoO 1911c1c53d
All checks were successful
CD Pipeline / deploy (push) Successful in 1m40s
補強套組與多品項比對否決
2026-05-19 18:07:26 +08:00

615 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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",
"la",
"paris",
}
PRODUCT_TYPES = {
"精華": ("精華", "精華液", "essence", "serum", "安瓶"),
"化妝水": ("化妝水", "機能水", "toner", "lotion"),
"乳液": ("乳液", "emulsion", "milk"),
"面霜": ("面霜", "乳霜", "", "cream"),
"防曬": ("防曬", "spf", "uv", "sunscreen"),
"洗面乳": ("洗面乳", "洗顏", "潔面", "cleanser", "foam"),
"面膜": ("面膜", "mask"),
"眼霜": ("眼霜", "眼部", "眼膜", "eye"),
"卸妝": ("卸妝", "cleansing", "remover"),
"粉底": ("粉底", "粉霜", "粉凝露", "foundation"),
"蜜粉": ("蜜粉", "powder"),
"精油": ("精油", "香氛", "擴香"),
"保健": ("", "膠囊", "", "", "", "健康食品"),
}
COUNT_UNITS = {"", "", "", "", "", "", "", "", "", "", ""}
PIECE_UNITS = {"", "", "", "", ""}
CONTAINER_UNITS = {"", "", ""}
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, ...]
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, ...]
@property
def tags(self) -> list[str]:
tags: list[str] = ["identity_v2"]
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
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 = {}
for alias, canonical in BRAND_NORMALIZE_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 BRAND_ALIASES.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))
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)
return None
def _extract_specs(text: str) -> tuple[tuple[float, ...], tuple[float, ...], tuple[tuple[int, str], ...], Optional[int]]:
volumes_ml: list[float] = []
weights_g: list[float] = []
for match in re.finditer(r"(\d+(?:\.\d+)?)\s*(ml|毫升|l|g|公克|kg)", 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)
else:
weights_g.append(number)
counts: list[tuple[int, str]] = []
for match in re.finditer(r"(\d+)\s*([入組瓶支條盒包片顆錠枚])", text):
counts.append((int(match.group(1)), match.group(2)))
for match in re.finditer(r"([一二兩雙三四五六七八九十])\s*([入組瓶支條盒包片顆錠枚])", text):
counts.append((CHINESE_COUNT[match.group(1)], match.group(2)))
for match in re.finditer(r"(?:x|乘)\s*(\d+)\s*([入組瓶支條盒包片顆錠枚])?", text, re.I):
unit = match.group(2) or ""
counts.append((int(match.group(1)), unit))
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))),
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)
brand_tokens = _known_brand_tokens(searchable) | _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)?", token)
}
core_tokens -= brand_tokens
volumes_ml, weights_g, counts, total_piece_count = _extract_specs(searchable)
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,
counts=counts,
total_piece_count=total_piece_count,
)
def _weighted_token_score(left: ProductIdentity, right: ProductIdentity) -> float:
left_tokens = left.brand_tokens | left.core_tokens
right_tokens = right.brand_tokens | right.core_tokens
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 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(left_values)
right_tuple = tuple(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
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 _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]
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)
if left.counts and right.counts:
if set(left.counts) & set(right.counts):
return 0.85, False
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)
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.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 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 "囤貨組" in text
or "超值組" in text
or "特惠組" in text
or "優惠套組" in text
or "禮盒組" in text
or "加大組" in text
or "套組" in text
)
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|毫升|公克)\s*x\s*\d+", text, re.I)
)
def _spec_mention_count(identity: ProductIdentity) -> int:
return len(re.findall(r"\d+(?:\.\d+)?\s*(?:ml|毫升|l|g|公克|kg)", identity.normalized_name, re.I))
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 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")
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")
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 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 and token_score < 0.55:
hard_veto = True
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 hard_veto:
score = min(score, 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),
)
def build_search_terms(name: str, max_terms: int = 3) -> list[str]:
identity = parse_product_identity(name)
terms: list[str] = []
brand_part = " ".join(sorted(identity.brand_tokens))[:24]
core = " ".join(sorted(identity.core_tokens, key=lambda token: (-len(token), token))[:4])
specs = []
if identity.volumes_ml:
specs.append(f"{int(identity.volumes_ml[0])}ml")
if identity.weights_g:
specs.append(f"{int(identity.weights_g[0])}g")
if identity.total_piece_count:
specs.append(f"{identity.total_piece_count}")
for value in (
" ".join(part for part in (brand_part, core, " ".join(specs)) if part),
" ".join(part for part in (brand_part, core) if part),
identity.searchable_name,
):
cleaned = re.sub(r"[^\w\u4e00-\u9fff]+", " ", value)
cleaned = re.sub(r"\s+", " ", cleaned).strip()
if cleaned and cleaned not in terms:
terms.append(cleaned[:42])
if len(terms) >= max_terms:
break
return terms