CSPON-TSC/landsite_agent/gistools/geo_analysis.py

import json
from shapely import wkt
from database import Database
from typing import Any, List, Dict

database = Database()

async def _intersect_query(
    geom: str,
    distance: float,
    sql_template: str,
    srid: int = 4490
) -> List[Dict[str, Any]]:
    """
    通用的空间相交查询函数。

    :param geom: WKT格式的几何对象字符串
    :param distance: 缓冲区距离（米）
    :param sql_template: SQL模板，需包含 {bufWkt} 占位符
    :param srid: 空间参考ID，默认为4490
    :return: 查询结果列表
    """
    geometry = wkt.loads(geom)
    buf = geometry.buffer(distance / 111194)
    bufWkt = buf.wkt
    if not bufWkt.startswith(f'SRID={srid}'):
        bufWkt = f'SRID={srid};' + bufWkt
    sql = sql_template.format(bufWkt=bufWkt)
    # logging.debug(sql)
    dataList = await database.execute_query(sql)
    # logging.debug(dataList)
    return dataList

async def intersect_kfq(geom: str) -> List[Dict[str, Any]]:
    """
    查询缓冲区内的开发区信息
    """
    sql_template = (
        "select kfqmc as name, gljgdz as address, zdcy as primaryIndustry, "
        "kfqjb1 as kfqjb, st_asgeojson(shape) as geom "
        "from sde.kfq where st_intersects(shape, '{bufWkt}')"
    )
    return await _intersect_query(geom, distance=5000, sql_template=sql_template)

async def intersect_gyyd(geom: str) -> List[Dict[str, Any]]:
    """
    查询缓冲区内的工业用地信息
    """
    sql_template = (
        "select tdqlr as name, tdzl as address, st_asgeojson(shape) as geom "
        "from sde.zjgyydc where st_intersects(shape, '{bufWkt}')"
    )
    return await _intersect_query(geom, distance=1000, sql_template=sql_template)