zbparse/flask_app/货物标/技术参数要求提取.py

# -*- encoding:utf-8 -*-
import json
import os
import re
import time

from flask_app.general.file2markdown import convert_pdf_to_markdown
from flask_app.general.多线程提问 import multi_threading
from flask_app.general.通义千问long import qianwen_long, upload_file
from flask_app.general.json_utils import clean_json_string, combine_json_results
from flask_app.货物标.截取pdf货物标版 import truncate_pdf_main
from flask_app.general.doubao import doubao_model, generate_full_user_query, pdf2txt
from flask_app.货物标.技术参数要求提取后处理函数 import postprocess, all_postprocess


def generate_key_paths(data, parent_key='', good_list=None, seen=None):
    """
    生成嵌套字典中的键路径，并提取最内层的键名。
    同时，提取特定模式的键（如 '交换机-1', '交换机-2'）的父路径。
    如果同一层级下只有'交换机-1'但没有'交换机-2'，则视为错误输入，删除后缀'-1'，将'交换机'加入key_paths。

    参数:
    data (dict): 输入的字典数据
    parent_key (str): 上级键路径，用于递归调用
    good_list (list): 用于存储去重后的最内层键名
    seen (set): 用于跟踪已添加到 good_list 的元素

    返回:
    tuple: 包含键路径列表、最内层键名列表、分组路径列表以及 no_keys_added 的元组
           (key_paths, good_list, grouped_paths, no_keys_added)
    """
    if good_list is None:
        good_list = []
    if seen is None:
        seen = set()

    key_paths = []
    grouped_paths = set()  # 使用集合避免重复路径
    no_keys_added = True  # 默认假设没有添加任何键

    # Step 1: Collect keys that match the pattern
    pattern = re.compile(r'(.+)-\d+$')  # 匹配形如 '交换机-1', '交换机-2' 的键
    prefix_groups = {}
    other_keys = []

    for key in data:
        clean_key = key.replace(" ", "")
        match = pattern.match(clean_key)
        if match:
            prefix = match.group(1)
            if prefix not in prefix_groups:
                prefix_groups[prefix] = []
            prefix_groups[prefix].append(key)
        else:
            other_keys.append(key)

    # Step 2: Handle grouped keys
    for prefix, keys in prefix_groups.items():
        current_prefix_path = f"{parent_key}.{prefix}" if parent_key else prefix
        if len(keys) > 1:
            # 多个键匹配同一前缀：添加到 grouped_paths
            grouped_paths.add(current_prefix_path)
            if prefix not in seen:
                good_list.append(prefix)
                seen.add(prefix)
            no_keys_added = False
        else:
            # 只有一个键匹配：删除后缀并添加到 key_paths
            key = keys[0]
            key_path = current_prefix_path  # 去掉后缀后，路径为父路径 + 前缀
            key_paths.append(key_path)
            if prefix not in seen:
                good_list.append(prefix)
                seen.add(prefix)
            no_keys_added = False

    # Step 3: Handle other keys
    for key in other_keys:
        value = data[key]
        current_key = f"{parent_key}.{key}" if parent_key else key

        if isinstance(value, dict):
            if value:
                # 递归调用，并获取子路径、子 good_list、子分组路径以及子 no_keys_added
                sub_key_paths, _, sub_grouped_paths, sub_no_keys_added = generate_key_paths(
                    value, current_key, good_list, seen
                )
                key_paths.extend(sub_key_paths)
                grouped_paths.update(sub_grouped_paths)  # 合并子分组路径到当前分组路径
                # 更新 no_keys_added
                no_keys_added = no_keys_added and sub_no_keys_added
            else:
                # 空字典视为叶子节点
                clean_key = key.replace(" ", "")
                key_paths.append(current_key.replace(" ", ""))
                if clean_key not in seen:
                    good_list.append(clean_key)  # 去掉空格后添加
                    seen.add(clean_key)

                # 更新 no_keys_added
                no_keys_added = False
        elif isinstance(value, list):
            # 列表类型视为叶子节点，无论是否为空
            key_paths.append(current_key.replace(" ", ""))
            clean_key = key.replace(" ", "")
            if clean_key not in seen:
                good_list.append(clean_key)  # 去掉空格后添加
                seen.add(clean_key)
            # 更新 no_keys_added
            no_keys_added = False
        elif value in {"未知", "", "/"}:
            # 特定值视为叶子节点
            key_paths.append(current_key.replace(" ", ""))
            clean_key = key.replace(" ", "")
            if clean_key not in seen:
                good_list.append(clean_key)  # 去掉空格后添加
                seen.add(clean_key)
            # 更新 no_keys_added
            no_keys_added = False
        else:
            # 其他情况视为叶子节点
            key_paths.append(current_key.replace(" ", ""))
            clean_key = key.replace(" ", "")
            if clean_key not in seen:
                good_list.append(clean_key)  # 去掉空格后添加
                seen.add(clean_key)
            # 更新 no_keys_added
            no_keys_added = False

    return key_paths, good_list, grouped_paths, no_keys_added

def combine_and_update_results(original_data, updates):
    def normalize_key(key):
        """
        规范化键名：
        - 替换全角点号为半角点号。
        - 删除所有空格（包括半角和全角）。
        """
        # 替换全角点号（．、。）为半角点号（.）
        key = key.replace('．', '.').replace('。', '.')
        # 删除所有空格（半角空格和全角空格）
        key = key.replace(' ', '').replace('\u3000', '')
        return key

    def normalize_original_data(d):
        """
        递归规范化原始数据字典的键。
        """
        if not isinstance(d, dict):
            return d
        normalized = {}
        for k, v in d.items():
            nk = normalize_key(k)
            normalized[nk] = normalize_original_data(v)
        return normalized

    def normalize_update_value(value):
        """
        递归规范化更新字典中嵌套的字典的键。
        """
        if isinstance(value, dict):
            return {normalize_key(k): normalize_update_value(v) for k, v in value.items()}
        else:
            return value

    def recursive_update(data, key, value):
        """
        递归更新嵌套字典。
        """
        keys = key.split('.')
        for k in keys[:-1]:
            data = data.setdefault(k, {})
        if isinstance(value, dict) and isinstance(data.get(keys[-1], None), dict):
            data[keys[-1]] = {**data.get(keys[-1], {}), **value}
        else:
            data[keys[-1]] = value

    # 1. 规范化原始数据字典的键
    original_data = normalize_original_data(original_data)

    # 2. 规范化更新字典的键
    normalized_updates = {}
    for key, value in updates.items():
        nk = normalize_key(key)
        nv = normalize_update_value(value)
        normalized_updates[nk] = nv

    # 3. 执行递归更新
    for key, value in normalized_updates.items():
        recursive_update(original_data, key, value)

    return original_data

#     user_query1 = """
# 请你首先定位该采购文件中的采购清单或采购需求部分，请告诉我需要采购的货物，如果有采购清单，请直接根据清单上的货物（或系统）名称给出结果，注意不要返回'说明'或'规格'或'技术参数'列中的内容；若没有采购清单，你要从表格中或文中摘取需要采购的系统和货物，采购需求中可能包含层次关系，例如采购的某系统中可能包含几种货物，那么你需要用嵌套键值对表示这种关系，且不要遗漏该系统中包含的货物，你的输出请以json格式返回，最外层键名为'采购需求'，嵌套键名为对应的系统名称或货物名称，需与原文保持一致，无需给出采购数量和单位。以下为需要考虑的特殊情况：如果采购清单中同一层级(或同一系统)下存在同名货物且它们的采购要求有所不同，请你以'货物名-编号'区分多种型号，编号为从 1 开始的自然数，依次递增,例如若采购清单中有两种型号的'交换机'，那么你应返回两个键名，'交换机-1'和'交换机-2'；如有未知内容，在对应键值处填'未知'。以下为考虑了特殊情况的示例输出：
# {
#     "采购需求": {
#         "交换机-1"：{},
#         "交换机-2":{},
#         "门禁管理系统": {},
#         "交通监控视频子系统": {
#             "高清视频抓拍像机":{},
#             "补光灯":{}
#         },
#         "LED全彩显示屏": {}
#     }
# }
#     """

#文件内容以markdown格式组织，其中表格部分（若有）以html语法组织，
def get_technical_requirements(file_path,invalid_path):
    file_id=upload_file(file_path)
    first_query_template="该文件是否说明了采购需求,即需要采购哪些货物?如果有,请回答'是',否则,回答'否'"
    # first_query=generate_full_user_query(file_path,first_query_template)
    # judge_res=doubao_model(first_query)
    judge_res=qianwen_long(file_id,first_query_template)
    prompt_template1 = '''
                    任务：解析采购文件，提取采购需求，并以JSON格式返回。

                        要求与指南：
                        1. 精准定位：请运用文档理解能力，找到文件中的采购需求部分，若有采购清单，请直接根据采购清单上的货物（或系统）名称给出结果。
                        2. 采购目标：采购目标通常有硬件（如设备、货物）和软件（如系统软件、应用APP），一次采购活动可能同时包含这两种类型。对于工程类的施工、建设采购需求，无需提取。
                        3. 非清单形式处理：若未出现采购清单，则从表格或文字中摘取采购信息。
                        4. 系统归属：一些采购活动可能将采购目标划分为若干系统和货物，每个系统可能包含若干货物，则将这些货物名称作为该系统的二级键；系统可以只包含总体'系统功能'而无货物。
                        5. 软件需求：对于软件应用或系统软件需求，仅需列出系统模块构成（若有），并作为系统键值的一部分，无需在模块下再细分功能。
                        6. 系统功能：若采购的某系统提及总体系统功能，则在系统值中添加'系统功能'二级键，不展开具体内容。
                        7. 完整性：确保不遗漏系统内的货物，也不添加未提及的内容。

                        输出格式：
                        1.JSON格式，最外层键名为'采购需求'。
                        2.层次关系用嵌套键值对表示。
                        3.嵌套键名为系统或货物或模块名称，与原文保持一致。
                        4.最内层键值应为空对象（{{}}）。
                        5.不包含'说明'、'规格'、'技术参数'等列内容，仅返回采购的货物或系统或模块名称。

                        特殊情况处理：
                        同一层级（如同一系统中）下同名但采购要求不同的货物，以'货物名-编号'区分，编号从1递增。

                        示例输出1，普通系统、货物类采购：
                        {{
                            "采购需求": {{
                                "交换机-1": {{}},
                                "交换机-2": {{}},
                                "门禁管理系统": {{
                                    "系统功能":{{}}
                                }},
                                "交通监控视频子系统": {{
                                    "系统功能": {{}},
                                    "高清视频抓拍像机": {{}},
                                    "补光灯": {{}}
                                }},
                                "LED全彩显示屏": {{}}
                                // 其他系统和货物
                            }}
                        }}
                        示例输出2，系统软件采购：
                        {{
                            "采购需求": {{
                                "信息管理系统": {{
                                    "通用模块":{{}},
                                    "用户管理":{{}}
                                }},
                                "信息检索系统": {{
                                    "系统功能":{{}},
                                    "权限管理模块":{{}}
                                }},
                                "XX小程序":{{}},
                                "数据分析中心":{{}}
                            }}
                        }}

                    注意事项：
                    1.严格按照上述要求执行，确保输出准确性和规范性。
                    2.如有任何疑问或不确定内容，请保留原文描述，必要时使用'未知'标注。
                '''
    prompt_template2 = '''
                        任务：解析采购文件，提取采购需求，并以JSON格式返回。

                        要求与指南：
                        1. 精准定位：请运用文档理解能力，找到文件中的采购需求部分，若有采购清单，请直接根据采购清单上的货物（或系统）名称给出结果。
                        2. 采购目标：采购目标通常有硬件（如设备、货物）和软件（如系统软件、应用APP），一次采购活动可能同时包含这两种类型。对于工程类的施工、建设采购需求，无需提取。
                        3. 非清单形式处理：若未出现采购清单，则从表格或文字中摘取采购信息。
                        4. 系统归属：一些采购活动可能将采购目标划分为若干系统和货物，每个系统可能包含若干货物，则将这些货物名称作为该系统的二级键；系统可以只包含总体'系统功能'而无货物。
                        5. 软件需求：对于软件应用或系统软件需求，仅需列出系统模块构成（若有），并作为系统键值的一部分，无需在模块下再细分功能。
                        6. 系统功能：若采购的某系统提及总体系统功能，则在系统值中添加'系统功能'二级键，不展开具体内容。
                        7. 完整性：确保不遗漏系统内的货物，也不添加未提及的内容。

                        输出格式：
                        1.JSON格式，最外层键名为'采购需求'。
                        2.层次关系用嵌套键值对表示。
                        3.嵌套键名为系统或货物或模块名称，与原文保持一致。
                        4.最内层键值应为空对象（{{}}）。
                        5.不包含'说明'、'规格'、'技术参数'等列内容，仅返回采购的货物或系统或模块名称。

                        特殊情况处理：
                        若同一层级（如同一系统中）下存在同名但采购要求不同的货物，以'货物名-编号'区分，编号从1递增，规避重复键名的问题；否则无需添加编号。

                        示例输出1，普通系统、货物类采购：
                        {{
                            "采购需求": {{
                                "交换机-1": {{}},
                                "交换机-2": {{}},
                                "门禁管理系统": {{
                                    "系统功能":{{}}
                                }},
                                "交通监控视频子系统": {{
                                    "系统功能": {{}},
                                    "高清视频抓拍像机": {{}},
                                    "补光灯": {{}}
                                }},
                                "LED全彩显示屏": {{}}
                                // 其他系统和货物
                            }}
                        }}
                        示例输出2，系统软件采购：
                        {{
                            "采购需求": {{
                                "信息管理系统": {{
                                    "通用模块":{{}},
                                    "用户管理":{{}}
                                }},
                                "信息检索系统": {{
                                    "系统功能":{{}},
                                    "权限管理模块":{{}}
                                }},
                                "XX小程序":{{}},
                                "数据分析中心":{{}}
                            }}
                        }}

                        文件内容：{full_text}

                        注意事项：
                        1.严格按照上述要求执行，确保输出准确性和规范性。
                        2.如有任何疑问或不确定内容，请保留原文描述，必要时使用'未知'标注。
                    '''
    if '否' in judge_res:
        file_id=upload_file(invalid_path)
        print("调用invalid_path")
        model_res=qianwen_long(file_id,prompt_template1)
        print(model_res)
    else:
        # processed_filepath = convert_pdf_to_markdown(file_path)   # 转markdown格式
        # processed_filepath=r"C:\Users\Administrator\Desktop\货物标\extract_files\107国道.txt"
        processed_filepath = pdf2txt(file_path)                   # 纯文本提取
        user_query=generate_full_user_query(processed_filepath,prompt_template2)
        model_res=doubao_model(user_query)
        # model_res = qianwen_long(file_id,prompt_template1)
        print(model_res)

    cleaned_res = clean_json_string(model_res)     #转字典
    normal_paths,good_list,grouped_paths,no_keys_added= generate_key_paths(cleaned_res['采购需求'])  # 提取需要采购的货物清单 key_list：交通监控视频子系统.高清视频抓拍像机 ... grouped_paths是同一系统下同时有'交换机-1'和'交换机-2'，提取'交换机'  ，输出eg:{'交通标志.标志牌铝板', '交通信号灯.交换机'}
    if no_keys_added:
        ffinal_res = postprocess(cleaned_res)
    else:
        # user_query_template = "请你根据该货物标中采购要求部分的内容，请你给出\"{}\"的技术参数（或采购要求），请以json格式返回结果，外层键名为\"{}\", 键值对中的键是你对该要求的总结，而值需要完全与原文保持一致，不可擅自总结删减。"
        user_query_template = """请根据货物标中采购要求部分的内容，告诉我\"{}\"的技术参数或采购要求是什么。请以 JSON 格式返回结果，键名为\"{}\"，键值为一个列表，列表中包含若干描述\"{}\"的技术参数或采购要求的字符串，内容需与原文一致，即若技术参数前存在序号也要保留，但不可擅自增添或删减。
要求与指南：
1. 如果该货物没有相关采购要求或技术参数要求，键值应为空列表。
2. 如果存在嵌套结构，且原文为Markdown 的表格语法，如'摄像机|有效像素|≥900W像素'， 请不要返回该Markdown语法，而是使用冒号':'将相关信息拼接在一起，生成一条完整且清晰的技术参数（或采购要求）描述，作为列表中的一个字符串。如"摄像机：有效像素：≥900W像素"。
3. 字符串中的内容为具体的技术参数要求或采购要求，请不要返回诸如'（1）高清录像功能'这种标题性质且不能体现要求的内容。
4. 你的键值应该全面，对于同一个单元格内的数据，尽量全面，不要遗漏。
        
### 示例输出1如下：
{{
    "摄像机控制键盘": [
        "1、支持串行 RS232/RS422 和 IP 混合控制，允许在一个控制器上使用 RS232/RS422/IP 控制单个系统中的摄像机；",
        "2、支持 2 组 RS422 串口 VISCA 协议菊花链控制 2x7 台摄像机。"
    ]
}}

### 示例输出2如下（包含嵌套结构）：
{{
    "摄像机": [
        "摄像机：有效像素：≥900W像素",
        "摄像机：最低照度：彩色≤0.001lx",
        "协议：routes 接口开放：具备；支持标准 ONVIF 协议与第三方厂家设备进行互联；支持 GB/T28181；应提供 SDK"
    ]    
}}
"""
        user_query_template_two="""请根据货物标中采购要求部分的内容，告诉我\"{}\"的技术参数或采购要求是什么。由于该货物存在多种不同的采购要求或技术参数，请逐一列出，并以 JSON 格式返回结果。请以'货物名-编号'区分多种型号，编号为从 1 开始的自然数，依次递增，即第一个键名为\"{}-1\", 键值为一个列表，列表中包含若干描述\"{}\"的技术参数（或采购要求）的字符串，需与原文完全一致，即若技术参数前存在序号也要保留，但不可擅自增添或删减。
请注意以下特殊情况：
要求与指南：
1. 如果该货物没有相关采购要求或技术参数要求，键值应为空列表。
2. 如果存在嵌套结构，且原文为Markdown 的表格语法，如'摄像机|有效像素|≥900W像素'， 请不要返回该Markdown语法，而是使用冒号':'将相关信息拼接在一起，生成一条完整且清晰的技术参数（或采购要求）描述，作为列表中的一个字符串。如"摄像机：有效像素：≥900W像素"。
3. 字符串中的内容为具体的技术参数要求或采购要求，请不要返回诸如'（1）高清录像功能'这种标题性质且不能体现要求的内容。

### 示例输出1如下：
{{
    "交换机-1": [
        "1、支持固化千兆电口≥8 个，固化千兆光口≥2 个，桌面型设备;",
        "2、支持静态链路聚合"
    ],
    "交换机-2": [
        "1、交换容量≥52Gbps，包转发率≥38.69Mpps，",
        "2、提供国家强制性产品认证证书及测试报告（3C）"
    ]
}}

### 示例输出2如下（包含嵌套结构）：
{{
    "摄像机-1": [
                "摄像机：有效像素：≥900W像素",
                "摄像机：最低照度：彩色≤0.001lx",
                "协议：routes 接口开放：具备；支持标准 ONVIF 协议与第三方厂家设备进行互联；支持 GB/T28181；应提供 SDK"
            ],
    "摄像机-2": [
        "支持夜视", "支持云存储"
    ]
}}
        """
        queries = []
        for key in normal_paths:
            # 将键中的 '.' 替换为 '下的'
            modified_key = key.replace('.', '下的')
            # 使用修改后的键填充第一个占位符，原始键填充第二个占位符
            new_query = user_query_template.format(modified_key, key, modified_key)
            queries.append(new_query)

            # 处理 grouped_paths 中的项，应用 user_query_template_two
        for grouped_key in grouped_paths:
            # 将键中的 '.' 替换为 '下的'
            modified_grouped_key = grouped_key.replace('.', '下的')
            # 使用修改后的键填充第一个占位符，原始键填充第二个占位符
            new_query = user_query_template_two.format(modified_grouped_key, grouped_key, modified_grouped_key)
            queries.append(new_query)
        results = multi_threading(queries, "", file_id, 2)
        technical_requirements = []
        if not results:
            print("errror!未获得大模型的回答！")
        else:
            # 打印结果
            for question, response in results:
                technical_requirements.append(response)
                # print(response)
        technical_requirements_combined_res = combine_json_results(technical_requirements)

        """根据所有键是否已添加处理技术要求"""
        # 更新原始采购需求字典
        final_res=combine_and_update_results(cleaned_res['采购需求'], technical_requirements_combined_res)
        ffinal_res=all_postprocess(final_res)
        # final_res = postprocess(cleaned_res)
        ffinal_res["货物列表"] = good_list
    # 输出最终的 JSON 字符串
    return {"采购需求":ffinal_res}

def test_all_files_in_folder(input_folder, output_folder):
    # 确保输出文件夹存在
    if not os.path.exists(output_folder):
        os.makedirs(output_folder)

    # 遍历指定文件夹中的所有文件
    for filename in os.listdir(input_folder):
        file_path = os.path.join(input_folder, filename)
        # 检查是否是文件
        if os.path.isfile(file_path):
            print(f"处理文件: {file_path}")
            # 调用函数处理文件
            try:
                json_result = get_technical_requirements(file_path)
                # 定义输出文件的路径
                output_file_path = os.path.join(output_folder, os.path.splitext(filename)[0] + '.json')
                # 保存JSON结果到文件
                with open(output_file_path, 'w', encoding='utf-8') as json_file:
                    json.dump(json_result, json_file, ensure_ascii=False, indent=4)
                print(f"结果已保存到: {output_file_path}")
            except Exception as e:
                print(f"处理文件 {file_path} 时出错: {e}")

#TODO:对于软件系统功能，提取偏离表的时候额外处理   包头市公安支队机动车查验监管系统招标文201907_procurement.pdf
#TODO: 提示词限制最多两层防止，嵌套过多，代码也要做处理，'系统'下的最多只有一层  eg 妇幼、陕西
# {
#     "采购需求": {
#         "查验视频监管软件系统": {
#             "视频设备管理": {
#                 "查验通道初始化": {},
#                 "视频设备绑定": {}
#             },
#             "业务视频应用": {
#                 "视频回放": {}
#             },

if __name__ == "__main__":
    start_time=time.time()
    # truncate_file="C:\\Users\\Administrator\\Desktop\\fsdownload\\469d2aee-9024-4993-896e-2ac7322d41b7\\ztbfile_procurement.docx"
    truncate_file=r"C:\Users\Administrator\Desktop\货物标\output1\招标文件(107国道)_procurement.pdf"
    # invalid_path="D:\\flask_project\\flask_app\\static\\output\\output1\\e7dda5cb-10ba-47a8-b989-d2993d34bb89\\ztbfile.pdf"
    # truncate_file="D:\\flask_project\\flask_app\\static\\output\\output1\\e7dda5cb-10ba-47a8-b989-d2993d34bb89\\ztbfile_procurement.docx"
    # output_folder="C:\\Users\\Administrator\\Desktop\\货物标\\output1\\tmp"
    # file_id = upload_file(truncate_file)
    invalid_path="C:\\Users\\Administrator\\Desktop\\fsdownload\\a110ed59-00e8-47ec-873a-bd4579a6e628\\ztbfile.pdf"
    # file_id=upload_file(truncate_file)
    res=get_technical_requirements(truncate_file,invalid_path)
    json_string = json.dumps(res, ensure_ascii=False, indent=4)
    print(json_string)
    # # input_folder = "C:\\Users\\Administrator\\Desktop\\货物标\\output1"
    # # output_folder = "C:\\Users\\Administrator\\Desktop\\货物标\\output3"
    # # test_all_files_in_folder(input_folder, output_folder)
    end_time=time.time()
    print("耗时:"+str(end_time-start_time))