zbparse/flask_app/general/商务技术评分提取.py

import json
import re
import time
from collections import defaultdict

from flask_app.general.通义千问long import upload_file, qianwen_long


def remove_unknown_scores(data):
    if isinstance(data, dict):
        return {
            k: remove_unknown_scores(v)
            for k, v in data.items()
            if not (k == "评分" and v in ["未知", "/", ""])
        }
    elif isinstance(data, list):
        return [remove_unknown_scores(item) for item in data]
    else:
        return data
def combine_technical_and_business(data, target_values):
  data=remove_unknown_scores(data)
  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 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 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 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(evaluation_method_path,invalid_path,zb_type):
    # 定义默认的评审结果字典
    DEFAULT_EVALUATION_REVIEW = {
        "技术评分": "",
        "商务评分": ""
    }
    # 如果 truncate_file 是空字符串，直接返回包含空字符串的字典
    if not evaluation_method_path:
        return DEFAULT_EVALUATION_REVIEW.copy()

    def run_first_query(file_path):
        print("判断有无评分")
        # 上传文件并获取文件ID
        file_id = upload_file(file_path)

        # 定义用户查询
        query = (
            """根据该文档，你判断它是否有关于技术评分或商务评分或投标报价的具体的评分及要求如果有，返回'是'，否则返回'否'。
            要求与指南：
            1. 竞争性磋商文件通常无评分要求
            2. 评分要求主要以表格形式呈现，且会有评分因素及评分要求。
            3. 仅返回'是'或'否'，不需要其他解释或内容。
            """
        )  # 应对竞争性谈判这种无评分要求的情况

        # 执行查询
        return qianwen_long(file_id, query),file_id
    def run_second_qeury(file_id):
        # 执行 user_query 相关的逻辑
        user_query_1 = (
            """
 你是一个对招投标业务非常熟悉的专家。根据该文档中的评标办法表格，请你列出该文件的技术评分，商务评分，投标报价评分以及它们对应的具体评分要求，请以JSON格式返回结果。
 格式要求：
     请以 JSON 格式返回结果，最外层键名为 '技术评分'、'商务评分' 和 '投标报价评分'。在每大项下，用键值对表示具体评分项，键为具体的评审因素，若评审因素存在嵌套（表格中存在层级），请使用嵌套键值对表示，外层键名为主评审因素，嵌套的子评审因素作为内层键名，最内键值为列表，列表中包含描述评分及要求的字典。每个字典的键包括：
         '评分'：具体得分或定性指标（如 '合格制'），无评分时可删去'评分'键值对。
         '要求'：说明评分标准。
     若这三大项评分中存在额外信息（不属于某个评审因素，即该大项评分的整体要求），在该评分项内部新增键名为'备注'。

 要求与指南：
     1. 请首先定位评分细则的表格，不要回答有关资格审查的内容，也不要从评标办法正文中提取回答
     2. 你无需将表格的单元格内的内容进行拆分，需要将它视为一个整体
     3. '评分'的键值不能是一个范围数字，如'0-5分'，应该是一个具体数字，如'5分'，或者是一个定性的指标如'合格制'
     4. 如果该招标活动有多个包，则最外层键名为对应的包名,否则最外层键名为各大评分项
     5. 若表格中商务和技术评分混合一起，请根据实际表格内容进行准确分类。
     6. 若表中的评分大项不是这三个，请你根据语义分别映射到'技术评分'、'商务评分'、'投标报价评分'，而不必严格按照表格中的名称。
     7. 若大项的'xx评分'要求未在文中说明，则键名'xx评分'的键值设为'本项目无xx评分项'，例如"技术评分":"本项目无技术评分项" 
     
禁止内容：
1. 确保所有输出内容均基于提供的实际招标文件内容（除了最外层的三个评分大项名称），不使用任何预设的示例作为回答。
2. 不得擅自添加不属于评审因素的键名以及 `'备注'` 之外的其他键名。

 以下为示例输出，仅供格式参考：
     {
         "一包": {
             "技术评分": {
                 "实施方案":{
                     "总体实施方案":[
                         {
                             "评分":8,
                             "要求":"根据投标人总体实施方案进行评分"
                         }
                     ],
                     "项目实施要点":[
                         {
                             "评分":8,
                             "要求":"根据投标人对项目实施要点、难点进行评分"
                         }
                     ]
                 },
                 "主要监理岗位的职责": [
                     {
                         "评分": "4分",
                         "要求": "1、总监理工程师的职责全面、清晰、合理得 1.2-2分；一般的1.2分。2、其他主要监理人员及岗位的职责全面、清晰、合理得 1.2-2分；一般的 1.2分。"
                     }
                 ],
                 "备注": "技术标采用定性方式评审的， “不合格”仅限于投标文件出现违反国家强制性条文标准的情况，否则技术标评审结论为“合格” 。"
             },
             "商务评分": {
                 "控制系统内主板": [
                     {
                         "评分": "10分",
                         "要求": "所投电梯控制系统内主板为制造商原厂原品牌制造生产且为进口部件得 10分。（提供进口部件报关单及原产地证明扫描件加盖公章，否则不得分）"
                     }
                 ],
                 "制造商技术实力": [
                     {
                         "评分": "3分",
                         "要求": "一级证书得3分，二级证书得1分，其他不得分"
                     },
                     {
                         "评分": "2分",
                         "要求": "行业销量排名连续前 2 名，得 2 分，第 4-6 名得 0.5 分，其他不得分"
                     }
                 ]
             },
             "投标报价评分": {
                 "投标报价是否出现违反计价规范": [
                     {
                         "评分": "合格制",
                         "要求": "A:投标报价未违反计价规范的，评审意见为“合格”；B：投标报价违反计价规范的，评审意见为“不合格”"
                     }
                 ]
             }
         }
     }
                 """
        )
        user_query_2 = (
            """
你是一个对招投标业务非常熟悉的专家。根据该文档中的评标办法表格，请你列出该文件的技术评分，商务评分，投标报价评分以及它们对应的具体评分要求，请以JSON格式返回结果。
格式要求：
请以 JSON 格式返回结果，最外层键名为 '技术评分'、'商务评分' 和 '投标报价评分'。在每大项下，用键值对表示具体评分项，键为具体的评审因素，若评审因素存在嵌套（表格中存在层级），请使用嵌套键值对表示，外层键名为主评审因素，嵌套的子评审因素作为内层键名，最内键值为列表，列表中包含描述评分及要求的字典。每个字典的键包括：
    '评分'：具体得分或定性指标（如 '合格制'），无评分时可删去'评分'键值对。
    '要求'：说明评分标准。
若这三大项评分中存在额外信息（不属于某个评审因素，即该大项评分的整体要求），在该评分项内部新增键名为'备注'，值为该要求。

要求与指南：
1. 请首先定位评分细则的表格，不要回答有关资格审查的内容，也不要从评标办法正文中提取回答
2. 你无需将表格的单元格内的内容进行拆分，需要将它视为一个整体
3. '评分'的键值不能是一个范围数字，如'0-5分'，应该是一个具体数字，如'5分'，或者是一个定性的指标如'合格制'
4. 如果该招标活动有多个包，则最外层键名为对应的包名,否则最外层键名为各大评分项
5. 若表格中商务和技术评分混合一起，请根据你对招投标业务的熟悉，对实际表格内容的评分因素进行准确分类，归类至商务评分或技术评分。
6. 若表中的评分大项不是这三个，请你根据语义分别映射到'技术评分'、'商务评分'、'投标报价评分'，而不必严格按照表格中的名称。
7. 若大项的'xx评分'要求未在文中说明，则键名'xx评分'的键值设为'本项目无xx评分项'，例如"技术评分":"本项目无技术评分项" 

禁止内容：
1. 确保所有输出内容均基于提供的实际招标文件内容（除了最外层的三个评分大项名称），不使用任何预设的示例作为回答。
2. 不得擅自添加不属于评审因素的键名以及 `'备注'` 之外的其他键名。

以下为示例输出，仅供格式参考：
{
    "一包": {
        "技术评分": {
            "主要监理岗位的职责": [
                {
                    "评分": "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：投标报价违反计价规范的，评审意见为“不合格”"
                }
            ]
        }
    }
}
            """
        )
        # 执行第二个查询
        user_query = user_query_1 if zb_type == 1 else user_query_2
        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
    try:
        judge_res,file_id = run_first_query(evaluation_method_path)
        # 检查 judge_res 的内容
        if '是' in judge_res:
            # 执行 user_query 相关的逻辑
            return run_second_qeury(file_id)
        else:
            judge_res,file_id=run_first_query(invalid_path)   #调用
            if '是' in judge_res:
                # 执行 user_query 相关的逻辑
                return run_second_qeury(file_id)
            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\招标文件-采购类\tmp2\2024-新疆-塔城地区公安局食药环分局快检实验室项目_evaluation_method.pdf"
    evaluation_method_path = r'C:\Users\Administrator\Desktop\fsdownload\aba81749-5986-4492-8b4b-16db9c69a09d\ztbfile_evaluation_method.pdf'
    invalid_path=r'C:\Users\Administrator\Desktop\fsdownload\91399aa4-1ee8-447d-a05b-03cd8d15ced5\ztbfile_invalid.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(evaluation_method_path,invalid_path,2)
    print(json.dumps(res, ensure_ascii=False, indent=4))
    end_time=time.time()
    print("elapsed time:"+str(end_time-start_time))