zbparse/flask_app/general/投标人须知正文提取指定内容.py

# -*- encoding:utf-8 -*-
import json
import re
from flask_app.general.json_utils import clean_json_string
from flask_app.general.llm.model_continue_query import process_continue_answers
from flask_app.general.llm.通义千问long import upload_file, qianwen_long_stream

#提取两个大标题之间的内容
def extract_between_sections(data, target_values,flag=False):
    target_found = False
    extracted_data = {}
    current_section_title = ""
    section_pattern = re.compile(r'^[一二三四五六七八九十]+$')  # 匹配 "一", "二", "三" 等大标题
    current_block = {}

    file_pattern = re.compile(r'文\s*件')
    # 遍历所有键值对
    for key, value in data.items():
        # 只匹配形如 "一": "竞争性磋商响应文件" 的章节标题
        if section_pattern.match(key):           #匹配到大标题...
            if target_found:
                # 如果已经找到了符合的章节，并且遇到了另一个章节
                # 保存当前块并重置
                if current_block:
                    extracted_data[current_section_title] = current_block
                    current_block = {}
                target_found = False

            # 检查当前标题是否包含 target_values 中的任意关键词
            if any(tv in value for tv in target_values):
                if (flag and not file_pattern.search(value)) or not flag:
                    target_found = True  # 找到了目标章节，开始捕获后续内容
                    current_section_title = value  # 保存章节标题内容

        elif target_found:  # 匹配到普通序号...
            current_block[key] = value

    # 保存最后一个块（如果有的话）
    if current_block:
        extracted_data[current_section_title] = current_block

    return extracted_data

#生成无结构的数据货物标
def postprocess_formatted1(section_content):
    # print(json.dumps(section_content, ensure_ascii=False, indent=4))
    """
    将章节内容的键值对拼接成一个字符串列表，每个元素为 "key value"。

    Args:
        section_content (dict): 章节内容的键值对。

    Returns:
        list of str: 拼接后的字符串列表。
    """
    concatenated = []

    def get_level(key):
        """
        计算给定键的层级。

        Args:
            key (str): 层级键，如 "30.1.1"

        Returns:
            int: 层级深度，例如 "30.1.1" 的层级为 3
        """
        key = key.rstrip('.')
        parts = key.split('.')
        return len(parts)

    for key, value in section_content.items():
        level = get_level(key)

        # 添加缩进：一级无缩进，二级缩进4个空格，三级及以上每多一级加4个空格
        indent = ' ' * 4 * (level - 1)
        concatenated.append(f"{indent}{key} {value}")

    return concatenated

#-----------以下为直接从序号中提取，无大标题的相关函数----------

# 对于每个target_value元素，如果有完美匹配json_data中的键，那就加入这个完美匹配的键名，否则，把全部模糊匹配到的键名都加入
def find_keys_by_value(target_value, json_data):
    matched_keys = [k for k, v in json_data.items() if v == target_value]  # 首先检查 JSON 中的每个键值对，如果值完全等于目标值，则将这些键收集起来。
    if not matched_keys:
        matched_keys = [k for k, v in json_data.items() if
                        isinstance(v, str) and v.startswith(target_value)]  # 如果没有找到完全匹配的键，它会检查字符串类型的值是否以目标值开头，并收集这些键。
    return matched_keys  # eg:[3.1,3.1.1,3.1.2,3.2...]


# 定义查找以特定前缀开始的键的函数，eg:若match_keys中有3.1，那么以3.1为前缀的键都会被找出来，如3.1.1 3.1.2...
def find_keys_with_prefix(key_prefix, json_data):
    subheadings = [k for k in json_data if k.startswith(key_prefix)]
    return subheadings
def extract_json(data, target_values):
    results = {}
    for target_value in target_values:
        matched_keys = find_keys_by_value(target_value, data)
        for key in matched_keys:
            key_and_subheadings = find_keys_with_prefix(key, data)
            for subkey in key_and_subheadings:
                if "." in subkey:
                    parent_key = subkey.rsplit('.', 1)[0]
                    top_level_key = parent_key.split('.')[0] + '.'
                    # 特别处理定标相关的顶级键，确保不会重复添加其他键
                    if top_level_key not in results:
                        results[top_level_key] = target_value
                    # 添加或更新父级键
                    if parent_key not in results:
                        if parent_key in data:
                            results[parent_key] = data[parent_key]
                # 添加当前键
                results[subkey] = data[subkey]
    return results

#生成无结构的数据工程标,对提取出的若干键值对，生成外键为target_value，值为列表的新键值对
def postprocess_formatted2(data, target_values):
    """
    从输入字典中提取目标值对应的章节，并为每个子章节添加缩进。

    Args:
        data (dict): 输入的字典，键为层级编号，值为章节内容。
        target_values (list): 需要提取的目标章节名称列表。

    Returns:
        dict: 包含目标章节名称作为键，格式化后的子章节列表作为值的字典。
    """
    result = {}
    merged_sections = []
    processed_keys = set()

    # 对键进行排序以保持顺序
    sorted_keys = sorted(
        data.keys(),
        key=lambda x: [int(part) for part in x.rstrip('.').split('.') if part.isdigit()]
    )

    def get_level(key):
        """
        计算给定键的层级。

        Args:
            key (str): 层级键，如 "3.5.1"

        Returns:
            int: 层级深度，例如 "3.5.1" 的层级为 3
        """
        key = key.rstrip('.')
        parts = key.split('.')
        return len(parts)

    for key in sorted_keys:
        if key in processed_keys:
            continue  # 跳过已经处理过的键
        value = data[key]
        # 使用子字符串匹配
        if any(target in value for target in target_values):
            section_key_prefix = key if key.endswith('.') else key + '.'
            section_name = value
            section_level = get_level(key)
            subitems = []
            for sub_key in sorted_keys:
                if sub_key.startswith(section_key_prefix) and sub_key != key:
                    sub_value = data[sub_key]
                    sub_level = get_level(sub_key)
                    # 根据层级添加缩进，二级无缩进，三级开始每多一级加4个空格
                    if sub_level == section_level + 1:
                        indent = ''  # 二级层级无缩进
                    else:
                        indent = ' ' * 4 * (sub_level - section_level - 1)  # 三级及以上层级增加缩进
                    subitems.append(f"{indent}{sub_key} {sub_value}")
                    processed_keys.add(sub_key)  # 标记子键为已处理
            # 检查是否需要合并 "定标" 和 "中标" 章节
            if any(target in section_name for target in ["定标", "中标"]):
                merged_sections.extend(subitems)
            else:
                result[section_name] = subitems
            processed_keys.add(key)  # 标记主键为已处理

    # 如果存在需要合并的章节，将其合并为 "定标与中标"
    if merged_sections:
        result["定标与中标"] = merged_sections

    return result

def get_requirements_with_gpt(merged_baseinfo_path, selection):
    """
    根据 selection 的值选择相应的用户查询，并调用大模型获取要求。

    Args:
        merged_baseinfo_path (str): 无效文件的路径，用于上传。
        selection (int): 选择的类型（1、2 或 3）。

    Returns:
        dict: 大模型返回的要求结果，或错误信息。
    """
    # 上传文件并获取 file_id
    file_id = upload_file(merged_baseinfo_path)
    # 定义 selection 对应的用户查询
    user_queries = {
        # 1: """
        #     该招标文件中对投标文件的要求是什么？你需要从'编写要求'、'格式要求'、'承诺书要求'、'递交要求'四个角度来回答，其中'格式'可以从投标文件格式要求、标记要求、装订要求、文件数量要求角度说明，而不一定一致；'递交要求'可以从投标地点、投标文件交标方式、投标文件的修改与撤回角度说明，而不一定一致；请以json格式返回给我结果，外层键名分别为'编写要求'，'格式'，'承诺书要求'，'递交要求'，你可以用嵌套键值对组织回答，嵌套键名为你对相关子要求的总结，而嵌套键名应该完全与原文内容保持一致，不得擅自总结删减，如果原文中未提及相关内容，在键值中填'未知'。输出格式示例如下，内容与该文件无关：
        #        {
        #          "编写要求":"编写要求xxx",
        #          "格式要求":{
        #              "投标文件格式要求":"投标文件格式要求",
        #              "标记要求":"投标文件标记要求",
        #              "装订要求":"投标文件装订要求",
        #              "文件数量":"投标文件文件数量要求"
        #          },
        #          "承诺书要求":"承诺书要求xxx",
        #          "递交要求":{
        #              "投标地点":"投标地点",
        #              "投标文件交标方式":"交标方式",
        #              "投标文件的修改与撤回":["投标文件的修改相关要求","投标文件的撤回相关要求"]
        #          }
        #         }
        # """,
        1:"""请根据提供的招标文件内容，提取其中有关投标文件的要求、内容，并以 JSON 格式返回结果。
    格式要求：
        - 外层键名应为文中提到的有关投标文件相关要求、内容的大项概览性标题。
        - 对于每个大项下的具体要求内容，有以下两种组织方式：
            -不使用嵌套键值对：
                - 如果要求是单一内容，键值应为单独的字符串，内容必须与原文保持一致，不得总结或删减。
                - 如果要求包含多个并列内容，键值应为一个字符串列表（数组），其中每个元素都是一条子要求。
            -使用嵌套键值对。
                -嵌套键名应为原文中的具体标题或对相关子要求的简明总结。
                -最内层的键值应与原文内容保持一致，不得进行任何总结、删减或改写。默认键值是单独的字符串，如果一个子要求包含多个并列内容，键值应为一个字符串列表（数组），其中每个元素都是子要求内容。
        - 特别限制：
            - 若文件中有类似“投标文件格式要求”的小节，禁止输出原文中的表格格式示例，请仅提取并描述具体的文字部分的格式要求，而不是重现表格内容；若无类似小节，请忽略这点，也无需返回该键值对。
    表格内容处理：
        - 如果原文中对应内容以表格形式呈现，请使用 Markdown 语法准确重现该表格。
        - 表格的每一行应作为键值（字符串列表）中的一个独立字符串，保持表格结构和内容的完整性。
    禁止内容：
        - 确保所有输出内容均基于提供的实际招标文件内容，不使用任何预设的示例作为回答。
        - 预设的示例中的外层键名仅供格式参考，以文中实际内容为主。
            
    示例格式（**不要**在回答中包含此内容，仅供参考）：
    {
        "投标的语言":"投标人提交的投标文件以及投标人与集中采购机构或采购人就有关投标的所有来往函电均应使用中文。",
        "格式要求":{
                "投标文件格式要求":"投标文件格式要求",
                "标记要求":"投标文件标记要求",
                "装订要求":"投标文件装订要求",
                "文件数量":"投标文件文件数量要求"
            },
        "投标报价":[
            "投标人所提供的货物（工程或服务）均以人民币计价。",
            "应包含所有相关费用；"
        ]
        "递交要求":{
            "递交投标文件的截止日期及递交地点":["递交投标文件的截止日期xxx","递交投标文件的递交地点xxx"],
            "投标文件交标方式":"交标方式xxx",
            "投标文件的修改与撤回":["投标文件的修改相关要求xxx","投标文件的撤回相关要求xxx"]
        }
    }
        """,
        # 2: """
        #     该招标文件中开标、评标、定标要求(或磋商流程内容)是什么？你需要从'开标'、'开标异议'、'评标'、'定标'四个角度回答，其中'评标'可以从特殊情况的处置、评标办法及流程、评标委员会的组建角度来说明，'定标'可以从定标流程、履约能力的审查角度来说明，请以json格式返回给我结果，外层键名分别为'开标'、'开标异议'、'评标'、'定标'，你可以用嵌套键值对组织回答，嵌套键名为你对相关子要求的总结，而嵌套键值应该完全与原文内容保持一致，不得擅自总结删减，如果原文中未提及相关内容，在键值中填'未知'。输出格式示例如下：
        #     {
        #         "开标":"招标文件关于项目开标的要求",
        #         "开标异议":"招标文件中关于开标异议的项",
        #         "评标":{
        #             "特殊情况的处置":"因“电子交易系统”系统故障导致无法投标的，交易中心及时通知招标人，招标人视情况决定是否顺延投标截止时间。因投标人自身原因导致无法完成投标的，由投标人自行承担后果。",
        #             "评标办法及流程":"评标流程",
        #             "评标委员会的组建":"评标由招标人依法组建的评标委员会负责。评标委员会由招标人或其委托的招标代理机构熟悉相关业务的代表，以及有关技术、经济等方面的专家组成。"
        #         },
        #         "定标":{
        #             "定标流程":"定标流程",
        #             "履约能力的审查":"履约能力的审查"
        #         }
        #     }
        # """,
        2:"""该招标文件中有关开标、评标、定标、中标要求、内容(或磋商流程内容)是什么？请以 JSON 格式返回结果。
    要求与指南：
        - 外层键名应为文中提到的有关开标、评标、定标、中标要求或磋商流程内容的大项概览性标题。
        - 提取的内容应该是招标文件中有关开评定标等环节的流程性说明及要求，无需提取具体的评分细则以及资格审查细则。
        - 对于每个大项下的子要求，使用嵌套的键值对进行组织。
            -其中嵌套键名应为原文中的具体标题或对相关子要求的简明总结。
            -最内层的键值应与原文内容保持一致，不得进行任何总结、删减或改写。默认键值是单独的字符串，如果一个子要求包含多个并列内容，键值应为一个字符串列表（数组），其中每个元素都是子要求内容。
    表格内容处理：
        - 如果原文中对应内容以表格形式呈现，请使用 Markdown 语法准确重现该表格。
        - 表格的每一行应作为键值中的一个独立字符串，保持表格结构和内容的完整性。
    禁止内容：
        - 确保所有输出内容均基于提供的实际招标文件内容，不使用任何预设的示例作为回答。
        - 预设的示例中的外层键名仅供格式参考，以文中实际内容为主。
        
    示例格式（**不要**在回答中包含此内容，仅供参考）：
    {
        "开标": {
            "开标时间":"2024.10.30",
            "开标地点":"洪山区人民政府",
            "资格审查":[
                "公开招标采购项目开标结束后，采购人与集中采购机构依据法律、法规及招标文件的规定",
                "资格审查详见第四章“资格审查方法及标准”。"
            ],
        },
        "评标": {
            "特殊情况的处置": "因“电子交易系统”系统故障导致无法投标的，交易中心及时通知招标人，招标人视情况决定是否顺延投标截止时间。因投标人自身原因导致无法完成投标的，由投标人自行承担后果。",
            "评标办法及流程": "评标流程",
            "评标委员会的组建": "评标由招标人依法组建的评标委员会负责。评标委员会由招标人或其委托的招标代理机构熟悉相关业务的代表，以及有关技术、经济等方面的专家组成。"
        },
        "定标": {
            "定标流程": ["定标流程1","定标流程2"],
            "履约能力的审查": "履约能力的审查"
        }
    }
        """,
        3: """
            该招标文件中重新招标（或重新采购）、不再招标（或不再采购）、终止招标（或终止采购）的情况分别是什么？请以json格式返回给我结果，键名分别为'重新招标'，'不再招标'，'终止招标'，键值应该完全与原文内容保持一致，不得擅自总结删减，如果原文中未提及相关内容，在键值中填'未知'。
            示例输出如下，仅供格式参考：
            {
                "重新招标":"有下列情形之一的，招标人将重新招标：（1）投标截止时间止，投标人少于3个的；（2）经评标委员会评审后否决所有投标的；",
                "不再招标":"重新招标后投标人仍少于3个或者所有投标被否决的，属于必须审批或核准的工程建设项目，经原审批或核准部门批准后不再进行招标。",
                "终止招标":"未知"
            }
        """
    }

    # 根据 selection 选择相应的 user_query
    user_query = user_queries.get(selection)

    if not user_query:
        return {"error": f"无效的 selection 值: {selection}. 请选择 1、2 或 3。"}
    # 调用大模型并处理响应
    try:
        questions_to_continue = []
        qianwen_res = qianwen_long_stream(file_id, user_query,2,1,True)
        message = qianwen_res[0]
        parsed = clean_json_string(message)
        total_tokens = qianwen_res[1]
        if not parsed and total_tokens >5900:
            questions_to_continue.append((user_query, message))
            if questions_to_continue:
                continued_results = process_continue_answers(questions_to_continue, 3, file_id)
                return continued_results
        else:
            return parsed
    except Exception as e:
        return {"error": "调用大模型失败"}
def check_consecutive_chinese_numerals(data_dict):
    """
    检查字典的键名中是否存在连续三个中文数字键名。此时通常视为clause1提取失败,直接调用大模型更可靠
    返回:
    bool: 如果存在连续三个中文数字键名，返回 False；否则返回 True。
    """
    # 定义中文数字集合
    chinese_numerals = {"一", "二", "三", "四", "五", "六", "七", "八", "九", "十", "十一", "十二", "十三"}

    # 获取字典的键列表，保持插入顺序
    keys = list(data_dict.keys())

    # 初始化计数器
    count = 0

    # 遍历所有键名
    for key in keys:
        if key in chinese_numerals:
            count += 1
            if count >= 3:
                print("有三个及以上连续中文数字!")
                return False
        else:
            count = 0  # 重置计数器

    return True  # 没有找到连续三个中文数字键名
# 读取JSON数据，提取内容，转换结构，并打印结果
def extract_from_notice(merged_baseinfo_path, clause_path, type):
    """
    从公告中提取特定类型的内容。

    Args:
        merged_baseinfo_path (str): 合并后的基础信息路径。
        clause_path (str): 包含条款的JSON文件路径。
        type (int): 提取的类型。
            1: ["投标", "投标文件", "响应文件"],
            2: ["开标", "评标", "定标", "评审", "成交", "合同", "磋商","谈判","中标", "程序", "步骤"],
            3: ["重新招标、不再招标和终止招标", "重新招标", "重新采购", "不再招标", "不再采购", "终止招标", "终止采购"],
            4: ["评标"]  # 测试

    Returns:
        dict 或 str: 提取并处理后的数据，或在 `clause_path` 为空或发生错误时返回空字符串 `""`。
    """
    # 定义默认的返回结果
    DEFAULT_RESULT = ""

    # 映射 type 到 target_values
    type_target_map = {
        1: ["投标", "投标文件", "响应文件"],   #投标文件要求
        2: ["开标", "评标", "定标", "评审", "成交", "合同", "磋商","谈判","中标", "程序", "步骤"],   #开评定标流程
        3: ["重新招标、不再招标和终止招标", "重新招标", "重新采购", "不再招标", "不再采购", "终止招标", "终止采购"],
        4: ["评标"]  # 测试
    }
    # 获取对应 type 的 target_values
    flag = (type == 2)
    target_values = type_target_map.get(type)
    if not target_values:
        print(f"Error: Invalid type specified: {type}. Use 1, 2, 3, or 4.")
        return DEFAULT_RESULT

    try:
        if clause_path and clause_path.strip():
            with open(clause_path, 'r', encoding='utf-8') as file:
                data = json.load(file)
            if len(data) >= 60 and check_consecutive_chinese_numerals(data):          #默认clause中少于60条视为json提取失败！
                # 尝试使用大章节筛选
                extracted_data = extract_between_sections(data, target_values,flag)
                if extracted_data:
                    # 后处理并返回结果
                    extracted_data_concatenated = {
                        section: postprocess_formatted1(content)
                        for section, content in extracted_data.items()
                    }
                    return extracted_data_concatenated

                # 如果大章节筛选失败，尝试使用另一种筛选方法
                extracted_data = extract_json(data, target_values)
                if extracted_data:
                    # 后处理并返回结果
                    final_result = postprocess_formatted2(extracted_data, target_values)
                    return final_result

        # 如果 clause_path 为空，或者所有筛选方法均失败，调用回退函数
        final_result = get_requirements_with_gpt(merged_baseinfo_path, type)
        return final_result

    except Exception as e:
        print(f"Error occurred while processing clause_path '{clause_path}': {e}")
        return DEFAULT_RESULT

if __name__ == "__main__":
    # file_path = 'C:\\Users\\Administrator\\Desktop\\fsdownload\\3bffaa84-2434-4bd0-a8ee-5c234ccd7fa0\\clause1.json'
    merged_baseinfo_path=r"C:\Users\Administrator\Desktop\新建文件夹 (3)\废标\2025-湖北-通羊镇港口村人饮项目谈判文件.docx"
    clause_path=r"D:\flask_project\flask_app\static\output\output1\ce679bd7-a17a-4a95-bfc9-3801bd4a86e4\tmp\clause1.json"
    try:
        res = extract_from_notice(merged_baseinfo_path,"", 2)  # 可以改变此处的 type 参数测试不同的场景 #1: ["投标", "投标文件", "响应文件"],  2:开评定标
        res2 = json.dumps(res, ensure_ascii=False, indent=4)
        print(res2)
    except ValueError as e:
        print(e)