point_description_search.py•3.16 kB
import logging
from typing import List, Optional
from fastapi import HTTPException, APIRouter
from pydantic import BaseModel, Field
from config.config import *
from routers.utils.openplant import OpenPlant
# 全局配置
opt = OpenPlant(host=config_host, port=config_port, timeout=config_timeout)
router = APIRouter()
logger = logging.getLogger("point_description_search")
class PointSearchRequest(BaseModel):
"""测点搜索请求模型
用于通过关键词搜索测点。
典型应用场景：查找特定描述的测点、通过部分名称查找测点等。
"""
keyword: str = Field(
...,
description="搜索关键词，可以是测点名称或描述的一部分",
example="温度",
)
limit: Optional[int] = Field(
default=100,
description="限制返回结果数量",
ge=1,
le=1000,
)
class PointSearchResponse(BaseModel):
"""测点搜索响应模型"""
points: List[dict] = Field(..., description="匹配的测点列表")
total_count: int = Field(..., description="匹配的测点总数")
search_info: dict = Field(..., description="搜索条件信息")
@router.post(
"/api/point_description_search",
response_model=PointSearchResponse,
operation_id="point_description_search",
tags=["测点搜索"],
)
async def point_description_search(request: PointSearchRequest):
"""
通过关键词搜索测点
**参数说明：**
- **keyword**: 搜索关键词，可以是测点名称或描述的一部分
- **limit**: 限制返回结果数量，默认为50
**返回结果：**
- points: 匹配的测点列表
- total_count: 匹配的测点总数
- search_info: 搜索条件信息
**使用示例：**
```json
{
"keyword": "温度",
"limit": 100,
}
```
"""
try:
# 使用OpenPlant API搜索测点
result = opt.api_search_points_by_description(
description_keyword=request.keyword,
limit=request.limit,
)
if not result:
return PointSearchResponse(
points=[],
total_count=0,
search_info={
"keyword": request.keyword,
"message": "未找到匹配的测点"
}
)
# 直接使用API返回的原始结果
return PointSearchResponse(
points=result,
total_count=len(result),
search_info={
"keyword": request.keyword
}
)
except Exception as e:
logger.error(f"搜索测点时发生错误: {str(e)}")
raise HTTPException(
status_code=500,
detail={
"error_type": "搜索错误",
"message": f"搜索测点时发生错误: {str(e)}",
"suggestions": [
"检查搜索关键词是否正确",
"确认数据库连接是否正常",
"尝试使用不同的搜索字段"
]
}
)