be52af70
 tangwang
first commit
|
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
|
"""
Query rewriter for handling synonyms, brand mappings, and query transformations.
"""
from typing import Dict, Optional
import re
class QueryRewriter:
"""Rewrites queries based on configured dictionary rules."""
def __init__(self, rewrite_dict: Dict[str, str] = None):
"""
Initialize query rewriter.
Args:
rewrite_dict: Dictionary mapping query patterns to rewrite expressions
e.g., {"芭比": "brand:芭比 OR name:芭比娃娃"}
"""
self.rewrite_dict = rewrite_dict or {}
def rewrite(self, query: str) -> str:
"""
Rewrite query based on dictionary rules.
Args:
query: Original query string
Returns:
Rewritten query string
"""
if not query or not query.strip():
return query
# Check for exact matches first
if query in self.rewrite_dict:
rewritten = self.rewrite_dict[query]
print(f"[QueryRewriter] Exact match: '{query}' -> '{rewritten}'")
return rewritten
# Check for partial matches (query contains a rewrite key)
for pattern, replacement in self.rewrite_dict.items():
if pattern in query:
# Replace the pattern
rewritten = query.replace(pattern, replacement)
print(f"[QueryRewriter] Partial match: '{query}' -> '{rewritten}'")
return rewritten
# No rewrite needed
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()
|