app/api/point_unit_grouping.py

@@ -0,0 +1,13 @@
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+from app.database import get_db
+from app.services import point_unit_grouping_service
+
+router = APIRouter(prefix="/point-unit-grouping", tags=["point-unit-grouping"])
+
+@router.get("/get-h-xtfx-result")
+def get_h_xtfx_result(db: Session = Depends(get_db)):
+    """接口：返回单元 h_xtfx结果"""
+    unit_rows, point_rows = point_unit_grouping_service.fetch_unit_and_point_data(db)
+    result = point_unit_grouping_service.check_h_xtfx_values(unit_rows, point_rows)
+    return result

app/main.py

@@ -1,5 +1,6 @@
 from fastapi import FastAPI
-from .api import vector, raster
+
+from .api import vector, raster,point_unit_grouping
 from .database import engine, Base
 
 # 创建数据库表
@@ -24,6 +25,8 @@ app = FastAPI(
 # 注册路由
 app.include_router(vector.router, prefix="/api/vector", tags=["vector"])
 app.include_router(raster.router, prefix="/api/raster", tags=["raster"])
+app.include_router(point_unit_grouping.router, prefix="/api/point-unit-grouping", tags=["point-unit-grouping"])
+
 
 @app.get("/")
 async def root():

point_unit_grouping_api.py → app/services/point_unit_grouping_service.py

@@ -2,7 +2,9 @@ import psycopg2
 from flask import Flask, jsonify
 from shapely import wkb
 from collections import Counter
-
+from app.database import engine  
+from sqlalchemy.orm import Session  
+from sqlalchemy import text  
 app = Flask(__name__)
 
 # 定义 h_xtfx 值的映射关系（数值用于插值计算）
@@ -14,44 +16,22 @@ h_xtfx_mapping = {
 reverse_h_xtfx_mapping = {v: k for k, v in h_xtfx_mapping.items()}
 
 
-def connect_to_database():
-    """建立数据库连接"""
-    try:
-        connection = psycopg2.connect(
-            user="postgres",
-            password="root",
-            host="localhost",
-            port="5432",
-            database="testdb1"
-        )
-        return connection
-    except Exception as error:
-        print("数据库连接失败:", error)
-        return None
-
 
-def fetch_unit_and_point_data(connection):
-    """获取单元和点位数据"""
-    if not connection:
-        return [], []
-    cursor = connection.cursor()
+def fetch_unit_and_point_data(db: Session):
+    """获取单元和点位数据（使用 SQLAlchemy ORM Session）"""
     try:
         # 查询单元信息（GID 和 几何图形）
-        unit_query = "SELECT gid, geom FROM unit_ceil"
-        cursor.execute(unit_query)
-        unit_rows = cursor.fetchall()
-        
+        unit_query = text("SELECT gid, geom FROM unit_ceil")
+        unit_result = db.execute(unit_query).fetchall()
+
         # 查询点位信息（ID、几何图形、h_xtfx 值）
-        point_query = "SELECT id, geom, h_xtfx FROM fifty_thousand_survey_data"
-        cursor.execute(point_query)
-        point_rows = cursor.fetchall()
-        
-        return unit_rows, point_rows
+        point_query = text("SELECT id, geom, h_xtfx FROM fifty_thousand_survey_data")
+        point_result = db.execute(point_query).fetchall()
+
+        return unit_result, point_result
     except Exception as error:
         print("数据查询失败:", error)
         return [], []
-    finally:
-        cursor.close()
 
 
 def idw_interpolation(points, target_point):
@@ -130,19 +110,4 @@ def check_h_xtfx_values(unit_rows, point_rows):
     return result
 
 
-@app.route('/get_h_xtfx_result', methods=['GET'])
-def get_h_xtfx_result():
-    """接口：返回单元 h_xtfx                                                                结果"""
-    connection = connect_to_database()
-    if not connection:
-        return jsonify({"error": "数据库连接失败"})
-    
-    unit_rows, point_rows = fetch_unit_and_point_data(connection)
-    result = check_h_xtfx_values(unit_rows, point_rows)
-    
-    connection.close()
-    return jsonify(result)
-
 
-if __name__ == '__main__':
-    app.run(debug=True)