be52af70
tangwang
first commit
|
1
2
3
4
5
6
7
8
9
|
"""
Query rewriter for handling synonyms, brand mappings, and query transformations.
"""
from typing import Dict, Optional
import re
class QueryRewriter:
|
bb52dba6
tangwang
API接口设计优化:
|
10
11
12
13
14
|
"""Rewrites queries based on exact word matching with configured dictionary rules.
Only performs full word matching - no partial matching or substring replacement.
The entire query must exactly match a key in the rewrite dictionary to be rewritten.
"""
|
be52af70
tangwang
first commit
|
15
16
17
|
def __init__(self, rewrite_dict: Dict[str, str] = None):
"""
|
bb52dba6
tangwang
API接口设计优化:
|
18
|
Initialize query rewriter for exact word matching only.
|
be52af70
tangwang
first commit
|
19
20
|
Args:
|
bb52dba6
tangwang
API接口设计优化:
|
21
|
rewrite_dict: Dictionary mapping exact query terms to rewrite expressions
|
be52af70
tangwang
first commit
|
22
|
e.g., {"芭比": "brand:芭比 OR name:芭比娃娃"}
|
bb52dba6
tangwang
API接口设计优化:
|
23
|
Only full word matches will be rewritten, no partial matching.
|
be52af70
tangwang
first commit
|
24
25
26
27
28
|
"""
self.rewrite_dict = rewrite_dict or {}
def rewrite(self, query: str) -> str:
"""
|
bb52dba6
tangwang
API接口设计优化:
|
29
|
Rewrite query based on dictionary rules using exact word matching only.
|
be52af70
tangwang
first commit
|
30
31
32
33
34
35
36
37
38
39
|
Args:
query: Original query string
Returns:
Rewritten query string
"""
if not query or not query.strip():
return query
|
bb52dba6
tangwang
API接口设计优化:
|
40
|
# Only check for exact matches - no partial matching
|
be52af70
tangwang
first commit
|
41
42
43
44
45
|
if query in self.rewrite_dict:
rewritten = self.rewrite_dict[query]
print(f"[QueryRewriter] Exact match: '{query}' -> '{rewritten}'")
return rewritten
|
bb52dba6
tangwang
API接口设计优化:
|
46
|
# No rewrite needed - no partial matching
|
be52af70
tangwang
first commit
|
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
|
return query
def add_rule(self, pattern: str, replacement: str) -> None:
"""
Add a rewrite rule.
Args:
pattern: Query pattern to match
replacement: Replacement expression
"""
self.rewrite_dict[pattern] = replacement
def remove_rule(self, pattern: str) -> bool:
"""
Remove a rewrite rule.
Args:
pattern: Query pattern to remove
Returns:
True if rule was removed, False if not found
"""
if pattern in self.rewrite_dict:
del self.rewrite_dict[pattern]
return True
return False
def get_rules(self) -> Dict[str, str]:
"""Get all rewrite rules."""
return self.rewrite_dict.copy()
def clear_rules(self) -> None:
"""Clear all rewrite rules."""
self.rewrite_dict.clear()
class QueryNormalizer:
"""Normalizes queries for consistent processing."""
@staticmethod
def normalize(query: str) -> str:
"""
Normalize query string.
- Trim whitespace
- Convert multiple spaces to single space
- Remove special characters (optional)
Args:
query: Original query
Returns:
Normalized query
"""
if not query:
return ""
# Trim and collapse whitespace
query = " ".join(query.split())
return query
@staticmethod
def remove_punctuation(query: str, keep_operators: bool = True) -> str:
"""
Remove punctuation from query.
Args:
query: Original query
keep_operators: Whether to keep boolean operators (AND, OR, etc.)
Returns:
Query without punctuation
"""
if not query:
return ""
if keep_operators:
# Keep alphanumeric, spaces, and operator characters
pattern = r'[^a-zA-Z0-9\u4e00-\u9fff\u0400-\u04ff\s\(\)|&!-]'
else:
# Keep only alphanumeric and spaces
pattern = r'[^a-zA-Z0-9\u4e00-\u9fff\u0400-\u04ff\s]'
return re.sub(pattern, '', query)
@staticmethod
def extract_domain_query(query: str) -> tuple:
"""
Extract domain prefix from query if present.
Examples:
"brand:Nike shoes" -> ("brand", "Nike shoes")
"category:toys" -> ("category", "toys")
"default query" -> ("default", "default query")
Args:
query: Query string
Returns:
Tuple of (domain, query_text)
"""
# Check for domain:query pattern
match = re.match(r'^(\w+):(.+)$', query.strip())
if match:
return match.group(1), match.group(2).strip()
# No domain specified, use default
return "default", query.strip()
|