CSPON-TSC/landsite_agent/xuanzhi_query.py

import json
from fastapi import APIRouter, Query
from typing import List, Any

from gistools.geo_analysis import intersect_kfq, intersect_gyyd
from database import Database

router = APIRouter()
database = Database()

@router.get("/kg-query")
async def kg_query(id: str = Query(..., description="ID列表, 逗号分隔")) -> List[Any]:
    """
    查询控制性详细规划信息
    """
    sql = (
        "select id, xzqmc, xzqdm, dymc, yddm, ydxz, ydmj, rjlsx, rjlxx, jzmdsx, jzmdxx, "
        "jzgdsx, jzgdxx, ldlsx, ldlxx, pfwh, pfsj,dkcsd,ghfhd,dkgzd,jtblx,cyppd, st_area(shape::geography) as pfmarea, "
        "st_astext(shape) as geom, st_astext(st_centroid(shape)) as center_wkt "
        "from sde.kzxxxgh where id in ({id})"
    )
    result = await database.execute_query(sql.format(id=id))
    return result

@router.get("/klyzy-query")
async def klyzy_query(id: str = Query(..., description="ID列表, 逗号分隔")) -> List[Any]:
    """
    查询可利用资源信息
    """
    sql = (
        "select *, st_astext(shape) as geom, st_astext(st_centroid(shape)) as center_wkt "
        "from sde.ecgap_klyzy where id in ({id})"
    )
    result = await database.execute_query(sql.format(id=id))
    return result

@router.get("/population")
async def population_query(name: str = Query(..., description="地名")) -> List[Any]:
    """
    查询人口信息
    """
    sql = (
        "select pop_age_o as 老年人口, pop_age_m as 中青年人口, pop_sex_m as 男性人口, "
        "pop_sex_f as 女性人口, pop as 人口总量 "
        "from sde.layer_poi_population where name = '{name}'"
    )
    result = await database.execute_query(sql.format(name=name))
    return result

@router.get("/kfq-intersect")
async def kfq_intersect(wkt: str = Query(..., description="WKT字符串")) -> List[Any]:
    """
    查询缓冲区内的开发区信息
    """
    result = await intersect_kfq(wkt)
    return result

@router.get("/gyyd-intersect")
async def gyyd_intersect(wkt: str = Query(..., description="WKT字符串")) -> List[Any]:
    """
    查询缓冲区内的工业用地信息
    """
    result = await intersect_gyyd(wkt)
    return result