siwei_ai/app.py

from flask import Flask, render_template, request, jsonify
import psycopg2
from psycopg2.extras import DictCursor
import logging
import ollama
import json
import datetime
import uuid

app = Flask(__name__)


# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 连接数据库
conn = psycopg2.connect(
    dbname="real3d",
    user="postgres",
    password="postgis",
    host="192.168.100.30",
    port="5432"
)


# 后台接口
@app.route("/")
def home():
    return render_template('index.html')

# 接收消息，大模型解析


@app.route('/msg', methods=['POST'])
def inputMsg():
    # 从请求中获取JSON数据
    data = request.get_json()

    # 检查是否接收到数据
    if not data:
        return jsonify({"error": "No data received"}), 400

    # 打印接收到的消息
    print(data['msg'])

    msg = data['msg']
    # 调用大模型解析
    # 这里调用大模型，并返回解析结果
    # 生成提示信息
    # 定义输入信息变量
    # 生成提示信息

    prompt = f"""请扮演地理领域的智能选址文本提取工具,基于以下因子选择、选择范围和用地类型提取其对应的相关数据，
    因子选择：['抱坡区','天涯区','崖州区','海棠区','吉阳区' ],
    用地类型：['规划因子', '永久基本农田', '城镇开发边界内', '生态保护红线', '文化保护区', '自然保护地', '风景名胜区', '权属因子', '国有使用权', '防控因子', '河道管理线', '水库', '公益林', '地形因子', '坡度', '邻避因子', '火葬场', '垃圾处理场', '污水处理场', '高压线', '变电站', '古树', '市政设施', '交通', '城市道路', '主要出入口', '管线', '排水', '供水', '燃气', '电力', '电信', '公共服务设施', '十五分钟社区生活圈邻里中心', '社区服务设施', '零售商业场所', '医疗卫生设施', '教育场所', '幼儿园服务半径', '小学服务半径', '为老服务设施', '文化活动设施', '体育运动场所'],
    选址范围：['园地', '耕地', '林地', '草地', '湿地', '公共卫生用地', '老年人社会福利用地', '儿童社会福利用地', '残疾人社会福利用地', '其他社会福利用地', '零售商业用地', '批发市场用地', '餐饮用地', '旅馆用地', '公用设施营业网点用地', '娱乐用地', '康体用地', '一类工业用地', '二类工业用地', '广播电视设施用地', '环卫用地', '消防用地', '干渠', '水工设施用地', '其他公用设施用地', '公园绿地', '防护绿地', '广场用地', '军事设施用地', '使领馆用地', '宗教用地', '文物古迹用地', '监教场所用地', '殡葬用地', '其他特殊用地', '河流水面', '湖泊水面', '水库水面', '坑塘水面', '沟渠', '冰川及常年积雪', '渔业基础设施用海', '增养殖用海', '捕捞海域', '工业用海', '盐田用海', '固体矿产用海', '油气用海', '可再生能源用海', '海底电缆管道用海', '港口用海', '农业设施建设用地', '耕地', '园地', '林地', '工矿用地', '畜禽养殖设施建设用地', '水产养殖设施建设用地', '城镇住宅用地', '草地', '湿地', '留白用地', '陆地水域', '游憩用海', '特殊用海', '特殊用地', '其他海域', '居住用地', '绿地与开敞空间用地', '水田', '水浇地', '旱地', '果园', '茶园', '橡胶园', '其他园地', '乔木林地', '竹林地', '城镇社区服务设施用地', '农村宅基地', '农村社区服务设施用地', '机关团体用地', '科研用地', '文化用地', '教育用地', '体育用地', '医疗卫生用地', '社会福利用地', '商业用地', '商务金融用地', '二类农村宅基地', '图书与展览用地', '文化活动用地', '高等教育用地', '中等职业教育用地', '体育训练用地', '其他交通设施用地', '供水用地', '排水用地', '供电用地', '供燃气用地', '供热用地', '通信用地', '邮政用地', '医院用地', '基层医疗卫生设施用地', '田间道', '盐碱地', '沙地', '裸土地', '裸岩石砾地', '村道用地', '村庄内部道路用地', '渔业用海', '工矿通信用海', '其他土地', '公共管理与公共服务用地', '仓储用地', '交通运输用地', '公用设施用地', '交通运输用海', '航运用海', '路桥隧道用海', '风景旅游用海', '文体休闲娱乐用海', '军事用海', '其他特殊用海', '空闲地', '田坎', '港口码头用地', '管道运输用地', '城市轨道交通用地', '城镇道路用地', '交通场站用地', '一类城镇住宅用地', '二类城镇住宅用地', '三类城镇住宅用地', '一类农村宅基地', '商业服务业用地', '三类工业用地', '一类物流仓储用地', '二类物流仓储用地', '三类物流仓储用地', '盐田', '对外交通场站用地', '公共交通场站用地', '社会停车场用地', '中小学用地', '幼儿园用地', '其他教育用地', '体育场馆用地', '灌木林地', '其他林地', '天然牧草地', '人工牧草地', '其他草地', '森林沼泽', '灌丛沼泽', '沼泽草地', '其他沼泽地', '沿海滩涂', '内陆滩涂', '红树林地', '乡村道路用地', '种植设施建设用地', '娱乐康体用地', '其他商业服务业用地', '工业用地', '采矿用地', '物流仓储用地', '储备库用地', '铁路用地', '公路用地', '机场用地'],
    landType是用地类型,
    districtName是选址范围,
    area是用地大小,单位统一转换为亩
    yxyz是因子选择,公里、千米的单位转换为米，
    输入以下信息:"{msg}",请基于因子选择、选址范围和用地类型，提取其对应的相关数据，并把提取结果中的景点转换为风景名胜区、农田转换为永久基本农田等。结果以下面格式输出:
    {{"districtName":"抱坡区","landType":"居住用地","area": {{ "min": 30,"max": 50}},"factors": [{{"type": "水库","condition": "大于","value": "100"}},{{"type": "小学服务半径","condition": "小于","value": "1000"}},{{"type": "永久基本农田","condition": "不相交" }}, {{"type": "城镇开发边界内","condition": "包含"}}]}},
    """

    try:
        res = ollama.generate(
            model="qwen2:7b",
            stream=False,
            prompt=prompt,
            options={"temperature": 0},
            format="json",
            keep_alive=-1
        )
        print(res["response"])
    except Exception as e:
        print(f"生成过程中出现错误: {e}")
    res1 = res["response"].replace("大于","gt").replace("小于","lt").replace("大于等于","get").replace("小于等于","let").replace("介于","between")
    json_res = res1
    json_res = json.loads(json_res)

    # 组织成选址需要的数据格式
    json_res = jsonResToDict(json_res)
    # 返回响应
    return jsonify(json_res)


# 将大模型解析的结果转换为选址需要的数据格式

def jsonResToDict(json_res):
    # 1.查询选址范围信息
    districtName = json_res["districtName"]
    ewkt = getAiDistrict(districtName)
    # 2.保存选址范围信息
    geomId = saveGeom(ewkt)
    # 3.获取用地类型信息
    landType = json_res["landType"]
    landType = getLandType(landType, "YDYHFLDM")
    # 4.获取模板信息
    factorTemplates = getTemplateByCode(landType)

    # TODO 以哪个因子列表为准，模版和因子个数怎么匹配

    now = datetime.datetime.now()
    formatted_time = now.strftime("%Y%m%d%H%M%S")
    res = {
        "xzmj": 1500,
        "xmmc": "规划选址项目_"+formatted_time,
        "jsdw": "建设单位",
        "ydxz_bsm": landType,
        "ydmjbegin": json_res["ydmjbegin"],
        "ydmjend":  json_res["ydmjend"],
        "geomId": geomId,
        "yxyz": [],
        # TODO: 循环遍历
        # "yxyz": [
        #     {
        #         "id": "259e5bbaab434dbfb9c679bd44d4bfa4",
        #         "name": "幼儿园服务半径",
        #         "bsm": "TB_YEY",
        #         "conditionInfo": {
        #             "spatial_type": "distance",
        #             "default": "lt",
        #             "hasValue": true,
        #             "defaultValue": "300",
        #             "unit": "米",
        #             "clip": false
        #         }
        #     }
        # ],
        "useMultiple": json_res["useMultiple"],
        "useLandType": json_res["useLandType"],
        "multipleDistance": json_res["multipleDistance"]
    }
    # 循环遍历输入因子
    factors = json_res["yxyz"]
    input_factors = {}
    for factor in factors:
        factorInfo = getFactorByName(factor["name"])
        if factorInfo == None:
            continue
        factorId = factorInfo["id"]
        factorBsm = factorInfo["bsm"]
        conditionInfo = factorInfo["condition_info"]
        conditionObj = json.loads(conditionInfo)

        factor_info = {
            "id": factorId,
            "name": factor["name"],
            "bsm": factorBsm,
            "conditionInfo": {
                "spatial_type": conditionObj["spatial_type"],
                "default": factor["default"],
                "hasValue": conditionObj["hasValue"],
                "defaultValue": factor["defaultValue"],
                "unit": conditionObj["unit"],
                "clip": conditionObj["clip"]
            }
        }
        input_factors[factor_info["id"]] = factor_info
    # 循环遍历模板
    for factorTemplate in factorTemplates:
        factorId = factorTemplate["id"]
        if factorId in input_factors:
            res["yxyz"].append(input_factors[factorId])
        else:
            factorTemplate["conditionInfo"]=json.loads(factorTemplate["conditionInfo"])
            res["yxyz"].append(factorTemplate)
    return res

# 获取因子信息


def getFactorByName(name):
    with conn.cursor(cursor_factory=DictCursor) as cur:
        sql = "SELECT * FROM base.t_fzss_fzxz_factor WHERE name = %s"
        complete_sql = cur.mogrify(sql, (name,)).decode('utf-8')
        logger.info(f"Executing SQL: {complete_sql}")
        cur.execute(sql, (name,))
        res = cur.fetchone()
    return res

# 获取内置模板信息


def getTemplateByCode(code):
    with conn.cursor(cursor_factory=DictCursor) as cur:
        sql = 'SELECT factor_id as id,factor_name as name,factor_bsm as bsm,condition_info as "conditionInfo" FROM base.t_fzss_fzxz_factor_temp WHERE land_type_code = %s'
        complete_sql = cur.mogrify(sql, (code,)).decode('utf-8')
        logger.info(f"Executing SQL: {complete_sql}")
        cur.execute(sql, (code,))
        res = cur.fetchall()
        # 将查询结果转换为字典列表
        result_list = [dict(row) for row in res]
    return result_list

# 获取选址范围信息


def getAiDistrict(name):
    with conn.cursor(cursor_factory=DictCursor) as cur:
        sql = "SELECT public.st_asewkt(geom) as geom FROM base.t_fzss_fzxz_ai_district WHERE name = %s"
        complete_sql = cur.mogrify(sql, (name,)).decode('utf-8')
        logger.info(f"Executing SQL: {complete_sql}")
        cur.execute(sql, (name,))
        res = cur.fetchone()
    return res["geom"]


# 保存选址范围信息
def saveGeom(ewkt):
    new_uuid = str(uuid.uuid4())  # 生成一个新的 UUID
    from_type = 3
    with conn.cursor() as cur:
        sql = "INSERT INTO base.t_fzss_zhxz_file(id,geom,from_type,create_time,area) VALUES (%s,public.st_geomfromewkt(%s),%s,now(),public.st_area(public.st_geomfromewkt(%s)::public.geography))"
        complete_sql = cur.mogrify(
            sql, (new_uuid, ewkt, from_type, ewkt)).decode('utf-8')
        logger.info(f"Executing SQL: {complete_sql}")
        cur.execute(sql, (new_uuid, ewkt, from_type, ewkt))
        conn.commit()
    return new_uuid

# 获取用地类型信息


def getLandType(landName, fzbs):
    with conn.cursor(cursor_factory=DictCursor) as cur:
        sql = "SELECT dm,mc,fzbs FROM base.t_fzss_fzxz_dict WHERE mc = %s and fzbs=%s"
        complete_sql = cur.mogrify(sql, (landName, fzbs)).decode('utf-8')
        logger.info(f"Executing SQL: {complete_sql}")
        cur.execute(sql, (landName, fzbs))
        res = cur.fetchone()
    return res["dm"]


# getTemplateByCode("08")

# getAiDistrict("抱坡区")

# ewkt="SRID=4326;POLYGON ((109.568515723151 18.2729002407864, 109.564270326708 18.2607742953866, 109.580087492139 18.2571512198688, 109.588461804591 18.2570597503377, 109.58884305979 18.2645363088176, 109.582107142538 18.2732736518031, 109.568515723151 18.2729002407864))"
# saveGeom(ewkt)

# getFactorByName("幼儿园服务半径")
if __name__ == '__main__':
    # app.run()
    app.run(host='0.0.0.0')