models.py
13.6 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
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
"""
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": "category.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="分页偏移量")
language: Literal["zh", "en"] = Field(
"zh",
description="响应语言：'zh'（中文）或 'en'（英文），用于选择 title/description/vendor 等多语言字段"
)
# 过滤器 - 精确匹配和多值匹配
filters: Optional[Dict[str, Union[str, int, bool, List[Union[str, int]], Dict[str, Any], List[Dict[str, Any]]]]] = Field(
None,
description="精确匹配过滤器。单值表示精确匹配，数组表示 OR 匹配（匹配任意一个值）。支持 specifications 嵌套过滤：{\"specifications\": {\"name\": \"color\", \"value\": \"green\"}} 或 [{\"name\": \"color\", \"value\": \"green\"}, ...]",
json_schema_extra={
"examples": [
{
"category_name": ["手机", "电子产品"],
"vendor_zh.keyword": "奇乐",
"specifications": {"name": "颜色", "value": "白色"}
},
{
"specifications": [
{"name": "颜色", "value": "白色"},
{"name": "尺寸", "value": "256GB"}
]
}
]
}
)
# 范围过滤器 - 数值范围
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="排序字段名（如 'min_price', 'max_price', 'title'）")
sort_order: Optional[str] = Field("desc", description="排序方向: 'asc'（升序）或 'desc'（降序）")
# 分面搜索 - 简化接口
facets: Optional[List[Union[str, FacetConfig]]] = Field(
None,
description="分面配置。可以是字段名列表（使用默认配置）或详细的分面配置对象。支持 specifications 分面：\"specifications\"（所有name）或 \"specifications.color\"（指定name）",
json_schema_extra={
"examples": [
# 简单模式：只指定字段名，使用默认配置
["category1_name", "category2_name", "specifications"],
# 指定specifications的某个name
["specifications.颜色", "specifications.尺寸"],
# 高级模式：详细配置
[
{"field": "category1_name", "size": 15},
{
"field": "min_price",
"type": "range",
"ranges": [
{"key": "0-50", "to": 50},
{"key": "50-100", "from": 50, "to": 100}
]
},
"specifications" # 所有specifications name的分面
]
]
}
)
# 高级选项
min_score: Optional[float] = Field(None, ge=0, description="最小相关性分数阈值")
highlight: bool = Field(False, description="是否高亮搜索关键词（暂不实现）")
debug: bool = Field(False, description="是否返回调试信息")
# SKU筛选参数
sku_filter_dimension: Optional[str] = Field(
None,
description="子SKU筛选维度（店铺配置）。指定后，每个SPU下的SKU将按该维度分组，每组选择第一个SKU返回。例如：'color'表示按颜色分组，每种颜色选一款。支持的值：'option1'、'option2'、'option3'或specifications中的name（如'color'、'size'）"
)
# 个性化参数（预留）
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 SkuResult(BaseModel):
"""SKU 结果"""
sku_id: str = Field(..., description="SKU ID")
# 与 ES nested skus 结构对齐
price: Optional[float] = Field(None, description="价格")
compare_at_price: Optional[float] = Field(None, description="原价")
sku_code: Optional[str] = Field(None, description="SKU编码")
stock: int = Field(0, description="库存数量")
weight: Optional[float] = Field(None, description="重量")
weight_unit: Optional[str] = Field(None, description="重量单位")
option1_value: Optional[str] = Field(None, description="选项1取值（如颜色）")
option2_value: Optional[str] = Field(None, description="选项2取值（如尺码）")
option3_value: Optional[str] = Field(None, description="选项3取值")
image_src: Optional[str] = Field(None, description="SKU图片地址")
class SpuResult(BaseModel):
"""SPU 搜索结果"""
spu_id: str = Field(..., description="SPU ID")
title: Optional[str] = Field(None, description="商品标题")
brief: Optional[str] = Field(None, description="商品短描述")
handle: Optional[str] = Field(None, description="商品handle")
description: Optional[str] = Field(None, description="商品描述")
vendor: Optional[str] = Field(None, description="供应商/品牌")
category: Optional[str] = Field(None, description="类目（兼容字段，等同于category_name）")
category_path: Optional[str] = Field(None, description="类目路径（多级，用于面包屑）")
category_name: Optional[str] = Field(None, description="类目名称（展示用）")
category_id: Optional[str] = Field(None, description="类目ID")
category_level: Optional[int] = Field(None, description="类目层级")
category1_name: Optional[str] = Field(None, description="一级类目名称")
category2_name: Optional[str] = Field(None, description="二级类目名称")
category3_name: Optional[str] = Field(None, description="三级类目名称")
tags: Optional[List[str]] = Field(None, description="标签列表")
price: Optional[float] = Field(None, description="价格（min_price）")
compare_at_price: Optional[float] = Field(None, description="原价")
currency: str = Field("USD", description="货币单位")
image_url: Optional[str] = Field(None, description="主图URL")
in_stock: bool = Field(True, description="是否有库存")
# SKU 扁平化信息
sku_prices: Optional[List[float]] = Field(None, description="所有SKU价格列表")
sku_weights: Optional[List[int]] = Field(None, description="所有SKU重量列表")
sku_weight_units: Optional[List[str]] = Field(None, description="所有SKU重量单位列表")
total_inventory: Optional[int] = Field(None, description="总库存")
option1_name: Optional[str] = Field(None, description="选项1名称（如颜色）")
option2_name: Optional[str] = Field(None, description="选项2名称（如尺码）")
option3_name: Optional[str] = Field(None, description="选项3名称")
specifications: Optional[List[Dict[str, Any]]] = Field(
None,
description="规格列表（与 ES specifications 字段对应）"
)
skus: List[SkuResult] = Field(default_factory=list, description="SKU列表")
relevance_score: float = Field(..., ge=0.0, description="相关性分数（ES原始分数）")
class SearchResponse(BaseModel):
"""搜索响应模型（外部友好格式）"""
# 核心结果
results: List[SpuResult] = 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="查询处理信息（原始查询、改写、语言检测、翻译等）"
)
# 推荐与建议
suggestions: List[str] = Field(default_factory=list, description="搜索建议")
related_searches: List[str] = Field(default_factory=list, 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")
class ErrorResponse(BaseModel):
"""Error response model."""
error: str = Field(..., description="Error message")
detail: Optional[str] = Field(None, description="Detailed error information")