# -*- encoding:utf-8 -*- import json import re import time from collections import defaultdict from flask_app.general.通义千问long import upload_file, qianwen_long def combine_technical_and_business(data, target_values): extracted_data = {} # 根级别存储所有数据 technical_found = False business_found = False def extract_nested(data, parent_key='', is_technical=False, is_business=False): nonlocal technical_found, business_found if isinstance(data, dict): for key, value in data.items(): current_key = f"{parent_key}.{key}" if parent_key else key # 检查是否为技术标的内容 if any(target in key for target in target_values): if not is_technical: extracted_data[key] = value technical_found = True continue # 默认其他所有内容都归为商务标 else: if not is_business: if '商务评分' not in extracted_data: extracted_data['商务评分'] = {} extracted_data['商务评分'][key] = value business_found = True continue if isinstance(value, dict) or isinstance(value, list): extract_nested(value, current_key, is_technical, is_business) elif isinstance(data, list): for index, item in enumerate(data): extract_nested(item, f"{parent_key}[{index}]", is_technical, is_business) extract_nested(data) if not technical_found: extracted_data['技术评分'] = '' if not business_found: extracted_data['商务评分'] = '' return extracted_data # 防止外键只有一个'一包'的情况 def process_data_based_on_key(data): exclude_word = ["包", "未知", "评分因素"] # 获取字典的键列表 keys = list(data.keys()) # 检查键的数量是否为1并且 exclude_word 中的任何词包含在 keys[0] 中 if len(keys) == 1 and any(word in keys[0] for word in exclude_word): # 返回内层的字典 return data[keys[0]] # 如果条件不满足，则返回原始字典 return data def parse_json_with_duplicates(raw_string): """ 解析具有重复键的 JSON 字符串，将所有重复的键值对存储为列表。 Args: json_string (str): 需要解析的 JSON 字符串。 Returns: dict: 解析后的字典，重复的键对应的值为列表。 eg:输入："综合实力": { "评分": "2分", "要求": "投标人具备电子与智能化工程专业承包二级资质及以上证书得 2分，不能够提供不得分（开标时需提供原件）。" }, "综合实力": { "评分": "2分", "要求": "投标人具有建筑机电安装工程专业承包三级资质或以上资质得 2分，否则不得分。（证书开标原件备查）。" } 输出："综合实力": [ { "评分": "2分", "要求": "投标人具备电子与智能化工程专业承包二级资质及以上证书得 2分，不能够提供不得分（开标时需提供原件）。" }, { "评分": "2分", "要求": "投标人具有建筑机电安装工程专业承包三级资质或以上资质得 2分，否则不得分。（证书开标原件备查）。" }] """ def custom_object_pairs_hook(pairs): d = defaultdict(list) for key, value in pairs: try: # 如果值是字典或列表，递归处理 if isinstance(value, dict): value = process_dict(value) elif isinstance(value, list): value = process_list(value) d[key].append(value) except Exception as e: d[key].append(value) # 根据需求决定是否跳过或保留原值 # 将有多个值的键转换为列表，单个值的键保持原样 return {key: (values if len(values) > 1 else values[0]) for key, values in d.items()} def process_dict(d): """ 递归处理字典，确保所有重复键的值为列表。 Args: d (dict): 需要处理的字典。 Returns: dict: 处理后的字典。 """ try: return custom_object_pairs_hook(d.items()) except Exception as e: return {} def process_list(l): """ 递归处理列表，确保列表中的所有字典也被处理。 Args: l (list): 需要处理的列表。 Returns: list: 处理后的列表。 """ try: return [process_dict(item) if isinstance(item, dict) else item for item in l] except Exception as e: return [] """输入字符串，提取 { 和 } 之间的内容，并将其解析为字典""" if not raw_string.strip(): return {} match = re.search(r'\{[\s\S]*\}', raw_string) if match: try: json_string = match.group(0) return json.loads(json_string, object_pairs_hook=custom_object_pairs_hook) except json.JSONDecodeError as e: print(f"json_utils: extract_content_from_json: JSON decode error: {e}") return {} else: print("json_utils: extract_content_from_json: No valid JSON content found.") return {} def reorganize_data(input_dict, include=None): """ 重组输入字典，将“技术评分”和“商务评分”提升为最外层键， 并将包含在 include 列表中的包名的数据嵌套在相应的评分类别下。 如果 input_dict 的顶层键不包含任何 include 列表中的项，则返回原始字典。 :param input_dict: 原始输入字典 :param include: 包名列表，例如 ['一包', '二包', '三包'] :return: 重组后的字典 """ if include is None: include = [] # 检查是否有任何顶层键包含在 include 列表中 has_include = any(key in include for key in input_dict.keys()) if not has_include: # 没有包含任何指定的包名，直接返回原始字典 return input_dict # 初始化新的字典结构 reorganized = { "技术评分": {}, "商务评分": {} } # 遍历每一个包（如 "一包", "二包"） for package, categories in input_dict.items(): # 处理技术评分 if "技术评分" in categories: reorganized["技术评分"][package] = categories["技术评分"] # 处理商务评分 if "商务评分" in categories: reorganized["商务评分"][package] = categories["商务评分"] return reorganized def combine_evaluation_standards(truncate_file): # 定义默认的评审结果字典 DEFAULT_EVALUATION_REVIEW = { "技术评分": "", "商务评分": "" } # 如果 truncate_file 是空字符串，直接返回包含空字符串的字典 if not truncate_file: return DEFAULT_EVALUATION_REVIEW.copy() try: # 上传文件并获取文件ID file_id = upload_file(truncate_file) # 定义用户查询 user_query1 = ( "根据该文档，你判断它是否有关于技术评分或商务评分或投标报价的具体的评分及要求，如果有，返回'是'，否则返回'否'。" ) # 应对竞争性谈判这种无评分要求的情况 # 执行查询 judge_res = qianwen_long(file_id, user_query1) # 默认 judge 为 True judge = True # 检查 judge_res 的内容 if '否' in judge_res: judge = False if judge: # 执行 user_query 相关的逻辑 # user_query = "根据该文档中的评标办法前附表或者评分标准表，请你列出该文件的技术评分，商务评分，投标报价评审标准以及它们对应的具体评分要求，外层键名分别为'技术评分','商务评分','投标报价'。如果评分内容不是这3个，则返回文档中给定的评分内容以及它的评分要求，都以json的格式返回结果，如果该采购活动有多个包，则最外层键名为对应的包名。请不要回答有关资格审查的内容" user_query = ( """ 根据该文档中的评分标准表格中的内容，请你列出该文件的技术评分，商务评分，投标报价评审以及它们对应的具体评分要求，请以json格式返回结果，最外层键名分别是'技术评分','商务评分','投标报价评分',请在这三大项评分中分别用若干键值对表示具体评分项，外层键名为各评审因素，键值为一个列表，列表中包含若干（可为一）描述该评审因素的评分及要求的字典，内层键名分别是'评分'和'要求'，若无评分，可删去'评分'键值对，'要求'中说明了该评审因素的评分标准；若这三大项评分中存在其他信息，则在相应评分大块内部新增键名'备注'存放该信息，键值为具体的要求，否则不需要。如果评分内容（因素）不是这三大项，则返回文档中给定的评分内容（因素）以及它们的具体评分要求。 要求与指南： 1.请首先定位评分细则的表格，不要回答有关资格审查、符合性审查的内容，也不要从评标办法正文中（表格外）提取回答 2.若大项的'xx评分'要求未在文中说明，则键名'xx评分'的键值设为'本项目无xx评分项'，例如"技术评分":"本项目无技术评分项" 3. 如果该招标活动有多个包，则最外层键名为对应的包名,否则不需要 4.你无需将表格的单元格内的内容进行拆分，需要将它视为一个整体。 5. '评分'的键值不能是一个范围数字，如'0-5分'，应该是一个具体数字，如'5分'，或者是一个定性的指标如'合格制' 以下为示例输出，仅供格式参考： { "一包": { "技术评分": { "主要监理岗位的职责": [ { "评分": "4分", "要求": "1、总监理工程师的职责全面、清晰、合理得 1.2-2分；一般的1.2分。2、其他主要监理人员及岗位的职责全面、清晰、合理得 1.2-2分；一般的 1.2分。" } ], "备注": "注：若不满足“与公安部、省公安厅、随州市公安局高清视频会议系统无缝对接互联互通”的要求，则本项技术部分（50分）不得分。" }, "商务评分": { "控制系统内主板": [ { "评分": "10分", "要求": "所投电梯控制系统内主板为制造商原厂原品牌制造生产且为进口部件得 10分。（提供进口部件报关单及原产地证明扫描件加盖公章，否则不得分）" } ], "制造商技术实力": [ { "评分": "3分", "要求": "一级证书得3分，二级证书得1分，其他不得分" }, { "评分": "2分", "要求": "行业销量排名连续前 2 名，得 2 分，第 4-6 名得 0.5 分，其他不得分" } ] }, "投标报价评审": { "投标报价是否出现违反计价规范": [ { "评分": "合格制", "要求": "A:投标报价未违反计价规范的，评审意见为“合格”；B：投标报价违反计价规范的，评审意见为“不合格”" } ] } } } """ ) # 执行第二个查询 evaluation_res = qianwen_long(file_id, user_query) #有些重复的键名，只有qianwen_long_text能保留 # print(evaluation_res) # 清理和处理响应 cleaned_evaluation_res = parse_json_with_duplicates(evaluation_res) #处理重复键名的情况 result_data = process_data_based_on_key(cleaned_evaluation_res) #处理不知名外键的情况 include = ['一包', '二包', '三包', '四包', '五包'] target_values = ['技术', '设计', '实施'] updated_jsons = {} # 检查是否有外层键匹配 include 列表 if any(key for key in result_data if any(included in key for included in include)): #检查result_data中的任何键是否包含include列表中的任意一个项。 # 有匹配的项，处理这些项 for key in result_data: if any(item in key for item in include): inner_dict = result_data[key] updated_jsons[key] = combine_technical_and_business(inner_dict, target_values) #对于分包，单独对分包内的'技术评分''商务评分'作处理 else: # 没有匹配的项，对整个字典运行 updated_jsons = combine_technical_and_business(result_data, target_values) final_res=reorganize_data(updated_jsons,include) #重新组织字典，尤其是分包的情况 return final_res else: # 如果 judge 是 False，直接返回默认的技术标和商务标的结构 result_data = {} result_data['技术评分'] = '本招标文件没有技术评分！' result_data['商务评分'] = '本招标文件没有商务评分！' return result_data except Exception as e: print(f"Error in combine_evaluation_standards: {e}") # 在出错时返回默认的包含空字符串的字典 return DEFAULT_EVALUATION_REVIEW.copy() if __name__ == "__main__": start_time=time.time() truncate_file=r"C:\Users\Administrator\Desktop\fsdownload\1ca1d27d-fc21-4697-8075-9027103df030\ztbfile_evaluation_method.pdf" # truncate_file = "C:\\Users\\Administrator\\Desktop\\货物标\\output2\\2-招标文件（统计局智能终端二次招标）_evaluation_method.pdf" # truncate_file="C:\\Users\\Administrator\\Desktop\\货物标\\output2\\广水市妇幼招标文件最新（W改）_evaluation_method.pdf" # truncate_file = "C:\\Users\\Administrator\\Desktop\\fsdownload\\2d481945-1f82-45a5-8e56-7fafea4a7793\\ztbfile_evaluation_method.pdf" # truncate_file="C:\\Users\\Administrator\\Desktop\\fsdownload\\ztbfile_evaluation_method.pdf" res = combine_evaluation_standards(truncate_file) print(json.dumps(res, ensure_ascii=False, indent=4)) end_time=time.time() print("elapsed time:"+str(end_time-start_time))