# -*- encoding:utf-8 -*- import json import os.path import re from flask_app.general.json_utils import extract_content_from_json, clean_json_string # 可以选择性地导入特定的函数 from flask_app.工程标.提取打勾符号 import read_pdf_and_judge_main from flask_app.general.多线程提问 import multi_threading from flask_app.general.通义千问long import qianwen_long,upload_file #调用qianwen-ask之后，组织提示词问百炼。 def construct_judge_questions(json_data): """ 根据提供的 JSON 数据，生成一个问题列表。每个问题针对一个值为 '未知' 的键， 要求以 JSON 格式提供该键的值，仅限于 '是', '否', '未知'。 Args: json_data (str or dict): 输入的 JSON 数据，可以是 JSON 字符串或字典。 Returns: list: 包含生成的问题的列表。如果没有符合条件的键，则返回空列表。 """ # 如果 json_data 是字符串，尝试将其解析为字典 if isinstance(json_data, str): try: parsed_data = json.loads(json_data) except json.JSONDecodeError: print("输入的 json_data 不是有效的 JSON 字符串。") return [] elif isinstance(json_data, dict): parsed_data = json_data else: print("输入的 json_data 必须是 JSON 字符串或字典。") return [] # 提取值为 '未知' 的键 question_keys = [key for key, value in parsed_data.items() if value == '未知'] # 如果没有符合条件的键，返回空列表 if not question_keys: return [] # 生成问题列表 questions = [] for key in question_keys: question = ( f"请你根据投标人须知前附表中的信息回答，本项目{key}，" f"请按json格式给我提供信息，键名为'{key}'，键值仅限于'是','否','未知'，如果原文中对应内容处为'/'或者'\'，键值填'否'。" ) questions.append(question) return questions def merge_json_to_list(merged): # print(json.dumps(merged,ensure_ascii=False,indent=4)) """Merge updates into the original data by modifying specific keys based on their value ('是' or '否'), and create a list based on these values.""" chosen_numbers = [] chosen_numbers.append(9) # # 定义键名映射 # key_mapping = { # '是否允许分包': '分包', # '是否递交投标保证金': '投标保证金', # '是否递交磋商保证金': '磋商保证金', # '是否提交履约保证金': '履约保证金', # '是否有招标代理服务费': '招标代理服务费', # '是否组织踏勘现场': '踏勘现场', # '是否召开投标预备会': '投标预备会', # '是否召开投标答疑会': '投标答疑会', # '是否允许偏离': '偏离', # '是否退还投标文件':'退还投标文件' # } # 处理是否允许分包 if merged.get('是否允许分包') == '是': chosen_numbers.append(1) merged.pop('是否允许分包', None) elif merged.get('是否允许分包') == '否': merged['分包'] = '不允许' merged.pop('是否允许分包', None) # 处理保证金 guarantee_processed = False for guarantee_key in ['是否递交投标保证金', '是否递交磋商保证金']: if guarantee_key in merged and not guarantee_processed: guarantee_type = '投标' if '投标' in guarantee_key else '磋商' if merged[guarantee_key] == '是': chosen_numbers.extend([2, 3]) merged.pop(guarantee_key, None) elif merged[guarantee_key] == '否': merged[f'{guarantee_type}保证金'] = '不提交' merged.pop(guarantee_key, None) merged[f'退还{guarantee_type}保证金'] = '/' else: # 当既不是 '是' 也不是 '否' 时执行 merged[f'是否退还{guarantee_type}保证金'] = '未知' guarantee_processed = True elif guarantee_key in merged and guarantee_processed: merged.pop(guarantee_key, None) # 处理是否有履约保证金 if merged.get('是否提交履约保证金') == '是': chosen_numbers.append(4) merged.pop('是否提交履约保证金', None) elif merged.get('是否提交履约保证金') == '否': merged['履约保证金'] = '不提交' merged.pop('是否提交履约保证金', None) # 处理是否有招标代理服务费 if merged.get('是否有招标代理服务费') == '是': chosen_numbers.append(5) merged.pop('是否有招标代理服务费', None) elif merged.get('是否有招标代理服务费') == '否': merged['招标代理服务费'] = '无' merged.pop('是否有招标代理服务费', None) # 处理是否组织踏勘现场 if merged.get('是否组织踏勘现场') == '是': chosen_numbers.append(6) merged.pop('是否组织踏勘现场', None) elif merged.get('是否组织踏勘现场') == '否': merged['踏勘现场'] = '不组织' merged.pop('是否组织踏勘现场', None) # 处理预备会/答疑会 meeting_processed = False for preparation_key in ['是否召开投标预备会', '是否召开投标答疑会']: if preparation_key in merged and not meeting_processed: meeting_type = '预备会' if '预备会' in preparation_key else '答疑会' if merged[preparation_key] == '是': chosen_numbers.append(7) merged.pop(preparation_key,None) elif merged[preparation_key] == '否': merged[f'投标{meeting_type}'] = '不召开' merged.pop(preparation_key, None) else: # 当既不是 '是' 也不是 '否' 时执行 merged[f'是否召开投标{meeting_type}'] = '未知' meeting_processed = True elif preparation_key in merged and meeting_processed: merged.pop(preparation_key, None) if merged.get('是否有质量保证金') == '是': chosen_numbers.append(8) merged.pop('是否有质量保证金',None) elif merged.get('是否有质量保证金') == '否': merged['质量保证金']='不提交' merged.pop('是否有质量保证金',None) #11.12这里有问题，正文部分的信息重要性不高，而且这里的key不好设置，先注释了 # # 初始化 questions_list # questions_list = [] # if tobidders_notice: # file_id = upload_file(tobidders_notice) # 假设 upload_file 函数已定义 # question_template = ( # f"根据该招标文件中的信息，{{key}}的内容是怎样的？请按json格式给我提供信息，" # f"键名是'{{key}}'，若存在嵌套信息，嵌套内容键名以文件中对应字段命名（或是你对相应要求的总结），" # f"而对应键值需要与原文保持一致。注意：默认情况用普通键值对返回结果即可，键名为{{key}}；若原文中未提及{{key}}相关内容，在键值中填'未知'。" # ) # # 遍历 merged，查找值为 '未知' 的键 # keys_to_remove=[] # for original_key, value in merged.items(): # if value == '未知': # keys_to_remove.append(original_key) # # 获取映射后的键名 # mapped_key = key_mapping.get(original_key, original_key) # # 生成问题 # question = question_template.format(key=mapped_key) # questions_list.append(question) # for k in keys_to_remove: # del merged[k] # # 假设 questions_list 需要传递给 multi_threading 函数 # baseinfo_results = multi_threading(questions_list, "", file_id, 2) # 假设 multi_threading 函数已定义 # baseinfo_list = [clean_json_string(res) for _, res in baseinfo_results] if baseinfo_results else [] # # 更新 merged 中的 '未知' 值 # for info in baseinfo_list: # merged.update(info) # print(chosen_numbers) # print(json.dumps(merged,ensure_ascii=False,indent=4)) return chosen_numbers, merged def read_questions_from_judge(file_path, indices): questions = [] # 读取文件内容 with open(file_path, 'r', encoding='utf-8') as file: content = file.read() # 正则表达式提取问题 pattern = r'(\d+)\.(.*?)#pdf提取之后的提示词|(\d+)\.(.*?)(?=\d+\.|$)' matches = re.findall(pattern, content, re.DOTALL) # 解析匹配到的内容并提取对应序号的问题 for match in matches: num = match[0] or match[2] question = match[1].strip() or match[3].strip() if int(num) in indices: questions.append(question) return questions def judge_whether_main(file_path,output_folder): #传入招标文件中‘投标人须知前附表’ user_query1 = "请你依据以上信息，回答以下问题：是否组织踏勘现场？是否召开投标预备会？是否允许偏离？是否退还投标文件？是否允许分包? 是否需要递交投标保证金？是否需要提交履约保证金（履约担保）？是否有招标代理服务费？请按json格式给我提供信息，键名分别为'是否组织踏勘现场','是否召开投标预备会','是否允许偏离','是否退还投标文件',是否允许分包','是否递交投标保证金','是否提交履约保证金','是否有招标代理服务费',键值仅限于'是','否','未知',若存在矛盾信息，请回答'未知'。" output_txt_path = os.path.join(output_folder,'judge_exist.txt') read_pdf_and_judge_main(file_path, output_txt_path) #提取打勾符号 file_id=upload_file(output_txt_path) qianwen_answer=qianwen_long(file_id,user_query1) user_querys = construct_judge_questions(extract_content_from_json(qianwen_answer)) # 提取回答为”未知“的键 # 判断user_query2是否为空 if user_querys: file_id = upload_file(file_path) qianwen_results = multi_threading(user_querys, "", file_id, 2) # 1代表使用百炼rag 2代表使用qianwen-long qianwen_list = [extract_content_from_json(res) for _, res in qianwen_results] if qianwen_results else [] #整个前附表一起传问千问long return process_judge_content(qianwen_answer, qianwen_list) else: print("正常现象，没有'未知',无需调用qianwen-long") original = extract_content_from_json(qianwen_answer) return merge_json_to_list(original) def process_judge_content(original_json, qianwen_list): #用新的数据合并旧数据 """Process judging content by merging updates into the original JSON data.""" original = extract_content_from_json(original_json) for i in qianwen_list: original.update(i) return merge_json_to_list(original) if __name__ == "__main__": file_path="C:\\Users\Administrator\\Desktop\\fsdownload\\ee2d8828-bae0-465a-9171-7b2dd7453251\\ztbfile_tobidders_notice_table.pdf" output_dir='C:\\Users\Administrator\\Desktop\\fsdownload\\ee2d8828-bae0-465a-9171-7b2dd7453251\\tmp' chosen_numbers, merged=judge_whether_main(file_path,output_dir) print(chosen_numbers) print(merged)