zbparse/flask_app/货物标/技术参数要求提取.py

498 lines
27 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# -*- 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
from flask_app.货物标.技术参数要求提取后处理函数 import postprocess, all_postprocess
def generate_key_paths(data, parent_key='', good_list=None, seen=None):
"""
生成嵌套字典中的键路径,并提取最内层的键名。
同时,提取特定模式的键(如 '交换机-1', '交换机-2')的父路径。
如果同一层级下只有'交换机-1'但没有'交换机-2',则视为错误输入,删除后缀'-1',将'交换机'加入key_paths。
参数:
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 # 默认假设没有添加任何键
# Step 1: Collect keys that match the pattern
pattern = re.compile(r'(.+)-\d+$') # 匹配形如 '交换机-1', '交换机-2' 的键
prefix_groups = {}
other_keys = []
for key in data:
clean_key = key.replace(" ", "")
match = pattern.match(clean_key)
if match:
prefix = match.group(1)
if prefix not in prefix_groups:
prefix_groups[prefix] = []
prefix_groups[prefix].append(key)
else:
other_keys.append(key)
# Step 2: Handle grouped keys
for prefix, keys in prefix_groups.items():
current_prefix_path = f"{parent_key}.{prefix}" if parent_key else prefix
if len(keys) > 1:
# 多个键匹配同一前缀:添加到 grouped_paths
grouped_paths.add(current_prefix_path)
if prefix not in seen:
good_list.append(prefix)
seen.add(prefix)
no_keys_added = False
else:
# 只有一个键匹配:删除后缀并添加到 key_paths
key = keys[0]
key_path = current_prefix_path # 去掉后缀后,路径为父路径 + 前缀
key_paths.append(key_path)
if prefix not in seen:
good_list.append(prefix)
seen.add(prefix)
no_keys_added = False
# Step 3: Handle other keys
for key in other_keys:
value = data[key]
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(" ", "")
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):
# 列表类型视为叶子节点,无论是否为空
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
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
# 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:
ffinal_res = postprocess(cleaned_res)
else:
# user_query_template = "请你根据该货物标中采购要求部分的内容,请你给出\"{}\"的技术参数或采购要求请以json格式返回结果外层键名为\"{}\", 键值对中的键是你对该要求的总结,而值需要完全与原文保持一致,不可擅自总结删减。"
user_query_template = """请根据以下货物标中采购要求部分的内容,为\"{}\"生成技术参数(或采购要求)。请以 JSON 格式返回结果,外层键名为\"{}\",最内层键值为一个列表,列表中包含若干描述\"{}\"的技术参数(或采购要求)的字符串,内容需与原文一致,即若技术参数前存在序号也要保留,但不可擅自增添或删减。
要求与指南:
1. 如果该货物没有相关采购要求或技术参数要求,键值应为空列表。
2. 如果存在嵌套结构且原文为Markdown 的表格语法,如'摄像机|有效像素|≥900W像素' 请不要返回该Markdown语法而是使用冒号':'将相关信息拼接在一起,生成一条完整且清晰的技术参数(或采购要求)描述,作为列表中的一个字符串。如"摄像机有效像素≥900W像素"
3. 字符串中的内容为具体的技术参数要求或采购要求,请不要返回诸如'1高清录像功能'这种标题性质且不能体现要求的内容。
### 示例输出1如下
{{
"摄像机控制键盘": [
"1、支持串行 RS232/RS422 和 IP 混合控制,允许在一个控制器上使用 RS232/RS422/IP 控制单个系统中的摄像机;",
"2、支持 2 组 RS422 串口 VISCA 协议菊花链控制 2x7 台摄像机。"
]
}}
### 示例输出2如下
{{
"摄像机": [
"摄像机有效像素≥900W像素",
"摄像机最低照度彩色≤0.001lx",
"协议API 接口开放:具备;支持标准 ONVIF 协议与第三方厂家设备进行互联;支持 GB/T28181应提供 SDK"
]
}}
"""
user_query_template_two="""请根据以下货物标中采购要求部分的内容,为\"{}\"生成技术参数(或采购要求)。由于该货物存在多种不同的采购要求或技术参数,请逐一列出,并以 JSON 格式返回结果。请以'货物名-编号'区分多种型号,编号为从 1 开始的自然数,依次递增,即第一个键名为\"{}-1\", 键值为一个列表,列表中包含若干描述\"{}\"的技术参数(或采购要求)的字符串,需与原文完全一致,即若技术参数前存在序号也要保留,但不可擅自增添或删减。
请注意以下特殊情况:
要求与指南:
1. 如果该货物没有相关采购要求或技术参数要求,键值应为空列表。
2. 如果存在嵌套结构且原文为Markdown 的表格语法,如'摄像机|有效像素|≥900W像素' 请不要返回该Markdown语法而是使用冒号':'将相关信息拼接在一起,生成一条完整且清晰的技术参数(或采购要求)描述,作为列表中的一个字符串。如"摄像机有效像素≥900W像素"
3. 字符串中的内容为具体的技术参数要求或采购要求,请不要返回诸如'1高清录像功能'这种标题性质且不能体现要求的内容。
### 示例输出参考1如下
{{
"交换机-1": [
"1、支持固化千兆电口≥8 个固化千兆光口≥2 个,桌面型设备;",
"2、支持静态链路聚合"
],
"交换机-2": [
"1、交换容量≥52Gbps包转发率≥38.69Mpps",
"2、提供国家强制性产品认证证书及测试报告3C"
]
}}
### 示例输出参考2如下包含嵌套结构
{{
"摄像机-1": [
"摄像机有效像素≥900W像素",
"摄像机最低照度彩色≤0.001lx",
"协议API 接口开放:具备;支持标准 ONVIF 协议与第三方厂家设备进行互联;支持 GB/T28181应提供 SDK"
],
"摄像机-2": [
"支持夜视", "支持云存储"
]
}}
"""
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)
ffinal_res=all_postprocess(final_res)
# final_res = postprocess(cleaned_res)
ffinal_res["货物列表"] = good_list
# 输出最终的 JSON 字符串
return {"采购需求": ffinal_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:对于软件系统功能,提取偏离表的时候额外处理 包头市公安支队机动车查验监管系统招标文201907_procurement.pdf
#TODO: 提示词限制最多两层防止,嵌套过多,代码也要做处理,'系统'下的最多只有一层 eg 妇幼、陕西
# {
# "采购需求": {
# "查验视频监管软件系统": {
# "视频设备管理": {
# "查验通道初始化": {},
# "视频设备绑定": {}
# },
# "业务视频应用": {
# "视频回放": {}
# },
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\货物标\output1\招标文件(107国道)_procurement.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))