# -*- encoding:utf-8 -*- # 资格审查中，首先排除'联合体投标'和'不得存在的情况',有'符合'等的，加入matching_keys列表，否则保留原字典 import json import re from flask_app.general.json_utils import clean_json_string, combine_json_results, add_keys_to_json from flask_app.general.多线程提问 import multi_threading, read_questions_from_file from flask_app.general.通义千问long import upload_file, qianwen_long # 这个函数的主要用途是将多个相关的字典（都包含 '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: {}} # 遍历列表中的每个字典 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格式返回结果，最外层键名为'资格评审'，次外层键名为这些评审因素（如资质条件、信誉要求等），可能存在嵌套关系，但最内层键值为一个描述该评审因素的要求及备注的字典，其内层键名分别是'要求'和'备注'，若无具体备注信息，可删去'备注'键值对。你的回答内容需要与原文一致，不可擅自总结删减。以下为你需要考虑的特殊情况：1.若评审因素是项目人员（如项目经理、技术负责人等），除了'要求'，'备注'，还应增加一个键名'数量'，对应的键值为该岗位所需人数，若无相关要求，键值为'未知' 2.若评审因素为信誉要求，那么它为'要求'的对应键值为一个字符串列表，其中每个字符串是一条信誉要求。以下为示例输出，仅供格式参考： {{ "资格评审": {{ "资质条件": {{ "要求": "具备在中华人民共和国境内注册，具有有效营业执照", "备注": "原件扫描上传" }}, "信誉要求": {{ "要求": [ "1.没有被依法暂停或取消投标资格；", "2.没有被责令停产停业、暂扣或者吊销许可证、暂扣或者吊销执照；" ] }}, "项目经理资格": {{ "要求": "项目经理具有建筑工程专业二级及以上注册建造师执业资格并持有效的安全生产考核合格证（B证）", "数量": "1人" }}, "其他要求":{{ "项目管理机构人员": {{ "项目技术负责人": {{ "要求": "具备市政工程相关专业中级职称或具备市政公用工程专业贰级注册建造师执业资格（不含临时证）", "数量": "1人" }}, "施工管理": {{ "要求": "持有施工员岗位培训考核合格证书。", "数量": "未知" }} }} }} }} }} """ ) # 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 = ['联合体', '禁止投标', '不存在', '不得存在', '资格', '管理机构', '负责人', '人员','见本项目','详见资格'] # 联合体、禁止投标的情况、人员需要额外问 excludes = ['联合体'] # 遍历字典中的每个键值对 for key, value in dict_data.items(): if len(dict_data) == 1 and any(keyword in key for keyword in ("见", "符合", "满足")): 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(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(invalid_path, ques=None): if ques is None: ques = [] 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) # 确保 questions 和 ques 都是列表 if not isinstance(questions, list): print("读取的问题不是一个列表。") return {'资格评审': None} if not isinstance(ques, list): print("传入的额外问题不是一个列表。") return {'资格评审': None} questions += ques 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, qualification_path, invalid_path, merged_baseinfo_path): # 资格评审 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 qualification_path != "": # 提取到资格审查附件的情况,有啥内容提啥内容 print("资格评审： type1") # matching_keys_list = ["资质条件", "财务要求", "业绩要求", "信誉要求", "其他要求"] # ques = generate_qual_question(matching_keys_list) ques="""该招标文件中规定的资格性审查标准是怎样的？请以json格式给出结果，最外层键名为'资格评审'，次外层键名为各项评审要求（如资质条件、信誉要求等），可能存在嵌套关系，但最内层键值为一个描述该评审因素的要求及备注的字典，其内层键名分别是'要求'和'备注'，若无具体备注信息，可删去'备注'键值对。你的回答内容需要与原文一致，不可擅自总结删减。以下为你需要考虑的特殊情况：1.若评审因素是项目人员（如项目经理、技术负责人等），除了'要求'，'备注'，还应增加一个键名'数量'，对应的键值为该岗位所需人数，若无相关要求，键值为'未知' 2.若评审因素为信誉要求，那么它为'要求'的对应键值为一个字符串列表，其中每个字符串是一条信誉要求。以下为示例输出，仅供格式参考： { "资格评审": { "资质条件": { "要求": "具备在中华人民共和国境内注册，具有有效营业执照", "备注": "原件扫描上传" }, "信誉要求": { "要求": [ "1.没有被依法暂停或取消投标资格；", "2.没有被责令停产停业、暂扣或者吊销许可证、暂扣或者吊销执照；" ] }, "项目经理资格": {{ "要求": "项目经理具有建筑工程专业二级及以上注册建造师执业资格并持有效的安全生产考核合格证（B证）", "数量": "1人" }}, "其他要求": { "项目管理机构人员": { "项目技术负责人": { "要求": "具备市政工程相关专业中级职称或具备市政公用工程专业贰级注册建造师执业资格（不含临时证）", "数量": "1人" }, "施工管理": { "要求": "持有施工员岗位培训考核合格证书。", "数量": "1人" } } } } } """ file_id2 = upload_file(qualification_path) res=clean_json_string(qianwen_long(file_id2,ques)) # results2 = multi_threading(ques, "", file_id2, 2) # 资格评审表，调用qianwen-long # res_list = [clean_json_string(res) for _, res in results2] if results2 else [] return res # if res_list: # 生成外键是'资格评审'的字典 # merged_dict = merge_dictionaries_under_common_key(res_list, '资格评审') # consortium_dict = get_consortium_dict(merged_baseinfo_path) # 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(invalid_path) 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(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 qualification_path == "": # 这种情况是评分办法前附表中有要求，但是没有正确截取到'资格审查表' # print("资格评审： type4") # target=["资质","业绩","财务","信誉","人员","项目经理","负责人","联合体"] # question_template="该招标文件中{key}的内容是怎样的？请你以json格式返回结果，键名为{key}，若存在嵌套内容，嵌套键名为你对相应要求的总结，而对应键值需要完全与原文保持一致，不要擅自总结、删减。" # final_qualification = get_all_dict(invalid_path) # final_qualify_json = add_keys_to_json(final_qualification, non_matching_dict) # return final_qualify_json or {"资格评审": ""} print("资格评审： type4") target = ["资质", "业绩", "财务", "信誉", "人员", "项目经理", "负责人", "联合体"] # 找出不在 target 中的 keys non_target_keys = [key for key in matching_keys_list if not any(t in key for t in target)] if non_target_keys: # 如果有不在 target 中的 keys，使用它们构建问题 question_template = "该招标文件中{key}的内容是怎样的？请你以json格式返回结果，键名为'{key}'，若存在嵌套内容，嵌套键名为你对相应要求的总结，而对应键值需要完全与原文保持一致，不要擅自总结、删减。" questions = [question_template.format(key=key) for key in non_target_keys] # print(questions) # 假设这里有一个函数来处理这些问题并获取结果 final_qualification = get_all_dict(invalid_path, questions) else: print("所有键都在target中") # 如果所有键都在 target 中，保持原有逻辑 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: # 大多数情况 print("资格评审： type5") user_querys = generate_qual_question(matching_keys_list) # 生成提问->‘附件：资格审查’ file_id2 = upload_file(qualification_path) 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(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 = {'营业执照': '具备有效的营业执照', '安全生产许可证': '具备有效的安全生产许可证', '资质等级': '符合第二章“投标人须知”规定', '财务状况': '符合第二章“投标人须知”规定'} qualification_path= "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, qualification_path, invalid_path, merged_baseinfo_path) print(json.dumps(res, ensure_ascii=False, indent=4)) # 该招标文件中资格评审关于财务状况的内容是怎样的？请你以json格式返回结果，外层键名为'财务状况'，请你忠于原文，回答要求完整准确，不要擅自总结、删减，且不要回答诸如'见投标人须知前附表'或'见第x.x项规定'这类无实质性内容的回答。