# -*- encoding:utf-8 -*- import json import os import re import time from flask_app.general.file2markdown import convert_pdf_to_markdown from flask_app.general.多线程提问 import multi_threading from flask_app.general.通义千问long import qianwen_long, upload_file from flask_app.general.json_utils import clean_json_string, combine_json_results from flask_app.货物标.截取pdf货物标版 import truncate_pdf_main from flask_app.general.doubao import doubao_model, generate_full_user_query, pdf2txt def generate_key_paths(data, parent_key='', good_list=None, seen=None): """ 生成嵌套字典中的键路径，并提取最内层的键名。 同时，提取特定模式的键（如 '交换机-1', '交换机-2'）的父路径。 参数: data (dict): 输入的字典数据 parent_key (str): 上级键路径，用于递归调用 good_list (list): 用于存储去重后的最内层键名 seen (set): 用于跟踪已添加到 good_list 的元素 返回: tuple: 包含键路径列表、最内层键名列表、分组路径列表以及 no_keys_added 的元组 (key_paths, good_list, grouped_paths, no_keys_added) """ if good_list is None: good_list = [] if seen is None: seen = set() key_paths = [] grouped_paths = set() # 使用集合避免重复路径 no_keys_added = True # 默认假设没有添加任何键 for key, value in data.items(): # 构建当前的键路径 current_key = f"{parent_key}.{key}" if parent_key else key if isinstance(value, dict): if value: # 递归调用，并获取子路径、子 good_list、子分组路径以及子 no_keys_added sub_key_paths, _, sub_grouped_paths, sub_no_keys_added = generate_key_paths( value, current_key, good_list, seen ) key_paths.extend(sub_key_paths) grouped_paths.update(sub_grouped_paths) # 合并子分组路径到当前分组路径 # 更新 no_keys_added no_keys_added = no_keys_added and sub_no_keys_added else: # 空字典视为叶子节点 clean_key = key.replace(" ", "") # 使用正则提取具有相同前缀的键 match = re.match(r'(.+)-\d+$', clean_key) if match: goods_name = match.group(1) # 提取货物名称前缀 # 构建分组路径 goods_path = f"{parent_key}.{goods_name}" if parent_key else goods_name goods_path = goods_path.replace(" ", "") grouped_paths.add(goods_path) # 添加到集合中，不添加到key_paths # **新增部分**：将匹配的键添加到 key_paths 和 good_list key_paths.append(current_key.replace(" ", "")) if goods_name not in seen: good_list.append(goods_name) seen.add(goods_name) else: # 非匹配项则直接添加到key_paths中 key_paths.append(current_key.replace(" ", "")) if clean_key not in seen: good_list.append(clean_key) # 去掉空格后添加 seen.add(clean_key) # 更新 no_keys_added no_keys_added = False elif isinstance(value, list): if value: # 非空列表视为叶子节点 key_paths.append(current_key.replace(" ", "")) clean_key = key.replace(" ", "") if clean_key not in seen: good_list.append(clean_key) # 去掉空格后添加 seen.add(clean_key) # 更新 no_keys_added no_keys_added = False else: # 空列表也视为叶子节点（根据需求可以调整） key_paths.append(current_key.replace(" ", "")) clean_key = key.replace(" ", "") if clean_key not in seen: good_list.append(clean_key) # 去掉空格后添加 seen.add(clean_key) # 更新 no_keys_added no_keys_added = False elif value in {"未知", "", "/"}: # 特定值视为叶子节点 key_paths.append(current_key.replace(" ", "")) clean_key = key.replace(" ", "") if clean_key not in seen: good_list.append(clean_key) # 去掉空格后添加 seen.add(clean_key) # 更新 no_keys_added no_keys_added = False else: # 其他情况视为叶子节点 key_paths.append(current_key.replace(" ", "")) clean_key = key.replace(" ", "") if clean_key not in seen: good_list.append(clean_key) # 去掉空格后添加 seen.add(clean_key) # 更新 no_keys_added no_keys_added = False return key_paths, good_list, grouped_paths, no_keys_added # xx_paths: 形如xx.xx，将嵌套键值对用'.'表示 def combine_and_update_results(original_data, updates): def normalize_key(key): """ 规范化键名： - 替换全角点号为半角点号。 - 删除所有空格（包括半角和全角）。 """ # 替换全角点号（．、。）为半角点号（.） key = key.replace('．', '.').replace('。', '.') # 删除所有空格（半角空格和全角空格） key = key.replace(' ', '').replace('\u3000', '') return key def normalize_original_data(d): """ 递归规范化原始数据字典的键。 """ if not isinstance(d, dict): return d normalized = {} for k, v in d.items(): nk = normalize_key(k) normalized[nk] = normalize_original_data(v) return normalized def normalize_update_value(value): """ 递归规范化更新字典中嵌套的字典的键。 """ if isinstance(value, dict): return {normalize_key(k): normalize_update_value(v) for k, v in value.items()} else: return value def recursive_update(data, key, value): """ 递归更新嵌套字典。 """ keys = key.split('.') for k in keys[:-1]: data = data.setdefault(k, {}) if isinstance(value, dict) and isinstance(data.get(keys[-1], None), dict): data[keys[-1]] = {**data.get(keys[-1], {}), **value} else: data[keys[-1]] = value # 1. 规范化原始数据字典的键 original_data = normalize_original_data(original_data) # 2. 规范化更新字典的键 normalized_updates = {} for key, value in updates.items(): nk = normalize_key(key) nv = normalize_update_value(value) normalized_updates[nk] = nv # 3. 执行递归更新 for key, value in normalized_updates.items(): recursive_update(original_data, key, value) return original_data def postprocess(data): """递归地转换字典中的值为列表，如果所有键对应的值都是'/', '{}' 或 '未知'""" def convert_dict(value): # 如果所有值是'/', '{}' 或 '未知' if all(v in ['/', '未知', {}] for v in value.values()): return list(value.keys()) else: # 如果不满足条件，则递归处理嵌套的字典 return {k: convert_dict(v) if isinstance(v, dict) else v for k, v in value.items()} # 递归处理顶层数据 return {key: convert_dict(val) if isinstance(val, dict) else val for key, val in data.items()} # user_query1 = """ # 请你首先定位该采购文件中的采购清单或采购需求部分，请告诉我需要采购的货物，如果有采购清单，请直接根据清单上的货物（或系统）名称给出结果，注意不要返回'说明'或'规格'或'技术参数'列中的内容；若没有采购清单，你要从表格中或文中摘取需要采购的系统和货物，采购需求中可能包含层次关系，例如采购的某系统中可能包含几种货物，那么你需要用嵌套键值对表示这种关系，且不要遗漏该系统中包含的货物，你的输出请以json格式返回，最外层键名为'采购需求'，嵌套键名为对应的系统名称或货物名称，需与原文保持一致，无需给出采购数量和单位。以下为需要考虑的特殊情况：如果采购清单中同一层级(或同一系统)下存在同名货物且它们的采购要求有所不同，请你以'货物名-编号'区分多种型号，编号为从 1 开始的自然数，依次递增,例如若采购清单中有两种型号的'交换机'，那么你应返回两个键名，'交换机-1'和'交换机-2'；如有未知内容，在对应键值处填'未知'。以下为考虑了特殊情况的示例输出： # { # "采购需求": { # "交换机-1"：{}, # "交换机-2":{}, # "门禁管理系统": {}, # "交通监控视频子系统": { # "高清视频抓拍像机":{}, # "补光灯":{} # }, # "LED全彩显示屏": {} # } # } # """ #文件内容以markdown格式组织，其中表格部分（若有）以html语法组织， def get_technical_requirements(file_path,invalid_path): file_id=upload_file(file_path) first_query_template="该文件是否说明了采购需求,即需要采购哪些货物?如果有,请回答'是',否则,回答'否'" # first_query=generate_full_user_query(file_path,first_query_template) # judge_res=doubao_model(first_query) judge_res=qianwen_long(file_id,first_query_template) prompt_template1 = ''' 任务：解析采购文件，提取采购需求，并以JSON格式返回。 要求与指南： 1. 精准定位：请运用文档理解能力，找到文件中的采购需求部分，若有采购清单，请直接根据采购清单上的货物（或系统）名称给出结果。 2. 采购目标：采购目标通常有硬件（如设备、货物）和软件（如系统软件、应用APP），一次采购活动可能同时包含这两种类型。对于工程类的施工、建设采购需求，无需提取。 3. 非清单形式处理：若未出现采购清单，则从表格或文字中摘取采购信息。 4. 系统归属：一些采购活动可能将采购目标划分为若干系统和货物，每个系统可能包含若干货物，则将这些货物名称作为该系统的二级键；系统可以只包含总体'系统功能'而无货物。 5. 软件需求：对于软件应用或系统软件需求，仅需列出系统模块构成（若有），并作为系统键值的一部分，无需在模块下再细分功能。 6. 系统功能：若采购的某系统提及总体系统功能，则在系统值中添加'系统功能'二级键，不展开具体内容。 7. 完整性：确保不遗漏系统内的货物，也不添加未提及的内容。 输出格式： 1.JSON格式，最外层键名为'采购需求'。 2.层次关系用嵌套键值对表示。 3.嵌套键名为系统或货物或模块名称，与原文保持一致。 4.最内层键值应为空对象（{{}}）。 5.不包含'说明'、'规格'、'技术参数'等列内容，仅返回采购的货物或系统或模块名称。 特殊情况处理： 同一层级（如同一系统中）下同名但采购要求不同的货物，以'货物名-编号'区分，编号从1递增。 示例输出1，普通系统、货物类采购： {{ "采购需求": {{ "交换机-1": {{}}, "交换机-2": {{}}, "门禁管理系统": {{ "系统功能":{{}} }}, "交通监控视频子系统": {{ "系统功能": {{}}, "高清视频抓拍像机": {{}}, "补光灯": {{}} }}, "LED全彩显示屏": {{}} // 其他系统和货物 }} }} 示例输出2，系统软件采购： {{ "采购需求": {{ "信息管理系统": {{ "通用模块":{{}}, "用户管理":{{}} }}, "信息检索系统": {{ "系统功能":{{}}, "权限管理模块":{{}} }}, "XX小程序":{{}}, "数据分析中心":{{}} }} }} 注意事项： 1.严格按照上述要求执行，确保输出准确性和规范性。 2.如有任何疑问或不确定内容，请保留原文描述，必要时使用'未知'标注。 ''' prompt_template2 = ''' 任务：解析采购文件，提取采购需求，并以JSON格式返回。 要求与指南： 1. 精准定位：请运用文档理解能力，找到文件中的采购需求部分，若有采购清单，请直接根据采购清单上的货物（或系统）名称给出结果。 2. 采购目标：采购目标通常有硬件（如设备、货物）和软件（如系统软件、应用APP），一次采购活动可能同时包含这两种类型。对于工程类的施工、建设采购需求，无需提取。 3. 非清单形式处理：若未出现采购清单，则从表格或文字中摘取采购信息。 4. 系统归属：一些采购活动可能将采购目标划分为若干系统和货物，每个系统可能包含若干货物，则将这些货物名称作为该系统的二级键；系统可以只包含总体'系统功能'而无货物。 5. 软件需求：对于软件应用或系统软件需求，仅需列出系统模块构成（若有），并作为系统键值的一部分，无需在模块下再细分功能。 6. 系统功能：若采购的某系统提及总体系统功能，则在系统值中添加'系统功能'二级键，不展开具体内容。 7. 完整性：确保不遗漏系统内的货物，也不添加未提及的内容。 输出格式： 1.JSON格式，最外层键名为'采购需求'。 2.层次关系用嵌套键值对表示。 3.嵌套键名为系统或货物或模块名称，与原文保持一致。 4.最内层键值应为空对象（{{}}）。 5.不包含'说明'、'规格'、'技术参数'等列内容，仅返回采购的货物或系统或模块名称。 特殊情况处理： 若同一层级（如同一系统中）下存在同名但采购要求不同的货物，以'货物名-编号'区分，编号从1递增，规避重复键名的问题；否则无需添加编号。 示例输出1，普通系统、货物类采购： {{ "采购需求": {{ "交换机-1": {{}}, "交换机-2": {{}}, "门禁管理系统": {{ "系统功能":{{}} }}, "交通监控视频子系统": {{ "系统功能": {{}}, "高清视频抓拍像机": {{}}, "补光灯": {{}} }}, "LED全彩显示屏": {{}} // 其他系统和货物 }} }} 示例输出2，系统软件采购： {{ "采购需求": {{ "信息管理系统": {{ "通用模块":{{}}, "用户管理":{{}} }}, "信息检索系统": {{ "系统功能":{{}}, "权限管理模块":{{}} }}, "XX小程序":{{}}, "数据分析中心":{{}} }} }} 文件内容：{full_text} 注意事项： 1.严格按照上述要求执行，确保输出准确性和规范性。 2.如有任何疑问或不确定内容，请保留原文描述，必要时使用'未知'标注。 ''' if '否' in judge_res: file_id=upload_file(invalid_path) print("调用invalid_path") model_res=qianwen_long(file_id,prompt_template1) print(model_res) else: # processed_filepath = convert_pdf_to_markdown(file_path) # 转markdown格式 # processed_filepath=r"C:\Users\Administrator\Desktop\货物标\extract_files\107国道.txt" processed_filepath = pdf2txt(file_path) # 纯文本提取 user_query=generate_full_user_query(processed_filepath,prompt_template2) model_res=doubao_model(user_query) # model_res = qianwen_long(file_id,prompt_template1) print(model_res) cleaned_res = clean_json_string(model_res) #转字典 normal_paths,good_list,grouped_paths,no_keys_added= generate_key_paths(cleaned_res['采购需求']) # 提取需要采购的货物清单 key_list：交通监控视频子系统.高清视频抓拍像机 ... grouped_paths是同一系统下同时有'交换机-1'和'交换机-2'，提取'交换机' ，输出eg:{'交通标志.标志牌铝板', '交通信号灯.交换机'} if no_keys_added: final_res = postprocess(cleaned_res) else: # user_query_template = "请你根据该货物标中采购要求部分的内容，请你给出\"{}\"的技术参数（或采购要求），请以json格式返回结果，外层键名为\"{}\", 键值对中的键是你对该要求的总结，而值需要完全与原文保持一致，不可擅自总结删减。" user_query_template = """ 请你根据该货物标中采购要求部分的内容，请你给出\"{}\"的技术参数（或采购要求），请以json格式返回结果，键名为\"{}\", 键值为一个列表，列表中包含若干描述\"{}\"的技术参数（或采购要求）的字符串，需与原文完全一致，即若技术参数前存在序号也要保留，但你不可擅自增添或删减。以下为需要考虑的特殊情况：如果该货物没有相关采购要求或技术参数要求，键值为空列表。示例输出格式如下： {{ "摄像机控制键盘": [ "1、支持串行 RS232/RS422 和 IP 混合控制，允许在一个控制器上使用 RS232/RS422/IP 控制单个系统中的摄像机；", "2、支持 2 组 RS422 串口 VISCA 协议菊花链控制 2x7 台摄像机。" ] }} """ user_query_template_two="""请你根据该货物标中采购要求部分的内容，请你给出\"{}\"的技术参数（或采购要求），由于该货物存在多种不同的采购要求或技术参数，请你请逐一列出，请以json格式返回结果，请你以'货物名-编号'区分多种型号，编号为从 1 开始的自然数，依次递增，即第一个键名为\"{}-1\", 键值为一个列表，列表中包含若干描述\"{}\"的技术参数（或采购要求）的字符串，需与原文完全一致，即若技术参数前存在序号也要保留，但你不可擅自增添或删减。示例输出格式如下： {{ "交换机-1": [ "1、支持固化千兆电口≥8 个，固化千兆光口≥2 个，桌面型设备;", "2、支持静态链路聚合" ] "交换机-2":[ "1、交换容量≥52Gbps，包转发率≥38.69Mpps，", "2、提供国家强制性产品认证证书及测试报告（3C）" ] }} """ queries = [] for key in normal_paths: # 将键中的 '.' 替换为 '下的' modified_key = key.replace('.', '下的') # 使用修改后的键填充第一个占位符，原始键填充第二个占位符 new_query = user_query_template.format(modified_key, key, modified_key) queries.append(new_query) # 处理 grouped_paths 中的项，应用 user_query_template_two for grouped_key in grouped_paths: # 将键中的 '.' 替换为 '下的' modified_grouped_key = grouped_key.replace('.', '下的') # 使用修改后的键填充第一个占位符，原始键填充第二个占位符 new_query = user_query_template_two.format(modified_grouped_key, grouped_key, modified_grouped_key) queries.append(new_query) results = multi_threading(queries, "", file_id, 2) technical_requirements = [] if not results: print("errror!未获得大模型的回答！") else: # 打印结果 for question, response in results: technical_requirements.append(response) technical_requirements_combined_res = combine_json_results(technical_requirements) """根据所有键是否已添加处理技术要求""" # 更新原始采购需求字典 final_res=combine_and_update_results(cleaned_res['采购需求'], technical_requirements_combined_res) # final_res = postprocess(cleaned_res) final_res["货物列表"] = good_list # 输出最终的 JSON 字符串 return {"采购需求": final_res} def test_all_files_in_folder(input_folder, output_folder): # 确保输出文件夹存在 if not os.path.exists(output_folder): os.makedirs(output_folder) # 遍历指定文件夹中的所有文件 for filename in os.listdir(input_folder): file_path = os.path.join(input_folder, filename) # 检查是否是文件 if os.path.isfile(file_path): print(f"处理文件: {file_path}") # 调用函数处理文件 try: json_result = get_technical_requirements(file_path) # 定义输出文件的路径 output_file_path = os.path.join(output_folder, os.path.splitext(filename)[0] + '.json') # 保存JSON结果到文件 with open(output_file_path, 'w', encoding='utf-8') as json_file: json.dump(json_result, json_file, ensure_ascii=False, indent=4) print(f"结果已保存到: {output_file_path}") except Exception as e: print(f"处理文件 {file_path} 时出错: {e}") #TODO:提示词限制序号生成 交换机-1 #TODO:对于软件系统功能，提取偏离表的时候额外处理 包头市公安支队机动车查验监管系统招标文201907_procurement.pdf #TODO: 提示词限制最多两层防止，嵌套过多，代码也要做处理，'系统'下的最多只有一层 eg 妇幼、陕西 # { # "采购需求": { # "查验视频监管软件系统": { # "视频设备管理": { # "查验通道初始化": {}, # "视频设备绑定": {} # }, # "业务视频应用": { # "视频回放": {} # }, #{ 对于这种误判-1，脚本中也要处理一下。 # "采购需求": { # "高清数字枪机-1": {}, # "枪机支架-1": {}, # "高清数字半球机-1": {}, # "网络硬盘录像机-1": {}, # "监控硬盘-1": {}, # "交换机-1": {}, # "交换机-2": {}, # "监视器-1": {}, # "电源线-1": {}, # "网线-1": {}, # "水晶头-1": {}, # "PVC线槽-1": {}, # "辅料-1": {}, # "安装调试-1": {} # } # } if __name__ == "__main__": start_time=time.time() # truncate_file="C:\\Users\\Administrator\\Desktop\\fsdownload\\469d2aee-9024-4993-896e-2ac7322d41b7\\ztbfile_procurement.docx" truncate_file=r"C:\Users\Administrator\Desktop\new招标文件\货物标\广水市公安局音视频监控系统设备采购项目.pdf" # invalid_path="D:\\flask_project\\flask_app\\static\\output\\output1\\e7dda5cb-10ba-47a8-b989-d2993d34bb89\\ztbfile.pdf" # truncate_file="D:\\flask_project\\flask_app\\static\\output\\output1\\e7dda5cb-10ba-47a8-b989-d2993d34bb89\\ztbfile_procurement.docx" # output_folder="C:\\Users\\Administrator\\Desktop\\货物标\\output1\\tmp" # file_id = upload_file(truncate_file) invalid_path="C:\\Users\\Administrator\\Desktop\\fsdownload\\a110ed59-00e8-47ec-873a-bd4579a6e628\\ztbfile.pdf" # file_id=upload_file(truncate_file) res=get_technical_requirements(truncate_file,invalid_path) json_string = json.dumps(res, ensure_ascii=False, indent=4) print(json_string) # # input_folder = "C:\\Users\\Administrator\\Desktop\\货物标\\output1" # # output_folder = "C:\\Users\\Administrator\\Desktop\\货物标\\output3" # # test_all_files_in_folder(input_folder, output_folder) end_time=time.time() print("耗时:"+str(end_time-start_time))