74fdf9bd
 tangwang
1.
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
|
"""
Product title exclusion detection for query understanding.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Set, Tuple
from .tokenization import TokenizedText, normalize_query_text, tokenize_text
def _dedupe_terms(terms: Iterable[str]) -> List[str]:
result: List[str] = []
seen: Set[str] = set()
for raw_term in terms:
term = normalize_query_text(raw_term)
if not term or term in seen:
continue
seen.add(term)
result.append(term)
return result
@dataclass(frozen=True)
class ProductTitleExclusionRule:
zh_trigger_terms: Tuple[str, ...]
en_trigger_terms: Tuple[str, ...]
zh_title_exclusions: Tuple[str, ...]
en_title_exclusions: Tuple[str, ...]
max_term_ngram: int = 3
@classmethod
def from_config_row(cls, row: Dict[str, Sequence[str]]) -> Optional["ProductTitleExclusionRule"]:
zh_trigger_terms = tuple(_dedupe_terms(row.get("zh_trigger_terms") or []))
en_trigger_terms = tuple(_dedupe_terms(row.get("en_trigger_terms") or []))
zh_title_exclusions = tuple(_dedupe_terms(row.get("zh_title_exclusions") or []))
en_title_exclusions = tuple(_dedupe_terms(row.get("en_title_exclusions") or []))
if not zh_title_exclusions and not en_title_exclusions:
return None
if not zh_trigger_terms and not en_trigger_terms:
return None
max_ngram = max(
[1]
+ [len(term.split()) for term in zh_trigger_terms]
+ [len(term.split()) for term in en_trigger_terms]
)
return cls(
zh_trigger_terms=zh_trigger_terms,
en_trigger_terms=en_trigger_terms,
zh_title_exclusions=zh_title_exclusions,
en_title_exclusions=en_title_exclusions,
max_term_ngram=max_ngram,
)
def match_candidates(self, candidates: Iterable[str]) -> Optional[str]:
normalized_candidates = {normalize_query_text(candidate) for candidate in candidates}
for term in self.zh_trigger_terms:
if term in normalized_candidates:
return term
for term in self.en_trigger_terms:
if term in normalized_candidates:
return term
return None
@dataclass(frozen=True)
class DetectedProductTitleExclusion:
matched_term: str
matched_query_text: str
zh_title_exclusions: Tuple[str, ...]
en_title_exclusions: Tuple[str, ...]
def to_dict(self) -> Dict[str, Any]:
return {
"matched_term": self.matched_term,
"matched_query_text": self.matched_query_text,
"zh_title_exclusions": list(self.zh_title_exclusions),
"en_title_exclusions": list(self.en_title_exclusions),
}
@dataclass(frozen=True)
class ProductTitleExclusionProfile:
query_variants: Tuple[TokenizedText, ...] = field(default_factory=tuple)
exclusions: Tuple[DetectedProductTitleExclusion, ...] = field(default_factory=tuple)
@property
def is_active(self) -> bool:
return bool(self.exclusions)
def to_dict(self) -> Dict[str, Any]:
return {
"active": self.is_active,
"exclusions": [item.to_dict() for item in self.exclusions],
"query_variants": [
{
"text": variant.text,
"normalized_text": variant.normalized_text,
"fine_tokens": list(variant.fine_tokens),
"coarse_tokens": list(variant.coarse_tokens),
"candidates": list(variant.candidates),
}
for variant in self.query_variants
],
}
def all_zh_title_exclusions(self) -> List[str]:
return _dedupe_terms(
term
for item in self.exclusions
for term in item.zh_title_exclusions
)
def all_en_title_exclusions(self) -> List[str]:
return _dedupe_terms(
term
for item in self.exclusions
for term in item.en_title_exclusions
)
class ProductTitleExclusionRegistry:
def __init__(
self,
rules: Sequence[ProductTitleExclusionRule],
*,
enabled: bool = True,
) -> None:
self.rules = tuple(rules)
self.enabled = bool(enabled)
self.max_term_ngram = max((rule.max_term_ngram for rule in self.rules), default=3)
@classmethod
def from_query_config(cls, query_config: Any) -> "ProductTitleExclusionRegistry":
raw_rules = getattr(query_config, "product_title_exclusion_rules", []) or []
rules: List[ProductTitleExclusionRule] = []
for row in raw_rules:
if not isinstance(row, dict):
continue
rule = ProductTitleExclusionRule.from_config_row(row)
if rule is not None:
rules.append(rule)
return cls(
rules,
enabled=bool(getattr(query_config, "product_title_exclusion_enabled", True)),
)
class ProductTitleExclusionDetector:
def __init__(
self,
registry: ProductTitleExclusionRegistry,
*,
tokenizer: Optional[Callable[[str], Any]] = None,
) -> None:
self.registry = registry
self.tokenizer = tokenizer
def _build_query_variants(self, parsed_query: Any) -> Tuple[TokenizedText, ...]:
seen = set()
variants: List[TokenizedText] = []
texts = [
getattr(parsed_query, "original_query", None),
getattr(parsed_query, "query_normalized", None),
getattr(parsed_query, "rewritten_query", None),
]
translations = getattr(parsed_query, "translations", {}) or {}
if isinstance(translations, dict):
texts.extend(translations.values())
for raw_text in texts:
text = str(raw_text or "").strip()
if not text:
continue
normalized = normalize_query_text(text)
if not normalized or normalized in seen:
continue
seen.add(normalized)
variants.append(
tokenize_text(
text,
tokenizer=self.tokenizer,
max_ngram=self.registry.max_term_ngram,
)
)
return tuple(variants)
def detect(self, parsed_query: Any) -> ProductTitleExclusionProfile:
if not self.registry.enabled or not self.registry.rules:
return ProductTitleExclusionProfile()
query_variants = self._build_query_variants(parsed_query)
detected: List[DetectedProductTitleExclusion] = []
seen_keys = set()
for variant in query_variants:
for rule in self.registry.rules:
matched_term = rule.match_candidates(variant.candidates)
if not matched_term:
continue
key = (
tuple(rule.zh_title_exclusions),
tuple(rule.en_title_exclusions),
)
if key in seen_keys:
continue
seen_keys.add(key)
detected.append(
DetectedProductTitleExclusion(
matched_term=matched_term,
matched_query_text=variant.text,
zh_title_exclusions=rule.zh_title_exclusions,
en_title_exclusions=rule.en_title_exclusions,
)
)
return ProductTitleExclusionProfile(
query_variants=query_variants,
exclusions=tuple(detected),
)
|