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

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']
    # 调用大模型解析
    # 这里调用大模型，并返回解析结果
    # 生成提示信息
    # 定义输入信息变量


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

    # 生成提示信息
    
    prompt = f"""请扮演文本提取工具,基于以下因子选择和用地类型提取其对应的相关数据,并提取其距离的相关数据，
    因子选择：{str(factor_selection)},
    用地类型：{str(land_types)},
    landType是用地类型,
    ydmjbegin是用地面积的开始值,是较小的那个数,ydmjend是用地面积的结束值,是较大的那个数,单位统一转换为亩
    yxyz中的name是因子选择,defalt是条件,defaultvalue是距离,公里、千米的单位转换为米，
    请把这句话:"{msg}",基于因子选择和用地类型提取其对应的相关数据,并提取其距离的相关数据。转换为以下格式:
    "landType":"林地","ydmjbegin":10,"ydmjend":100,"yxyz":["name":"幼儿园服务半径","default":"大于","defaultValue":"300"],"useMultiple":false,"useLandType":true,"multipleDistance":0
    并把yxyz中大于、小于、大于等于、小于等于、介于分别替换为gt、lt、get、let、between,然后把结果进行输出"""

    try:
        res = ollama.generate(
            model="qwen2:7b",
            stream=False,
            prompt=prompt,
            options={"temperature": 0},
            format="json"
        )
        print(res["response"])
    except Exception as e:
        print(f"生成过程中出现错误: {e}")

    json_res = res["response"]


    json_res = json.loads(json_res)

    # 返回响应
    return jsonify(json_res)

# 获取因子信息
def getFactorByName(name):
    with conn.cursor(cursor_factory=DictCursor) as cur:
        sql = "SELECT id,bsm 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.fetchall()
        res = cur.fetchone()
        id=str(res['id'])
    return res

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