ai调用excel整理板厚,零件,预计板耗信息保存为json

把llm对话开放到内网-CSDN博客

sse mcp flask 开放mcp服务到内网-CSDN博客

#get_thick_part_json.py
import json
import re
import requests
import os
import logging
from flask import Flask, request, jsonify# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)app = Flask(__name__)@app.route('/process_excel', methods=['GET'])
def process_excel_api():"""处理Excel文件并发送给AI服务的API接口---parameters:- name: filepathin: querytype: stringrequired: truedescription: Excel文件路径- name: sheet_namein: querytype: stringrequired: falsedefault: 零件清单description: 工作表名称responses:200:description: 成功处理400:description: 请求参数错误500:description: 服务器内部错误"""# 获取查询参数filepath = request.args.get('filepath')sheet_name = request.args.get('sheet_name', '零件清单')# 检查必需参数if not filepath:return jsonify({"error": "缺少必需参数: filepath"}), 400try:# 第一步：从Excel获取数据excel_url = "http://localhost:5000/read_excel"# 对URL参数进行编码params = {"filepath": filepath,"sheet_name": sheet_name}# 获取Excel数据excel_response = requests.get(excel_url, params=params)excel_response.raise_for_status()excel_data = excel_response.json()# 提取板厚和构件信息extracted_info = excel_data# 在获取excel_memory之前，先从环境变量或配置中获取该信息excel_memory = os.environ.get('excel_memory', '')ai_message_content = f"""请从以下Excel数据中提取板厚,编号,数量和构件信息，以及需要平方米的板,并按照指定格式返回JSON，使用[START_JSON]和[END_JSON]标记框起JSON部分。特别注意：请将零件按照厚度进行分组。关于材料的一些记忆信息：{excel_memory}示例格式：[START_JSON]{{"thickness_groups": [{{"thickness": "10mm","area_required": 2.5,"components": [{{"part_name": "零件A","quantity": 5,"thickness": "10mm"}}]}}]}}[END_JSON]Excel数据：{json.dumps(extracted_info, ensure_ascii=False)}"""# 第二步：使用POST请求发送数据给AI服务，避免URL过长问题ai_url = "http://localhost:5003/chat"ai_payload = {"messages": [{"role": "user","content": ai_message_content}],"tools":  None}# 发送POST请求到AI服务ai_response = requests.post(ai_url, json=ai_payload)ai_response.raise_for_status()# 解析AI响应ai_response_data = ai_response.json()ai_content = ai_response_data['choices'][0]['message']['content']# 从AI响应中提取被标记框起的JSONjson_pattern = r'\[START_JSON\]\s*(.*?)\s*\[END_JSON\]'match = re.search(json_pattern, ai_content, re.DOTALL)structured_data = Noneif match:try:json_content = match.group(1).strip()logger.info(f"提取到的JSON内容: {json_content}")# 将提取的JSON字符串解析为Python对象structured_data = json.loads(json_content)logger.info("JSON解析成功")# 保存解析出的JSON到文件save_path = os.path.join(os.getcwd(), 'ai_parsed_result.json')with open(save_path, 'w', encoding='utf-8') as f:json.dump(structured_data, f, ensure_ascii=False, indent=2)logger.info(f"结果已保存至: {save_path}")except json.JSONDecodeError as json_err:logger.error(f"JSON解析失败: {json_err}")logger.error(f"尝试解析的内容: {json_content}")passexcept Exception as save_err:logger.error(f"保存文件时出错: {save_err}")passelse:logger.warning("未在AI响应中找到JSON标记")# 返回处理后的数据return jsonify({"status": "success",# "excel_data": excel_data,"structured_data": structured_data  # 已解析的结构化数据})except requests.exceptions.RequestException as e:logger.error(f"请求出错: {str(e)}")return jsonify({"error": f"请求出错: {str(e)}"}), 500except json.JSONDecodeError as e:logger.error(f"JSON解析出错: {str(e)}")return jsonify({"error": f"JSON解析出错: {str(e)}"}), 500except Exception as e:logger.error(f"处理过程中出错: {str(e)}")return jsonify({"error": f"处理过程中出错: {str(e)}"}), [e-mail]('/health', methods=['GET'])
def health_check():"""健康检查接口"""return jsonify({"status": "healthy"})if __name__ == "__main__":app.run(host='0.0.0.0', port=5001, debug=True)