87 lines
5.1 KiB
Python
87 lines
5.1 KiB
Python
#资格审查中，首先排除'联合体投标'和'不得存在的情况',有'符合'等的，加入matching_keys列表，否则保留原字典
|
||
import re
|
||
|
||
from json_utils import clean_json_string, combine_json_results, add_keys_to_json
|
||
from 多线程提问 import multi_threading, read_questions_from_file
|
||
from 通义千问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):
|
||
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 any(ex in key for ex in excludes):
|
||
continue # 如果包含排除项，则跳过当前键值对
|
||
# 检查值是否符合任何一个包含模式
|
||
if any(pattern.search(value) for pattern in include_patterns):
|
||
# 如果匹配，将键添加到列表中
|
||
matching_keys.append(key)
|
||
else:
|
||
# 如果不匹配，将键值对添加到不匹配字典中
|
||
non_matching_keys[key] = value
|
||
return matching_keys,non_matching_keys #matching:['资质条件', '财务状况'] non_matching_keys:{'营业执照': '具备有效的营业执照', '施工机械设备': '具备完善的施工设备'}
|
||
|
||
def process_qualification(qualification_review,truncate4,knowledge_name):
|
||
# 资格评审
|
||
matching_keys_list, non_matching_dict = extract_matching_keys_qual(qualification_review)
|
||
user_querys = generate_qual_question(matching_keys_list) # 生成提问->附件：资格审查
|
||
file_id2 = upload_file(truncate4)
|
||
results2 = multi_threading(user_querys, "", file_id2, 2) # 资格评审表
|
||
res_list = []
|
||
if not results2:
|
||
print("errror!")
|
||
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, '资格评审')
|
||
qualify_list = []
|
||
# qualification_review_file_path = '../static/提示词/资格评审问题.txt' # 替换为你的txt文件路径
|
||
qualification_review_file_path='static/提示词/资格评审问题.txt'
|
||
qualification_review_questions = read_questions_from_file(qualification_review_file_path) # 联合体投标
|
||
results1 = multi_threading(qualification_review_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}")
|
||
qualify_combined_dict = combine_json_results(qualify_list)
|
||
updated_qualify_json = add_keys_to_json(merged_dict, qualify_combined_dict) # 合并字典
|
||
final_qualify_json = add_keys_to_json(updated_qualify_json, non_matching_dict)
|
||
return final_qualify_json |