query_rewriter.py
4.3 KB
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
"""
Query rewriter for handling synonyms, brand mappings, and query transformations.
"""
from typing import Dict, Optional
import re
class QueryRewriter:
"""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.
"""
def __init__(self, rewrite_dict: Dict[str, str] = None):
"""
Initialize query rewriter for exact word matching only.
Args:
rewrite_dict: Dictionary mapping exact query terms to rewrite expressions
e.g., {"芭比": "brand:芭比 OR name:芭比娃娃"}
Only full word matches will be rewritten, no partial matching.
"""
self.rewrite_dict = rewrite_dict or {}
def rewrite(self, query: str) -> str:
"""
Rewrite query based on dictionary rules using exact word matching only.
Args:
query: Original query string
Returns:
Rewritten query string
"""
if not query or not query.strip():
return query
# Only check for exact matches - no partial matching
if query in self.rewrite_dict:
rewritten = self.rewrite_dict[query]
print(f"[QueryRewriter] Exact match: '{query}' -> '{rewritten}'")
return rewritten
# No rewrite needed - no partial matching
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()