线路编码获取point_id

app/api/checkpoint.py

@@ -0,0 +1,50 @@
+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:
+        point_ids = checkpoint_service.get_point_ids_by_linecode(db, request.linecode)
+
+        return {
+            "code": 0,
+            "message": "查询成功",
+            "data": {
+                "linecode": request.linecode,
+                "point_ids": point_ids,
+                "count": len(point_ids)
+            }
+        }
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"查询失败: {str(e)}")

app/main.py

@@ -14,6 +14,7 @@ from .api.export_excel import router as export_excel_router
 from .api.daily import router as daily_router
 from .api.section_data import router as section_data_router
 from .api.level_data import router as level_data_router
+from .api.checkpoint import router as checkpoint_router
 from .utils.scheduler import task_scheduler
 
 # 初始化日志系统
@@ -76,6 +77,7 @@ app.include_router(export_excel_router, prefix="/api")
 app.include_router(daily_router, prefix="/api")
 app.include_router(section_data_router, prefix="/api")
 app.include_router(level_data_router, prefix="/api")
+app.include_router(checkpoint_router, prefix="/api")
 # app.include_router(test_router, prefix="/api")
 
 # 根路径

app/services/checkpoint.py

@@ -128,3 +128,32 @@ class CheckpointService(BaseService[Checkpoint]):
     def get_by_section_ids(self, db: Session, section_ids: List[str]) -> List[Checkpoint]:
         """根据多个section_id批量获取观测点数据"""
         return db.query(Checkpoint).filter(Checkpoint.section_id.in_(section_ids)).all()
+
+    def get_point_ids_by_linecode(self, db: Session, linecode: str) -> List[str]:
+        """
+        根据水准线路编码获取全部观测点ID
+        业务逻辑：
+        1. linecode在水准数据表(LevelData)查询获取NYID，去重
+        2. NYID在沉降数据表(SettlementData)找到全部沉降数据，提取point_id，去重
+        3. 响应point_id数据集
+
+        使用in查询避免循环查询，提高查询效率
+        """
+        from ..models.level_data import LevelData
+        from ..models.settlement_data import SettlementData
+
+        # 1. 根据linecode查询水准数据表，获取所有NYID（去重）
+        nyid_query = db.query(LevelData.NYID).filter(LevelData.linecode == linecode).distinct()
+        nyid_list = [result.NYID for result in nyid_query.all() if result.NYID]
+
+        if not nyid_list:
+            return []
+
+        # 2. 根据NYID列表查询沉降数据表，获取所有point_id（去重）
+        point_id_query = db.query(SettlementData.point_id).filter(
+            SettlementData.NYID.in_(nyid_list)
+        ).distinct()
+
+        point_id_list = [result.point_id for result in point_id_query.all() if result.point_id]
+
+        return point_id_list