75 lines
2.3 KiB
Python
75 lines
2.3 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
||
from pydantic import BaseModel
|
||
from sqlalchemy.orm import Session
|
||
from typing import List, Optional, Dict, Any
|
||
from ..core.database import get_db
|
||
from ..services.checkpoint import CheckpointService
|
||
from ..schemas.comprehensive_data import DataResponse
|
||
|
||
router = APIRouter(prefix="/checkpoint", tags=["checkpoint"])
|
||
|
||
class PointIdsByLinecodeRequest(BaseModel):
|
||
linecode: str
|
||
|
||
class PointIdsResponse(BaseModel):
|
||
point_ids: List[str]
|
||
count: int
|
||
|
||
@router.post("/point-ids-by-linecode", response_model=Dict[str, Any])
|
||
async def get_point_ids_by_linecode(
|
||
request: PointIdsByLinecodeRequest,
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""
|
||
根据水准线路编码获取全部观测点ID
|
||
|
||
业务逻辑：
|
||
1. linecode在水准数据表(LevelData)查询获取NYID，去重
|
||
2. NYID在沉降数据表(SettlementData)找到全部沉降数据，提取point_id，去重
|
||
3. 响应point_id数据集
|
||
|
||
使用in查询避免循环查询，提高查询效率
|
||
|
||
- **linecode**: 水准线路编码
|
||
"""
|
||
checkpoint_service = CheckpointService()
|
||
|
||
try:
|
||
print( f"Fetching point IDs for linecode: {request.linecode}" )
|
||
point_ids, point_id_dict = checkpoint_service.get_point_ids_by_linecode(db, request.linecode)
|
||
|
||
# 检查结果是否为空
|
||
if not point_ids or len(point_ids) == 0:
|
||
return {
|
||
"code": 0,
|
||
"message": f"未找到线路编码 '{request.linecode}' 对应的观测点数据",
|
||
"data": {
|
||
"linecode": request.linecode,
|
||
"point_ids": [],
|
||
"point_id_dict": {},
|
||
"count": 0
|
||
}
|
||
}
|
||
|
||
return {
|
||
"code": 0,
|
||
"message": "查询成功",
|
||
"data": {
|
||
"linecode": request.linecode,
|
||
"point_ids": point_ids,
|
||
"point_id_dict": point_id_dict,
|
||
"count": len(point_ids)
|
||
}
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
"code": 500,
|
||
"message": f"查询失败: {str(e)}",
|
||
"data": {
|
||
"linecode": request.linecode,
|
||
"point_ids": [],
|
||
"point_id_dict": {},
|
||
"count": 0
|
||
}
|
||
}
|