api-interface-analysis-42918612.plan.2.md
38.4 KB
<!-- 42918612-89ae-45a6-9ccb-e2b0df0a7aec c0a3c6ad-c0f3-447c-8809-bb4c2ef6a845 -->
搜索引擎 API 接口重构实施计划
第一部分：现状分析
1. 当前实现存在的问题
问题 1：硬编码的价格范围过滤
位置：search/es_query_builder.py 第 205-233 行
问题描述：
if field == 'price_ranges':
# 硬编码特定字符串值
if price_range == '0-50':
price_ranges.append({"lt": 50})
elif price_range == '50-100':
price_ranges.append({"gte": 50, "lt": 100})
# ...
影响：
- 只支持
price字段，无法扩展到其他数值字段 - 范围值硬编码，无法根据业务需求调整
- 不符合 SaaS 系统的通用性要求
问题 2：聚合参数直接暴露 ES DSL
位置：
api/models.py第 17 行：aggregations: Optional[Dict[str, Any]]search/es_query_builder.py第 298-319 行：add_dynamic_aggregationsfrontend/static/js/app.js第 57-87 行：前端硬编码 ES DSL
问题描述：
前端需要了解 Elasticsearch 的聚合语法：
const aggregations = {
"category_stats": {
"terms": {
"field": "categoryName_keyword",
"size": 15
}
},
"price_ranges": {
"range": {
"field": "price",
"ranges": [
{"key": "0-50", "to": 50},
// ...
]
}
}
};
影响：
- 前端需要了解 ES 语法，增加集成难度
- 不符合 SaaS 产品易用性原则
- 难以进行参数验证和文档生成
问题 3：分面搜索结果格式不统一
位置：frontend/static/js/app.js 第 208-258 行
问题描述：
- 直接返回 ES 原始格式（
buckets结构） - 前端需要知道不同聚合类型的响应结构
- 没有统一的分面结果模型
影响：
- 前端解析逻辑复杂
- 不同类型的聚合处理方式不一致
- 难以扩展新的聚合类型
问题 4：缺少搜索建议功能
当前状态：完全没有实现
需求：
- 自动补全（Autocomplete）
- 搜索建议（Suggestions）
- 搜索即时反馈（Instant Search）
2. 依赖关系分析
影响范围：
- 后端模型层：
api/models.py - 查询构建层：
search/es_query_builder.py - 搜索执行层：
search/searcher.py - API 路由层：
api/routes/search.py - 前端代码：
frontend/static/js/app.js - 测试代码：
test_aggregation_api.py,test_complete_search.py
第二部分：优化方案设计
方案概述
采用结构化过滤参数方案（方案 A 的简化版）：
- 分离
filters（精确匹配）和range_filters（范围过滤） - 不支持单字段多个不连续范围，简化设计
- 标准化聚合参数，使用简化的接口
- 统一分面搜索响应格式
1. 新的请求模型设计
1.1 核心模型定义
文件：api/models.py
from pydantic import BaseModel, Field, field_validator
from typing import List, Dict, Any, Optional, Union, Literal
class RangeFilter(BaseModel):
"""数值范围过滤器"""
gte: Optional[float] = Field(None, description="大于等于 (>=)")
gt: Optional[float] = Field(None, description="大于 (>)")
lte: Optional[float] = Field(None, description="小于等于 (<=)")
lt: Optional[float] = Field(None, description="小于 (<)")
@field_validator('*')
def check_at_least_one(cls, v, info):
"""确保至少指定一个边界"""
values = info.data
if not any([values.get('gte'), values.get('gt'),
values.get('lte'), values.get('lt')]):
raise ValueError('至少需要指定一个范围边界')
return v
class Config:
json_schema_extra = {
"examples": [
{"gte": 50, "lte": 200},
{"gt": 100},
{"lt": 50}
]
}
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（品牌建议）"
)
1.2 响应模型定义
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="耗时（毫秒）")
2. 查询构建器重构
2.1 移除硬编码的 price_ranges 逻辑
文件：search/es_query_builder.py
需要修改的方法：_build_filters(self, filters, range_filters)
改进点：
- 移除
if field == 'price_ranges'的特殊处理 - 分离 filters 和 range_filters 的处理逻辑
- 添加字段类型验证（利用配置系统）
新的实现逻辑：
def _build_filters(
self,
filters: Optional[Dict[str, Any]] = None,
range_filters: Optional[Dict[str, Any]] = None
) -> List[Dict[str, Any]]:
"""
构建过滤子句（重构版）。
Args:
filters: 精确匹配过滤器字典
range_filters: 范围过滤器字典
Returns:
ES filter子句列表
"""
filter_clauses = []
# 1. 处理精确匹配过滤
if filters:
for field, value in filters.items():
if isinstance(value, list):
# 多值匹配（OR）
filter_clauses.append({
"terms": {field: value}
})
else:
# 单值精确匹配
filter_clauses.append({
"term": {field: value}
})
# 2. 处理范围过滤
if range_filters:
for field, range_spec in range_filters.items():
# 验证字段是否为数值类型（可选，基于配置）
# TODO: 添加字段类型验证
# 构建范围查询
range_conditions = {}
if isinstance(range_spec, dict):
for op in ['gte', 'gt', 'lte', 'lt']:
if op in range_spec and range_spec[op] is not None:
range_conditions[op] = range_spec[op]
if range_conditions:
filter_clauses.append({
"range": {field: range_conditions}
})
return filter_clauses
2.2 优化聚合参数接口
新增方法：build_facets(self, facet_configs)
改进点：
- 移除
add_dynamic_aggregations（直接暴露 ES DSL） - 重构
add_aggregations为更通用的build_facets - 支持简化配置和高级配置两种模式
新的实现逻辑：
def build_facets(
self,
facet_configs: Optional[List[Union[str, Dict[str, Any]]]] = None
) -> Dict[str, Any]:
"""
构建分面聚合（重构版）。
Args:
facet_configs: 分面配置列表。可以是：
- 字符串列表：字段名，使用默认配置
- 配置对象列表：详细的分面配置
Returns:
ES aggregations字典
"""
if not facet_configs:
return {}
aggs = {}
for config in facet_configs:
# 1. 简单模式：只有字段名
if isinstance(config, str):
field = config
agg_name = f"{field}_facet"
aggs[agg_name] = {
"terms": {
"field": field,
"size": 10, # 默认大小
"order": {"_count": "desc"}
}
}
# 2. 高级模式：详细配置对象
elif isinstance(config, dict):
field = config['field']
facet_type = config.get('type', 'terms')
size = config.get('size', 10)
agg_name = f"{field}_facet"
if facet_type == 'terms':
# Terms 聚合（分组统计）
aggs[agg_name] = {
"terms": {
"field": field,
"size": size,
"order": {"_count": "desc"}
}
}
elif facet_type == 'range':
# Range 聚合（范围统计）
ranges = config.get('ranges', [])
if ranges:
aggs[agg_name] = {
"range": {
"field": field,
"ranges": ranges
}
}
return aggs
2.3 更新主查询构建方法
修改方法签名：build_query()
def build_query(
self,
query_text: str,
query_vector: Optional[np.ndarray] = None,
query_node: Optional[QueryNode] = None,
filters: Optional[Dict[str, Any]] = None,
range_filters: Optional[Dict[str, Any]] = None, # 新增
size: int = 10,
from_: int = 0,
enable_knn: bool = True,
knn_k: int = 50,
knn_num_candidates: int = 200,
min_score: Optional[float] = None
) -> Dict[str, Any]:
"""构建完整的 ES 查询（重构版）"""
# ... 实现
# 添加过滤器
if filters or range_filters:
filter_clauses = self._build_filters(filters, range_filters)
if filter_clauses:
es_query["query"] = {
"bool": {
"must": [query_clause],
"filter": filter_clauses
}
}
3. 搜索执行层重构
文件：search/searcher.py
需要修改的方法：search()
改进点：
- 更新方法签名，接受
range_filters参数 - 使用新的
build_facets方法替代旧的聚合逻辑 - 标准化分面搜索结果
关键代码片段：
def search(
self,
query: str,
size: int = 10,
from_: int = 0,
filters: Optional[Dict[str, Any]] = None,
range_filters: Optional[Dict[str, Any]] = None, # 新增
facets: Optional[List[Union[str, Dict]]] = None, # 替代 aggregations
min_score: Optional[float] = None,
sort_by: Optional[str] = None,
sort_order: Optional[str] = "desc",
debug: bool = False,
context: Optional[RequestContext] = None
) -> SearchResult:
"""执行搜索（重构版）"""
# ... 查询解析 ...
# 构建 ES 查询
es_query = self.query_builder.build_multilang_query(
parsed_query=parsed_query,
query_vector=parsed_query.query_vector,
query_node=query_node,
filters=filters,
range_filters=range_filters, # 新增
size=size,
from_=from_,
enable_knn=enable_embedding,
min_score=min_score
)
# 添加分面聚合
if facets:
facet_aggs = self.query_builder.build_facets(facets)
if facet_aggs:
if "aggs" not in es_query:
es_query["aggs"] = {}
es_query["aggs"].update(facet_aggs)
# ... 执行搜索 ...
# 标准化分面结果
standardized_facets = self._standardize_facets(
es_response.get('aggregations', {}),
facets,
filters
)
return SearchResult(
hits=hits,
total=total_value,
max_score=max_score,
took_ms=int(total_duration),
facets=standardized_facets, # 标准化格式
query_info=parsed_query.to_dict(),
debug_info=debug_info
)
新增辅助方法：
def _standardize_facets(
self,
es_aggregations: Dict[str, Any],
facet_configs: Optional[List[Union[str, Dict]]],
current_filters: Optional[Dict[str, Any]]
) -> Optional[List[Dict[str, Any]]]:
"""
将 ES 聚合结果转换为标准化的分面格式。
Args:
es_aggregations: ES 原始聚合结果
facet_configs: 分面配置列表
current_filters: 当前应用的过滤器
Returns:
标准化的分面结果列表
"""
if not es_aggregations or not facet_configs:
return None
standardized_facets = []
for config in facet_configs:
# 解析配置
if isinstance(config, str):
field = config
facet_type = "terms"
else:
field = config['field']
facet_type = config.get('type', 'terms')
agg_name = f"{field}_facet"
if agg_name not in es_aggregations:
continue
agg_result = es_aggregations[agg_name]
# 构建标准化分面结果
facet = {
"field": field,
"label": self._get_field_label(field), # 从配置获取
"type": facet_type,
"values": []
}
# 获取当前字段的选中值
selected_values = set()
if current_filters and field in current_filters:
filter_value = current_filters[field]
if isinstance(filter_value, list):
selected_values = set(filter_value)
else:
selected_values = {filter_value}
# 转换 buckets
if 'buckets' in agg_result:
for bucket in agg_result['buckets']:
value = bucket.get('key')
count = bucket.get('doc_count', 0)
facet['values'].append({
"value": value,
"label": str(value), # 可以从配置映射
"count": count,
"selected": value in selected_values
})
standardized_facets.append(facet)
return standardized_facets
def _get_field_label(self, field: str) -> str:
"""获取字段的显示标签"""
# 从配置中获取字段标签
for field_config in self.config.fields:
if field_config.name == field:
# 假设配置中有 label 字段
return getattr(field_config, 'label', field)
return field
4. API 路由层更新
文件：api/routes/search.py
改进点：
- 接受新的请求模型参数
- 添加搜索建议端点（框架）
新增端点：
@router.get("/suggestions", response_model=SearchSuggestResponse)
async def search_suggestions(
q: str = Query(..., min_length=1, description="搜索查询"),
size: int = Query(5, ge=1, le=20, description="建议数量"),
types: str = Query("query", description="建议类型（逗号分隔）")
):
"""
获取搜索建议（自动补全）。
功能说明：
- 查询建议（query）：基于历史搜索和热门搜索
- 商品建议（product）：匹配的商品
- 类目建议（category）：匹配的类目
- 品牌建议（brand）：匹配的品牌
注意：此功能暂未实现，仅返回框架响应。
"""
import time
start_time = time.time()
# TODO: 实现搜索建议逻辑
# 1. 从搜索历史中获取建议
# 2. 从商品标题中匹配前缀
# 3. 从类目、品牌中匹配
# 临时返回空结果
suggestions = []
# 示例结构（暂不实现）
# suggestions = [
# {
# "text": "芭比娃娃",
# "type": "query",
# "highlight": "<em>芭</em>比娃娃",
# "popularity": 850
# }
# ]
took_ms = int((time.time() - start_time) * 1000)
return SearchSuggestResponse(
query=q,
suggestions=suggestions,
took_ms=took_ms
)
@router.get("/instant", response_model=SearchResponse)
async def instant_search(
q: str = Query(..., min_length=2, description="搜索查询"),
size: int = Query(5, ge=1, le=20, description="结果数量")
):
"""
即时搜索（Instant Search）。
功能说明：
- 边输入边搜索，无需点击搜索按钮
- 返回简化的搜索结果
- 性能优化：缓存、限流
注意：此功能暂未实现，调用标准搜索接口。
"""
# TODO: 优化即时搜索性能
# 1. 添加防抖/节流
# 2. 实现结果缓存
# 3. 简化返回字段
# 临时使用标准搜索接口
from api.app import get_searcher
searcher = get_searcher()
result = searcher.search(
query=q,
size=size,
from_=0
)
return SearchResponse(
hits=result.hits,
total=result.total,
max_score=result.max_score,
took_ms=result.took_ms,
query_info=result.query_info
)
5. 前端适配
文件：frontend/static/js/app.js
需要修改的地方：
- 聚合参数改用简化配置（第 57-87 行）： ```javascript // 旧的方式（直接 ES DSL） const aggregations = { "category_stats": { "terms": { "field": "categoryName_keyword", "size": 15 } } };
// 新的方式（简化配置） const facets = [ { "field": "categoryName_keyword", "size": 15, "type": "terms" }, { "field": "brandName_keyword", "size": 15, "type": "terms" }, { "field": "price", "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} ] } ];
2. **过滤器使用新格式**（第 103 行）：
```javascript
// 旧的方式
filters: {
"price_ranges": ["0-50", "50-100"] // 硬编码
}
// 新的方式
filters: {
"categoryName_keyword": ["玩具"],
"in_stock": true
},
range_filters: {
"price": {"gte": 50, "lte": 100}
}
- 解析标准化的分面结果（第 208-258 行）： ```javascript // 旧的方式（直接访问 ES 结构） if (aggregations.category_stats && aggregations.category_stats.buckets) { aggregations.category_stats.buckets.forEach(bucket => { // ... }); }
// 新的方式（标准化格式） if (data.facets) { data.facets.forEach(facet => { if (facet.field === 'categoryName_keyword') { facet.values.forEach(facetValue => { const value = facetValue.value; const count = facetValue.count; const selected = facetValue.selected; // ... }); } }); }
### 6. 测试代码更新
**文件**：`test_aggregation_api.py`
**需要修改的地方**：
1. 移除 `price_ranges` 硬编码测试（第 93 行）
2. 使用新的 `range_filters` 格式
3. 使用新的 `facets` 配置
**新的测试代码**：
```python
def test_search_with_filters():
"""测试新的过滤器格式"""
test_request = {
"query": "玩具",
"size": 5,
"filters": {
"categoryName_keyword": ["玩具"]
},
"range_filters": {
"price": {"gte": 50, "lte": 100}
}
}
# ...
def test_search_with_facets():
"""测试新的分面配置"""
test_request = {
"query": "玩具",
"size": 20,
"facets": [
{
"field": "categoryName_keyword",
"size": 15
},
{
"field": "price",
"type": "range",
"ranges": [
{"key": "0-50", "to": 50},
{"key": "50-100", "from": 50, "to": 100}
]
}
]
}
# ...
第三部分：实施步骤
阶段 1：后端模型层重构（高优先级）
任务清单：
- [ ] 更新
api/models.py- [ ] 定义RangeFilter模型 - [ ] 定义FacetConfig模型 - [ ] 更新SearchRequest，添加range_filters和facets- [ ] 移除aggregations参数 - [ ] 定义FacetValue和FacetResult模型 - [ ] 更新SearchResponse，使用标准化分面格式 - [ ] 添加SearchSuggestRequest和SearchSuggestResponse（框架）
验证方式：
- 运行 Pydantic 模型验证
- 检查 API 文档（
/docs）是否正确生成
阶段 2：查询构建器重构（高优先级）
任务清单：
- [ ] 重构
search/es_query_builder.py- [ ] 移除price_ranges硬编码逻辑（第 205-233 行） - [ ] 重构_build_filters方法，支持range_filters- [ ] 移除add_dynamic_aggregations方法 - [ ] 重构add_aggregations为build_facets- [ ] 更新build_query方法签名 - [ ] 更新
search/multilang_query_builder.py（如果需要）
验证方式：
- 编写单元测试验证过滤器构建逻辑
- 打印生成的 ES DSL，检查正确性
阶段 3：搜索执行层重构（高优先级）
任务清单：
- [ ] 更新
search/searcher.py- [ ] 更新search()方法签名 - [ ] 使用新的build_facets方法 - [ ] 实现_standardize_facets()辅助方法 - [ ] 实现_get_field_label()辅助方法 - [ ] 更新SearchResult类，使用标准化分面格式
验证方式：
- 编写集成测试
- 手动测试搜索功能
阶段 4：API 路由层更新（中优先级）
任务清单：
- [ ] 更新
api/routes/search.py- [ ] 更新/search/端点，接受新的请求参数 - [ ] 添加/search/suggestions端点（框架，返回空结果） - [ ] 添加/search/instant端点（框架，调用标准搜索） - [ ] 添加端点文档和示例
验证方式：
- 使用 Swagger UI 测试端点
- 检查 API 文档完整性
阶段 5：前端适配（中优先级）
任务清单：
- [ ] 更新
frontend/static/js/app.js- [ ] 修改聚合参数为facets简化配置 - [ ] 修改过滤器参数，分离filters和range_filters- [ ] 更新displayAggregations()方法，解析标准化分面结果 - [ ] 添加范围过滤器 UI（如价格滑块） - [ ] 移除硬编码的price_ranges
验证方式：
- 浏览器测试前端功能
- 检查网络请求和响应格式
阶段 6：测试代码更新（低优先级）
任务清单：
- [ ] 更新
test_aggregation_api.py- [ ] 移除price_ranges测试 - [ ] 添加range_filters测试 - [ ] 添加新的facets测试 - [ ] 更新
test_complete_search.py - [ ] 更新
tests/integration/test_aggregation_api.py - [ ] 更新
tests/unit/test_searcher.py
验证方式：
- 运行所有测试，确保通过
- 检查测试覆盖率
阶段 7：文档更新（低优先级）
任务清单：
- [ ] 撰写完整的 API 接口文档
- [ ] 更新
README.md - [ ] 更新
USER_GUIDE.md - [ ] 添加接口使用示例
- [ ] 添加迁移指南（旧接口 → 新接口）
第四部分：API 使用示例
示例 1：简单搜索
POST /search/
{
"query": "芭比娃娃",
"size": 20
}
示例 2：带过滤器的搜索
POST /search/
{
"query": "玩具",
"size": 20,
"filters": {
"categoryName_keyword": ["玩具", "益智玩具"],
"in_stock": true
},
"range_filters": {
"price": {"gte": 50, "lte": 200}
}
}
示例 3：带分面搜索的请求
POST /search/
{
"query": "玩具",
"size": 20,
"facets": [
{
"field": "categoryName_keyword",
"size": 15
},
{
"field": "brandName_keyword",
"size": 15
},
{
"field": "price",
"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}
]
}
]
}
响应示例（标准化分面格式）：
{
"hits": [...],
"total": 118,
"max_score": 8.5,
"took_ms": 45,
"facets": [
{
"field": "categoryName_keyword",
"label": "商品类目",
"type": "terms",
"values": [
{"value": "玩具", "label": "玩具", "count": 85, "selected": false},
{"value": "益智玩具", "label": "益智玩具", "count": 33, "selected": false}
]
},
{
"field": "price",
"label": "价格区间",
"type": "range",
"values": [
{"value": "0-50", "label": "0-50元", "count": 23, "selected": false},
{"value": "50-100", "label": "50-100元", "count": 45, "selected": false},
{"value": "100-200", "label": "100-200元", "count": 38, "selected": false},
{"value": "200+", "label": "200元以上", "count": 12, "selected": false}
]
}
]
}
示例 4：搜索建议（框架）
GET /search/suggestions?q=芭&size=5
{
"query": "芭",
"suggestions": [
{
"text": "芭比娃娃",
"type": "query",
"highlight": "<em>芭</em>比娃娃",
"popularity": 850
},
{
"text": "芭比娃娃屋",
"type": "query",
"highlight": "<em>芭</em>比娃娃屋",
"popularity": 320
}
],
"took_ms": 5
}
第五部分：向后兼容性
兼容策略
为保持向后兼容，在过渡期（1-2 个版本）内：
同时支持旧参数和新参数：
class SearchRequest(BaseModel): # 新参数 range_filters: Optional[Dict[str, RangeFilter]] = None facets: Optional[List[Union[str, FacetConfig]]] = None # 旧参数（标记为废弃） aggregations: Optional[Dict[str, Any]] = Field( None, deprecated=True, description="已废弃。请使用 'facets' 参数" )在后端自动转换旧格式：
# 在 searcher.py 中 if request.aggregations and not request.facets: # 将旧的 aggregations 转换为新的 facets request.facets = self._convert_legacy_aggregations(request.aggregations)在响应中提供迁移提示：
if request.aggregations: warnings.append({ "type": "deprecation", "message": "'aggregations' 参数已废弃，请使用 'facets' 参数", "migration_guide": "https://docs.example.com/migration" })
迁移时间线
- v3.0（当前版本）：发布新接口，旧接口标记为废弃
- v3.1（1 个月后）：移除旧接口的自动转换
- v4.0（3 个月后）：完全移除旧接口
第六部分：风险评估与缓解
风险点
破坏性变更风险：
- 风险：现有客户代码可能依赖旧接口 - 缓解：提供向后兼容层，发布详细迁移指南性能影响风险：
- 风险：新的标准化处理可能增加延迟 - 缓解：添加性能测试，优化关键路径测试覆盖不足风险：
- 风险：重构可能引入新 bug - 缓解：全面的单元测试和集成测试
验收标准
- [ ] 所有单元测试通过
- [ ] 所有集成测试通过
- [ ] API 文档完整且准确
- [ ] 性能无明显下降（
- [ ] 前端功能正常工作
- [ ] 提供完整的迁移指南
总结
本计划通过系统性的重构，将搜索 API 从硬编码、暴露 ES 细节的实现，转变为灵活、通用、易用的 SaaS 产品接口。关键改进包括：
- ✅ 移除硬编码的 price_ranges 逻辑
- ✅ 实现结构化的过滤参数（filters + range_filters）
- ✅ 简化聚合参数接口，不暴露 ES DSL
- ✅ 标准化分面搜索响应格式
- ✅ 添加搜索建议功能框架（暂不实现）
通过这些改进，系统将具备更好的通用性、可维护性和可扩展性，为未来功能扩展奠定基础。