zy123/zbparse
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

10.18小解析

Browse Source
This commit is contained in:
zy123 2024-10-19 17:25:56 +08:00
parent 91b4f1efc1
commit 0f85e9e5e4
6 changed files with 365 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
flask_app/main/json_utils.py
View File

@ -50,15 +50,25 @@ def clean_json_string(json_string):
"""清理JSON字符串，移除多余的反引号并解析为字典"""
return extract_content_from_json(json_string)
def combine_json_results(json_lists):
"""
将类json格式的列表整合成json数据（即大括号{}包裹）
将类json格式的列表整合成json数据（即大括号{}包裹）。
支持列表中的元素既是字符串又是字典。
"""
combined_result = {}
for json_str in json_lists:
if json_str.strip():
json_data = clean_json_string(json_str)
combined_result.update(json_data)
for item in json_lists:
if isinstance(item, str):
if item.strip():
json_data = clean_json_string(item)
if isinstance(json_data, dict):
combined_result.update(json_data)
else:
print(f"警告: 解析后的数据不是字典类型，跳过。内容: {item}")
elif isinstance(item, dict):
combined_result.update(item)
else:
print(f"警告: 不支持的类型 {type(item)}，跳过。内容: {item}")
return combined_result

99
flask_app/main/test.py
View File

@ -1,43 +1,78 @@
import re
import json
from flask_app.main.资格评审 import process_qualification
def read_questions_from_file(file_path):
questions = []
current_question = ""
current_number = 0
def test_process_qualification_type1():
# Test inputs
qualification_review = {} # Empty dict
# output_folder = "C:\\Users\\Administrator\\Desktop\\招标文件\\new_test"
truncate3 = "C:\\Users\\Administrator\\Desktop\\招标文件\\new_test\\zbtest2_qualification.pdf"
invalid_path = "C:\\Users\\Administrator\\Desktop\\招标文件\\new_test\\zbtest2_invalid.pdf" # Not used in this case
merged_baseinfo_path = "C:\\Users\\Administrator\\Desktop\\招标文件\\new_test\\zbtest2_merged_baseinfo.pdf"
with open(file_path, 'r', encoding='utf-8') as file:
for line in file:
line = line.strip()
# Call the function
res = process_qualification(qualification_review, truncate3, invalid_path, merged_baseinfo_path)
if not line: # 跳过空行
continue
# Print the output
print("Test Case 1 - Type1 Output:")
print(json.dumps(res, ensure_ascii=False, indent=4))
# Execute the test
# test_process_qualification_type1()
def test_process_qualification_type2():
# Test inputs
qualification_review = {} # Empty dict
truncate3 = ""
invalid_path = "C:\\Users\\Administrator\\Desktop\\招标文件\\new_test\\zbtest2_invalid.pdf"
merged_baseinfo_path = "C:\\Users\\Administrator\\Desktop\\招标文件\\new_test\\zbtest2_merged_baseinfo.pdf"
if line.startswith('#'): # 跳过以#开头的行
continue
# Call the function
res = process_qualification(qualification_review, truncate3, invalid_path, merged_baseinfo_path)
# Print the output
print("Test Case 2 - Type2 Output:")
print(json.dumps(res, ensure_ascii=False, indent=4))
# 检查是否是新的问题编号，例如 "1."
match = re.match(r'^(\d+)\.', line)
if match:
# 如果有之前的问题，保存它
if current_question:
questions.append(current_question.strip())
# Execute the test
# test_process_qualification_type2()
# 开始新的问题
current_number = int(match.group(1))
# 提取问题内容，去掉编号和点
current_question = line.split('.', 1)[1].strip() + "\n"
else:
# 继续添加到当前问题
current_question += line + "\n"
def test_process_qualification_type3():
# Test inputs
qualification_review = {
'营业执照': '具备有效的营业执照',
'不存在禁止投标的': '不存在第二章“投标人须知”第 1.4.3 项规定的情形'
}
truncate3 = "" # No qualification attachment
invalid_path = "C:\\Users\\Administrator\\Desktop\\招标文件\\new_test\\zbtest2_invalid.pdf" # Not used in this case
merged_baseinfo_path = "C:\\Users\\Administrator\\Desktop\\招标文件\\new_test\\zbtest2_merged_baseinfo.pdf"
# 添加最后一个问题（如果存在）
if current_question:
questions.append(current_question.strip())
# Call the function
res = process_qualification(qualification_review, truncate3, invalid_path, merged_baseinfo_path)
return questions
# Print the output
print("Test Case 3 - Type3 Output:")
print(json.dumps(res, ensure_ascii=False, indent=4))
qualification_review_file_path = 'D:\\flask_project\\flask_app\\static\\提示词\\资格评审.txt'
ques=read_questions_from_file(qualification_review_file_path)
print(ques)
# Execute the test
# test_process_qualification_type3()
def test_process_qualification_type4():
# Test inputs
qualification_review = {
'资质条件': '符合相关资质要求',
'财务状况': '符合财务健康标准'
}
truncate3 = "" # Missing qualification attachment
invalid_path = "C:\\Users\\Administrator\\Desktop\\招标文件\\new_test\\zbtest2_invalid.pdf" # Not used in this case
merged_baseinfo_path = "C:\\Users\\Administrator\\Desktop\\招标文件\\new_test\\zbtest2_merged_baseinfo.pdf"
# Call the function
res = process_qualification(qualification_review, truncate3, invalid_path, merged_baseinfo_path)
# Print the output
print("Test Case 4 - Type4 Output:")
print(json.dumps(res, ensure_ascii=False, indent=4))
# Execute the test
test_process_qualification_type4()

8
flask_app/main/截取pdf.py
View File

@ -111,7 +111,7 @@ def save_pages_to_new_pdf(pdf_path, output_folder, output_suffix, start_page, en
before_doc.add_page(pdf_document.pages[page_num])
with open(before_pdf_path, 'wb') as f_before:
before_doc.write(f_before)
print(f"已保存页面从 0 到 {start_page - 1} 为 {before_pdf_path}")
print(f"已保存页面从 0 到 {pages_to_extract} 为 {before_pdf_path}")
# 提取指定范围的页面
output_doc = PdfWriter()
@ -314,7 +314,7 @@ def truncate_pdf_multiple_old(input_path, output_folder):
def truncate_pdf_multiple(input_path, output_folder):
base_file_name = os.path.splitext(os.path.basename(input_path))[0] #纯文件名
truncate_files = []
for selection in range(1, 6):
for selection in range(1, 7):
files = truncate_pdf_main(input_path, output_folder, selection)
truncate_files.extend(files)
if truncate_files:
@ -429,8 +429,8 @@ def truncate_pdf_specific_engineering(pdf_path, output_folder,selections):
if __name__ == "__main__":
# input_path = "C:\\Users\\Administrator\\Desktop\\fsdownload\\4bda9fde-89fc-4e5e-94a4-ce6c43010f74\\ztbfile.pdf"
# output_folder = "C:\\Users\\Administrator\\Desktop\\fsdownload\\4bda9fde-89fc-4e5e-94a4-ce6c43010f74"
input_path="C:\\Users\\Administrator\\Desktop\\招标文件\\招标test文件夹\\zbtest16.pdf"
output_folder="C:\\Users\\Administrator\\Desktop\\招标文件\\special_output"
input_path="C:\\Users\\Administrator\\Desktop\\招标文件\\招标test文件夹\\zbtest2.pdf"
output_folder="C:\\Users\\Administrator\\Desktop\\招标文件\\new_test"
files=truncate_pdf_multiple(input_path,output_folder)
# selections = [5, 1] # 仅处理 selection 5、1 和 3
# files=truncate_pdf_specific_engineering(input_path,output_folder,selections)

158
flask_app/main/资格评审.py
View File

@ -1,12 +1,39 @@
# -*- encoding:utf-8 -*-
#资格审查中，首先排除'联合体投标'和'不得存在的情况',有'符合'等的，加入matching_keys列表，否则保留原字典
# 资格审查中，首先排除'联合体投标'和'不得存在的情况',有'符合'等的，加入matching_keys列表，否则保留原字典
import json
import os
import re
from flask_app.main.json_utils import clean_json_string, combine_json_results, add_keys_to_json, nest_json_under_key
from flask_app.main.多线程提问 import multi_threading,read_questions_from_file
from flask_app.main.通义千问long import upload_file
from flask_app.main.多线程提问 import multi_threading, read_questions_from_file
from flask_app.main.通义千问long import upload_file, qianwen_long
from flask_app.main.截取pdf import truncate_pdf_main
#这个函数的主要用途是将多个相关的字典（都包含 'common_key' 键）合并成一个更大的、综合的字典，所有相关信息都集中在 'common_key' 键下
def merge_dictionaries_under_common_key(dicts, common_key):
"""
dict1 = {
'资格评审': {
'公司资质': '通过',
'注册资金': '1000万元'
}
}
dict2 = {
'资格评审': {
'技术能力': '优秀',
'过往业绩': '5个相似项目'
}
}
return:
merged_dict = {
'资格评审': {
'公司资质': '通过',
'注册资金': '1000万元',
'技术能力': '优秀',
'过往业绩': '5个相似项目'
}
}
"""
# 初始化一个空字典来保存合并的结果
merged_dict = {common_key: {}}
@ -19,30 +46,35 @@ def merge_dictionaries_under_common_key(dicts, common_key):
print(f"资格评审： Warning: Dictionary does not contain the key {common_key}")
return merged_dict
def generate_qual_question(matching_keys_list): #这里假设资质、信誉与人员要求 要不都有、要不都没
def generate_qual_question(matching_keys_list): # 这里假设资质、信誉与人员要求 要不都有、要不都没
if not matching_keys_list:
return []
else:
questions=[]
questions = []
# 将列表转换为单引号包裹的格式，并用逗号和空格分隔
formatted_keys = ["'{}'".format(key) for key in matching_keys_list]
# 将格式化后的关键词列表连接成字符串
keys_string = "、".join(formatted_keys)
# 构造完整的问题语句
question1 = (f"该招标文件中资格评审的内容是怎样的？具体内容包括{keys_string}，"
"请你以json格式返回结果，外层键名为'资格评审'，嵌套键名为具体的字段，请你忠于原文，回答要求完整准确，不要擅自总结、删减。")
question2="该招标文件中资格评审中有关人员资格的要求是怎样的？请依次给出所需的岗位、需要的数量、资格要求、需要提交的证明材料（如具体的社保证明、技能证书等，若有时间要求请注明时间范围）、在岗要求、备注，若相关要求不存在，则无需返回该键值对。请你以json格式返回结果，外层键名为'资格评审'，嵌套键名为具体的要求，请你忠于原文，回答要求完整准确，不要擅自总结、删减。"
"请你以json格式返回结果，外层键名为'资格评审'，嵌套键名为具体的字段，请你忠于原文，回答要求完整准确，不要擅自总结、删减。")
question2 = "该招标文件中资格评审中有关人员资格的要求是怎样的？请依次给出所需的岗位、需要的数量、资格要求、需要提交的证明材料（如具体的社保证明、技能证书等，若有时间要求请注明时间范围）、在岗要求、备注，若相关要求不存在，则无需返回该键值对。请你以json格式返回结果，外层键名为'资格评审'，嵌套键名为具体的要求，请你忠于原文，回答要求完整准确，不要擅自总结、删减。"
questions.append(question1)
questions.append(question2)
return questions
def extract_matching_keys_qual(dict_data):
# 定义包含模式的列表
include_patterns = [re.compile(r"第.*?章"), re.compile(r"第.*?款"), re.compile(r"第.*?项"), re.compile(r"第.*?目"),re.compile(r"符合")]
include_patterns = [re.compile(r"第.*?章"), re.compile(r"第.*?款"), re.compile(r"第.*?项"), re.compile(r"第.*?目"),
re.compile(r"符合")]
# 初始化列表，用于存储匹配的键
matching_keys = []
non_matching_keys = {}
# 定义排除项
excludes = ['联合体', '禁止投标', '不存在', '不得存在','资格','管理机构','负责人','人员'] #联合体、禁止投标的情况、人员需要额外问
excludes = ['联合体', '禁止投标', '不存在', '不得存在', '资格', '管理机构', '负责人', '人员'] # 联合体、禁止投标的情况、人员需要额外问
# 遍历字典中的每个键值对
for key, value in dict_data.items():
if "附件" in key and "资质" in key:
@ -57,62 +89,47 @@ def extract_matching_keys_qual(dict_data):
# value中有实质的内容，不需要额外问。
non_matching_keys[key] = value
return matching_keys, non_matching_keys #matching:['资质条件', '财务状况'] non_matching_keys:{'营业执照': '具备有效的营业执照', '施工机械设备': '具备完善的施工设备'}
return matching_keys, non_matching_keys # matching:['资质条件', '财务状况'] non_matching_keys:{'营业执照': '具备有效的营业执照', '施工机械设备': '具备完善的施工设备'}
#获取联合体投标的要求，由于它不一定在资格审查表中，故调用rag
def get_consortium_dict(knowledge_name):
qualify_list = []
consortium_questions = [
"该招标文件对于联合体投标的要求是怎样的，请按json格式给我提供信息，外层键名为'联合体投标要求（如有）'，嵌套键名为你对该要求的总结，而键值需要完全与原文保持一致，不要擅自总结、删减。"]
results1 = multi_threading(consortium_questions, knowledge_name)
for _, response in results1: # _占位，代表ques;response[0]也是ques;response[1]是ans
try:
if response and len(response) > 1: # 检查response存在且有至少两个元素
qualify_list.append(response[1])
else:
print(f"资格评审： Warning: Missing or incomplete response data for query index {_}.")
except Exception as e:
print(f"资格评审： Error processing response for query index {_}: {e}")
consortium_dict = combine_json_results(qualify_list)
# 获取联合体投标的要求，由于它不一定在资格审查表中，故调用rag
def get_consortium_dict(merged_baseinfo_path):
consortium_dict={"联合体投标要求（如有）":""}
consortium_questions = "该招标文件对于联合体投标的要求是怎样的，请按json格式给我提供信息，外层键名为'联合体投标要求（如有）'，嵌套键名为你对该要求的总结，而键值需要完全与原文保持一致，不要擅自总结、删减。"
file_id=upload_file(merged_baseinfo_path)
results1 = qianwen_long(file_id,consortium_questions)
if results1:
consortium_dict=clean_json_string(results1)
return consortium_dict
def get_all_dict(knowledge_name):
qualification_review_file_path='flask_app/static/提示词/资格评审.txt'
def get_all_dict(invalid_path):
# qualification_review_file_path = 'flask_app/static/提示词/资格评审.txt'
qualification_review_file_path='D:\\flask_project\\flask_app\\static\\提示词\\资格评审.txt'
questions = read_questions_from_file(qualification_review_file_path)
qualification_list = []
res1 = multi_threading(questions, knowledge_name)
for _, response in res1: # _占位，代表ques;response[0]也是ques;response[1]是ans
try:
if response and len(response) > 1: # 检查response存在且有至少两个元素
qualification_list.append(response[1])
else:
print(f"资格评审： Warning: Missing or incomplete response data for query index {_}.")
except Exception as e:
print(f"资格评审： Error processing response for query index {_}: {e}")
file_id=upload_file(invalid_path)
res1 = multi_threading(questions,"",file_id,2)
qualification_list = [clean_json_string(res) for _, res in res1] if res1 else []
qualification_combined_res = combine_json_results(qualification_list)
return {'资格评审': qualification_combined_res}
def process_qualification(qualification_review,truncate3,knowledge_name):
def process_qualification(qualification_review, truncate3, invalid_path,merged_baseinfo_path):
# 资格评审
matching_keys_list, non_matching_dict = extract_matching_keys_qual(qualification_review)
if not matching_keys_list: #此时要求全部写在评分办法前附表中，不需要额外提取。
if not non_matching_dict: #古法提取
if truncate3!="":
matching_keys_list, non_matching_dict = extract_matching_keys_qual(
qualification_review) # matching_keys_list:['资质条件', '财务状况'] non_matching_dict:{'营业执照': '具备有效的营业执照', '施工机械设备': '具备完善的施工设备'}
if not matching_keys_list:
if not non_matching_dict: # 古法提取 non_matching_dict和matching_keys_list都为空
if truncate3 != "": # 提取到资格审查附件的情况
print("资格评审： type1")
matching_keys_list=["资质条件","财务要求","业绩要求","信誉要求","其他要求"]
ques=generate_qual_question(matching_keys_list)
matching_keys_list = ["资质条件", "财务要求", "业绩要求", "信誉要求", "其他要求"]
ques = generate_qual_question(matching_keys_list)
file_id2 = upload_file(truncate3)
results2 = multi_threading(ques, "", file_id2, 2) # 资格评审表，调用qianwen-long
res_list = []
if not results2:
print("资格评审： 调用大模型未成功获取资格评审文件中的要求！")
else:
# 打印结果
for question, response in results2:
cleaned_res = clean_json_string(response)
res_list.append(cleaned_res) # 都是问资格评审表得出的
res_list = [clean_json_string(res) for _, res in results2] if results2 else []
if res_list:
#生成外键是'资格评审'的字典
merged_dict = merge_dictionaries_under_common_key(res_list, '资格评审')
consortium_dict = get_consortium_dict(knowledge_name)
consortium_dict = get_consortium_dict(merged_baseinfo_path)
updated_qualify_json = add_keys_to_json(merged_dict, consortium_dict)
return updated_qualify_json
else:
@ -120,26 +137,26 @@ def process_qualification(qualification_review,truncate3,knowledge_name):
return {"资格评审": ""}
else:
print("资格评审： type2")
return get_all_dict(knowledge_name) or {"资格评审": ""}
return get_all_dict(invalid_path) or {"资格评审": ""}
else:
else: # 此时要求全部写在评标办法前附表中，不需要额外提取。
print("资格评审： type3")
new_non_matching_json={'资格评审':non_matching_dict}
new_non_matching_json = {'资格评审': non_matching_dict}
substring = '联合体'
found_key = any(substring in key for key in non_matching_dict.keys()) #没有联合体投标，则需生成，防止重复
found_key = any(substring in key for key in non_matching_dict.keys()) # 没有联合体投标，则需生成，防止重复
if not found_key:
consortium_dict=get_consortium_dict(knowledge_name)
consortium_dict = get_consortium_dict(merged_baseinfo_path)
final_qualify_json = add_keys_to_json(new_non_matching_json, consortium_dict)
return final_qualify_json
else:
return new_non_matching_json or {"资格评审": ""}
elif matching_keys_list and truncate3=="": #这种情况是评分办法前附表中有要求，但是没有正确截取到'资格审查表'
elif matching_keys_list and truncate3 == "": # 这种情况是评分办法前附表中有要求，但是没有正确截取到'资格审查表'
print("资格评审： type4")
final_qualification=get_all_dict(knowledge_name)
final_qualification = get_all_dict(invalid_path)
final_qualify_json = add_keys_to_json(final_qualification, non_matching_dict)
return final_qualify_json or {"资格评审": ""}
else: #大多数情况
else: # 大多数情况
print("资格评审： type5")
user_querys = generate_qual_question(matching_keys_list) # 生成提问->‘附件：资格审查’
file_id2 = upload_file(truncate3)
@ -154,17 +171,22 @@ def process_qualification(qualification_review,truncate3,knowledge_name):
cleaned_res = clean_json_string(response)
res_list.append(cleaned_res) # 都是问资格评审表得出的
merged_dict = merge_dictionaries_under_common_key(res_list, '资格评审')
consortium_dict=get_consortium_dict(knowledge_name)
consortium_dict = get_consortium_dict(merged_baseinfo_path)
updated_qualify_json = add_keys_to_json(merged_dict, consortium_dict) # 合并字典
final_qualify_json = add_keys_to_json(updated_qualify_json, non_matching_dict)
return final_qualify_json or {"资格评审": ""}
if __name__ == "__main__":
# qualification_review={'营业执照': '具备有效的营业执照', '资质等级': '具备建设行政主管部门颁发的市政公用工程监理乙级及以上资质或房屋建筑工程监理乙级及以上资质或工程监理综合资质证书', '财务状况': '投标人须提供近三年（2018 年、2019 年、2020 年）完', '类似项目业绩': '投标人近 5 年（2017 年至今）须具有至少一项投资概算在4000 万元及以上房屋建筑工程或市政公用工程监理业绩，同时提供中标通知书及监理服务合同，项目规模及项目时间认定以监理服务合同内信息为准', '信誉': '根据《关于在招标投标活动中对失信被执行', '主要人员': '监理工程师：至少提供 1 名具有房屋建筑工程专','不存在禁止投标的':'不存在第二章“投标人须知”第 1.4.3 项规定的情形','联合体投标':'hha'}
qualification_review={'营业执照':'具备有效的营业执照','安全生产许可证':'具备有效的安全生产许可证','资质等级':'符合第二章“投标人须知”规定','财务状况':'符合第二章“投标人须知”规定'}
truncate3="C:\\Users\\Administrator\\Desktop\\招标文件\\招标test文件夹\\zbtest13_qualification.pdf"
knowledge_name="招标解析word13"
res=process_qualification(qualification_review,truncate3,knowledge_name)
print(json.dumps(res,ensure_ascii=False,indent=4))
qualification_review = {'营业执照': '具备有效的营业执照', '安全生产许可证': '具备有效的安全生产许可证',
'资质等级': '符合第二章“投标人须知”规定', '财务状况': '符合第二章“投标人须知”规定'}
truncate3 = "C:\\Users\\Administrator\\Desktop\\招标文件\\new_test\\zbtest2_qualification.pdf"
# output_folder="C:\\Users\\Administrator\\Desktop\\招标文件\\new_test"
invalid_path="C:\\Users\\Administrator\\Desktop\\招标文件\\new_test\\zbtest2_invalid.pdf"
merged_baseinfo_path="C:\\Users\\Administrator\\Desktop\\招标文件\\new_test\\zbtest2_merged_baseinfo.pdf"
# knowledge_name = "招标解析word13"
res = process_qualification(qualification_review, truncate3,invalid_path,merged_baseinfo_path)
print(json.dumps(res, ensure_ascii=False, indent=4))
#该招标文件中资格评审关于财务状况的内容是怎样的？请你以json格式返回结果，外层键名为'财务状况'，请你忠于原文，回答要求完整准确，不要擅自总结、删减，且不要回答诸如'见投标人须知前附表'或'见第x.x项规定'这类无实质性内容的回答。
# 该招标文件中资格评审关于财务状况的内容是怎样的？请你以json格式返回结果，外层键名为'财务状况'，请你忠于原文，回答要求完整准确，不要擅自总结、删减，且不要回答诸如'见投标人须知前附表'或'见第x.x项规定'这类无实质性内容的回答。

182
flask_app/main/资格评审old.py Normal file
View File

@ -0,0 +1,182 @@
# -*- encoding:utf-8 -*-
# 资格审查中，首先排除'联合体投标'和'不得存在的情况',有'符合'等的，加入matching_keys列表，否则保留原字典
import json
import re
from flask_app.main.json_utils import clean_json_string, combine_json_results, add_keys_to_json, nest_json_under_key
from flask_app.main.多线程提问 import multi_threading, read_questions_from_file
from flask_app.main.通义千问long import upload_file
def merge_dictionaries_under_common_key(dicts, common_key):
# 初始化一个空字典来保存合并的结果
merged_dict = {common_key: {}}
# 遍历列表中的每个字典
for d in dicts:
if common_key in d:
# 使用字典解包来合并字典
merged_dict[common_key].update(d[common_key])
else:
print(f"资格评审： Warning: Dictionary does not contain the key {common_key}")
return merged_dict
def generate_qual_question(matching_keys_list): # 这里假设资质、信誉与人员要求 要不都有、要不都没
if not matching_keys_list:
return []
else:
questions = []
# 将列表转换为单引号包裹的格式，并用逗号和空格分隔
formatted_keys = ["'{}'".format(key) for key in matching_keys_list]
# 将格式化后的关键词列表连接成字符串
keys_string = "、".join(formatted_keys)
# 构造完整的问题语句
question1 = (f"该招标文件中资格评审的内容是怎样的？具体内容包括{keys_string}，"
"请你以json格式返回结果，外层键名为'资格评审'，嵌套键名为具体的字段，请你忠于原文，回答要求完整准确，不要擅自总结、删减。")
question2 = "该招标文件中资格评审中有关人员资格的要求是怎样的？请依次给出所需的岗位、需要的数量、资格要求、需要提交的证明材料（如具体的社保证明、技能证书等，若有时间要求请注明时间范围）、在岗要求、备注，若相关要求不存在，则无需返回该键值对。请你以json格式返回结果，外层键名为'资格评审'，嵌套键名为具体的要求，请你忠于原文，回答要求完整准确，不要擅自总结、删减。"
questions.append(question1)
questions.append(question2)
return questions
def extract_matching_keys_qual(dict_data):
# 定义包含模式的列表
include_patterns = [re.compile(r"第.*?章"), re.compile(r"第.*?款"), re.compile(r"第.*?项"), re.compile(r"第.*?目"),
re.compile(r"符合")]
# 初始化列表，用于存储匹配的键
matching_keys = []
non_matching_keys = {}
# 定义排除项
excludes = ['联合体', '禁止投标', '不存在', '不得存在', '资格', '管理机构', '负责人', '人员'] # 联合体、禁止投标的情况、人员需要额外问
# 遍历字典中的每个键值对
for key, value in dict_data.items():
if "附件" in key and "资质" in key:
return [], [] # 如果同时出现，立即返回空列表
# 检查值是否符合任何一个包含模式
if any(pattern.search(value) for pattern in include_patterns):
# 如果值符合包含模式，再检查键是否包含任何排除项
if not any(ex in key for ex in excludes):
# 如果键不包含排除项，则添加到匹配键列表中
matching_keys.append(key)
else:
# value中有实质的内容，不需要额外问。
non_matching_keys[key] = value
return matching_keys, non_matching_keys # matching:['资质条件', '财务状况'] non_matching_keys:{'营业执照': '具备有效的营业执照', '施工机械设备': '具备完善的施工设备'}
# 获取联合体投标的要求，由于它不一定在资格审查表中，故调用rag
def get_consortium_dict(knowledge_name):
qualify_list = []
consortium_questions = [
"该招标文件对于联合体投标的要求是怎样的，请按json格式给我提供信息，外层键名为'联合体投标要求（如有）'，嵌套键名为你对该要求的总结，而键值需要完全与原文保持一致，不要擅自总结、删减。"]
results1 = multi_threading(consortium_questions, knowledge_name)
for _, response in results1: # _占位，代表ques;response[0]也是ques;response[1]是ans
try:
if response and len(response) > 1: # 检查response存在且有至少两个元素
qualify_list.append(response[1])
else:
print(f"资格评审： Warning: Missing or incomplete response data for query index {_}.")
except Exception as e:
print(f"资格评审： Error processing response for query index {_}: {e}")
consortium_dict = combine_json_results(qualify_list)
return consortium_dict
def get_all_dict(knowledge_name):
qualification_review_file_path = 'flask_app/static/提示词/资格评审.txt'
questions = read_questions_from_file(qualification_review_file_path)
qualification_list = []
res1 = multi_threading(questions, knowledge_name)
for _, response in res1: # _占位，代表ques;response[0]也是ques;response[1]是ans
try:
if response and len(response) > 1: # 检查response存在且有至少两个元素
qualification_list.append(response[1])
else:
print(f"资格评审： Warning: Missing or incomplete response data for query index {_}.")
except Exception as e:
print(f"资格评审： Error processing response for query index {_}: {e}")
qualification_combined_res = combine_json_results(qualification_list)
return {'资格评审': qualification_combined_res}
def process_qualification(qualification_review, truncate3, knowledge_name):
# 资格评审
matching_keys_list, non_matching_dict = extract_matching_keys_qual(
qualification_review) # matching_keys_list:['资质条件', '财务状况'] non_matching_dict:{'营业执照': '具备有效的营业执照', '施工机械设备': '具备完善的施工设备'}
if not matching_keys_list:
if not non_matching_dict: # 古法提取
if truncate3 != "": # 提取到资格审查附件的情况
print("资格评审： type1")
matching_keys_list = ["资质条件", "财务要求", "业绩要求", "信誉要求", "其他要求"]
ques = generate_qual_question(matching_keys_list)
file_id2 = upload_file(truncate3)
results2 = multi_threading(ques, "", file_id2, 2) # 资格评审表，调用qianwen-long
res_list = []
if not results2:
print("资格评审： 调用大模型未成功获取资格评审文件中的要求！")
else:
# 打印结果
for question, response in results2:
res_list.append(clean_json_string(response)) # 都是问资格评审表得出的
if res_list:
merged_dict = merge_dictionaries_under_common_key(res_list, '资格评审')
consortium_dict = get_consortium_dict(knowledge_name)
updated_qualify_json = add_keys_to_json(merged_dict, consortium_dict)
return updated_qualify_json
else:
print("资格评审： 无法获取大模型结果，返回空值")
return {"资格评审": ""}
else:
print("资格评审： type2")
return get_all_dict(knowledge_name) or {"资格评审": ""}
else: # 此时要求全部写在评分办法前附表中，不需要额外提取。
print("资格评审： type3")
new_non_matching_json = {'资格评审': non_matching_dict}
substring = '联合体'
found_key = any(substring in key for key in non_matching_dict.keys()) # 没有联合体投标，则需生成，防止重复
if not found_key:
consortium_dict = get_consortium_dict(knowledge_name)
final_qualify_json = add_keys_to_json(new_non_matching_json, consortium_dict)
return final_qualify_json
else:
return new_non_matching_json or {"资格评审": ""}
elif matching_keys_list and truncate3 == "": # 这种情况是评分办法前附表中有要求，但是没有正确截取到'资格审查表'
print("资格评审： type4")
final_qualification = get_all_dict(knowledge_name)
final_qualify_json = add_keys_to_json(final_qualification, non_matching_dict)
return final_qualify_json or {"资格评审": ""}
else: # 大多数情况
print("资格评审： type5")
user_querys = generate_qual_question(matching_keys_list) # 生成提问->‘附件：资格审查’
file_id2 = upload_file(truncate3)
results2 = multi_threading(user_querys, "", file_id2, 2) # 资格评审表，调用qianwen-long
res_list = []
if not results2:
print("资格评审： 调用大模型未成功获取资格评审文件中的要求！")
return {"资格评审": ""}
else:
# 打印结果
for question, response in results2:
cleaned_res = clean_json_string(response)
res_list.append(cleaned_res) # 都是问资格评审表得出的
merged_dict = merge_dictionaries_under_common_key(res_list, '资格评审')
consortium_dict = get_consortium_dict(knowledge_name)
updated_qualify_json = add_keys_to_json(merged_dict, consortium_dict) # 合并字典
final_qualify_json = add_keys_to_json(updated_qualify_json, non_matching_dict)
return final_qualify_json or {"资格评审": ""}
if __name__ == "__main__":
# qualification_review={'营业执照': '具备有效的营业执照', '资质等级': '具备建设行政主管部门颁发的市政公用工程监理乙级及以上资质或房屋建筑工程监理乙级及以上资质或工程监理综合资质证书', '财务状况': '投标人须提供近三年（2018 年、2019 年、2020 年）完', '类似项目业绩': '投标人近 5 年（2017 年至今）须具有至少一项投资概算在4000 万元及以上房屋建筑工程或市政公用工程监理业绩，同时提供中标通知书及监理服务合同，项目规模及项目时间认定以监理服务合同内信息为准', '信誉': '根据《关于在招标投标活动中对失信被执行', '主要人员': '监理工程师：至少提供 1 名具有房屋建筑工程专','不存在禁止投标的':'不存在第二章“投标人须知”第 1.4.3 项规定的情形','联合体投标':'hha'}
qualification_review = {'营业执照': '具备有效的营业执照', '安全生产许可证': '具备有效的安全生产许可证',
'资质等级': '符合第二章“投标人须知”规定', '财务状况': '符合第二章“投标人须知”规定'}
truncate3 = "C:\\Users\\Administrator\\Desktop\\招标文件\\new_test\\zbtest2_qualification.pdf"
knowledge_name = "招标解析word13"
res = process_qualification(qualification_review, truncate3, knowledge_name)
print(json.dumps(res, ensure_ascii=False, indent=4))
# 该招标文件中资格评审关于财务状况的内容是怎样的？请你以json格式返回结果，外层键名为'财务状况'，请你忠于原文，回答要求完整准确，不要擅自总结、删减，且不要回答诸如'见投标人须知前附表'或'见第x.x项规定'这类无实质性内容的回答。

8
flask_app/static/提示词/资格评审.txt
View File

@ -8,7 +8,13 @@
3.该招标文件中资格审查部分对于投标人的财务要求是怎样的，并按json格式给我提供信息，外层键名为'财务要求'，嵌套键名为你对相应要求的总结，而对应键值需要完全与原文保持一致，不要擅自总结、删减，不要回答有关资质要求、业绩要求、人员要求、信誉要求的内容。
#信誉要求：
4.该招标文件中资格审查部分对于投标人的信誉要求是怎样的，请按json列表格式给我提供信息，键名为'信誉要求'，键值需要完全与原文保持一致，不要擅自总结、删减，不要回答无关信誉要求的内容。
4.该招标文件中资格审查部分对于投标人的信誉要求是怎样的，请按json列表格式给我提供信息，键名为'信誉要求'，若有多条信誉要求，用字符串列表保存各个要求，各要求需完全与原文保持一致，不要擅自总结、删减，不要回答无关信誉要求的内容。以下是示例格式：
{
"信誉要求":[
"没有被责令停业；没有被暂停或取消投标资格；财产没有被接管或冻结",
"投标人在“信用中国”网站（www.creditchina.gov.cn）中未被列入失信被执行人、重大税收违法案件当事人名单的不良行为记录"
]
}
#（存在问题）主要人员要求：
5.该招标文件中资格审查部分对于项目经理（监理）和技术负责人的要求和数量是怎样的，以json的形式给出，键名分别是"项目经理"和"技术负责人"，其他嵌套键名为你对相应要求的总结，而对应键值需要完全与原文保持一致，不要擅自总结、删减。