models.py
8.96 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
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
226
227
228
229
"""
Request and response models for the API.
"""
from pydantic import BaseModel, Field, field_validator
from typing import List, Dict, Any, Optional, Union, Literal
class RangeFilter(BaseModel):
"""范围过滤器（支持数值和日期时间字符串）"""
gte: Optional[Union[float, str]] = Field(None, description="大于等于 (>=)。数值或ISO日期时间字符串")
gt: Optional[Union[float, str]] = Field(None, description="大于 (>)。数值或ISO日期时间字符串")
lte: Optional[Union[float, str]] = Field(None, description="小于等于 (<=)。数值或ISO日期时间字符串")
lt: Optional[Union[float, str]] = Field(None, description="小于 (<)。数值或ISO日期时间字符串")
def model_post_init(self, __context):
"""确保至少指定一个边界值"""
if not any([self.gte, self.gt, self.lte, self.lt]):
raise ValueError('至少需要指定一个范围边界（gte, gt, lte, lt）')
class Config:
json_schema_extra = {
"examples": [
{"gte": 50, "lte": 200},
{"gt": 100},
{"lt": 50},
{"gte": "2023-01-01T00:00:00Z"}
]
}
class FacetConfig(BaseModel):
"""分面配置（简化版）"""
field: str = Field(..., description="分面字段名")
size: int = Field(10, ge=1, le=100, description="返回的分面值数量")
type: Literal["terms", "range"] = Field("terms", description="分面类型")
ranges: Optional[List[Dict[str, Any]]] = Field(
None,
description="范围分面的范围定义（仅当 type='range' 时需要）"
)
class Config:
json_schema_extra = {
"examples": [
{
"field": "categoryName_keyword",
"size": 15,
"type": "terms"
},
{
"field": "price",
"size": 4,
"type": "range",
"ranges": [
{"key": "0-50", "to": 50},
{"key": "50-100", "from": 50, "to": 100},
{"key": "100-200", "from": 100, "to": 200},
{"key": "200+", "from": 200}
]
}
]
}
class SearchRequest(BaseModel):
"""搜索请求模型（重构版）"""
# 基础搜索参数
query: str = Field(..., description="搜索查询字符串，支持布尔表达式（AND, OR, RANK, ANDNOT）")
size: int = Field(10, ge=1, le=100, description="返回结果数量")
from_: int = Field(0, ge=0, alias="from", description="分页偏移量")
# 过滤器 - 精确匹配和多值匹配
filters: Optional[Dict[str, Union[str, int, bool, List[Union[str, int]]]]] = Field(
None,
description="精确匹配过滤器。单值表示精确匹配，数组表示 OR 匹配（匹配任意一个值）",
json_schema_extra={
"examples": [
{
"categoryName_keyword": ["玩具", "益智玩具"],
"brandName_keyword": "乐高",
"in_stock": True
}
]
}
)
# 范围过滤器 - 数值范围
range_filters: Optional[Dict[str, RangeFilter]] = Field(
None,
description="数值范围过滤器。支持 gte, gt, lte, lt 操作符",
json_schema_extra={
"examples": [
{
"price": {"gte": 50, "lte": 200},
"days_since_last_update": {"lte": 30}
}
]
}
)
# 排序
sort_by: Optional[str] = Field(None, description="排序字段名（如 'price', 'create_time'）")
sort_order: Optional[str] = Field("desc", description="排序方向: 'asc'（升序）或 'desc'（降序）")
# 分面搜索 - 简化接口
facets: Optional[List[Union[str, FacetConfig]]] = Field(
None,
description="分面配置。可以是字段名列表（使用默认配置）或详细的分面配置对象",
json_schema_extra={
"examples": [
# 简单模式：只指定字段名，使用默认配置
["categoryName_keyword", "brandName_keyword"],
# 高级模式：详细配置
[
{"field": "categoryName_keyword", "size": 15},
{
"field": "price",
"type": "range",
"ranges": [
{"key": "0-50", "to": 50},
{"key": "50-100", "from": 50, "to": 100}
]
}
]
]
}
)
# 高级选项
min_score: Optional[float] = Field(None, ge=0, description="最小相关性分数阈值")
highlight: bool = Field(False, description="是否高亮搜索关键词（暂不实现）")
debug: bool = Field(False, description="是否返回调试信息")
# 个性化参数（预留）
user_id: Optional[str] = Field(None, description="用户ID，用于个性化搜索和推荐")
session_id: Optional[str] = Field(None, description="会话ID，用于搜索分析")
class ImageSearchRequest(BaseModel):
"""图片搜索请求模型"""
image_url: str = Field(..., description="查询图片的 URL")
size: int = Field(10, ge=1, le=100, description="返回结果数量")
filters: Optional[Dict[str, Union[str, int, bool, List[Union[str, int]]]]] = None
range_filters: Optional[Dict[str, RangeFilter]] = None
class SearchSuggestRequest(BaseModel):
"""搜索建议请求模型（框架，暂不实现）"""
query: str = Field(..., min_length=1, description="搜索查询字符串")
size: int = Field(5, ge=1, le=20, description="返回建议数量")
types: List[Literal["query", "product", "category", "brand"]] = Field(
["query"],
description="建议类型：query（查询建议）, product（商品建议）, category（类目建议）, brand（品牌建议）"
)
class FacetValue(BaseModel):
"""分面值"""
value: Union[str, int, float] = Field(..., description="分面值")
label: Optional[str] = Field(None, description="显示标签（如果与 value 不同）")
count: int = Field(..., description="匹配的文档数量")
selected: bool = Field(False, description="是否已选中（当前过滤器中）")
class FacetResult(BaseModel):
"""分面结果（标准化格式）"""
field: str = Field(..., description="字段名")
label: str = Field(..., description="分面显示名称")
type: Literal["terms", "range"] = Field(..., description="分面类型")
values: List[FacetValue] = Field(..., description="分面值列表")
total_count: Optional[int] = Field(None, description="该字段的总文档数")
class SearchResponse(BaseModel):
"""搜索响应模型（重构版）"""
# 核心结果
hits: List[Dict[str, Any]] = Field(..., description="搜索结果列表")
total: int = Field(..., description="匹配的总文档数")
max_score: float = Field(..., description="最高相关性分数")
# 分面搜索结果（标准化格式）
facets: Optional[List[FacetResult]] = Field(
None,
description="分面统计结果（标准化格式）"
)
# 查询信息
query_info: Dict[str, Any] = Field(
default_factory=dict,
description="查询处理信息（原始查询、改写、语言检测、翻译等）"
)
# 推荐与建议（预留）
related_queries: Optional[List[str]] = Field(None, description="相关搜索查询")
# 性能指标
took_ms: int = Field(..., description="搜索总耗时（毫秒）")
performance_info: Optional[Dict[str, Any]] = Field(None, description="详细性能信息")
# 调试信息
debug_info: Optional[Dict[str, Any]] = Field(None, description="调试信息（仅当 debug=True）")
class SearchSuggestResponse(BaseModel):
"""搜索建议响应模型（框架，暂不实现）"""
query: str = Field(..., description="原始查询")
suggestions: List[Dict[str, Any]] = Field(..., description="建议列表")
took_ms: int = Field(..., description="耗时（毫秒）")
class DocumentResponse(BaseModel):
"""Single document response model."""
id: str = Field(..., description="Document ID")
source: Dict[str, Any] = Field(..., description="Document source")
class HealthResponse(BaseModel):
"""Health check response model."""
status: str = Field(..., description="Service status")
elasticsearch: str = Field(..., description="Elasticsearch status")
customer_id: str = Field(..., description="Customer configuration ID")
class ErrorResponse(BaseModel):
"""Error response model."""
error: str = Field(..., description="Error message")
detail: Optional[str] = Field(None, description="Detailed error information")