zbparse/flask_app/货物标/商务服务其他要求提取.py

# -*- encoding:utf-8 -*-
import json
import re
from PyPDF2 import PdfReader
import textwrap
from flask_app.general.doubao import read_txt_to_string, pdf2txt
from flask_app.general.json_utils import combine_json_results, clean_json_string
from flask_app.general.通义千问long import upload_file, qianwen_long_stream
from flask_app.货物标.截取pdf货物标版 import extract_common_header, clean_page_content
import concurrent.futures
from flask_app.general.doubao import doubao_model


# 正则表达式判断原文中是否有商务、服务、其他要求
def find_exists(truncate_file, required_keys):
    # if not truncate_file:
    #     return ["技术要求", "商务要求", "服务要求", "其他要求"]
    common_header = extract_common_header(truncate_file)  # 假设该函数已定义
    pdf_document = PdfReader(truncate_file)

    # 定义正则模式
    begin_pattern = re.compile(
        r'(?:^第[一二三四五六七八九十百千]+(?:章|部分)\s*'  # 匹配“第X章”或“第X部分”
        r'[\u4e00-\u9fff、()（）]*?'  # 匹配允许的字符
        r'(?:(?:服务|项目|商务|技术)[\u4e00-\u9fff、()（）]*?要求|'  # 匹配“服务”、“项目”、“商务”或“技术”后跟“要求”
        r'(?:采购|需求)[\u4e00-\u9fff、()（）]*?)'  # 匹配“采购”或“需求”
        r'\s*$|'  # 匹配行尾
        r'^第[一二三四五六七八九十百千]+(?:章|部分)(?!.*说明).*?'  # 匹配“第X章”后带“采购内容”等，排除“说明”
        r'(?:采购内容|采购要求|需求).*|'  # 匹配“采购内容”或“采购要求”关键词
        r'^[一二三四五六七八九十百千]+、\s*采购清单)'  # 匹配“一、采购清单”
        r'\s*$',  # 匹配行尾
        re.MULTILINE
    )
    end_pattern = re.compile(
        r'第[一二三四五六七八九1-9]+(?:章|部分)\s*[\u4e00-\u9fff、()（）]+\s*$', re.MULTILINE)

    # 只处理第一页和最后一页
    first_page = pdf_document.pages[0].extract_text() or ""
    last_page = pdf_document.pages[-1].extract_text() or ""

    # 清理页面内容
    first_page_clean = clean_page_content(first_page, common_header)
    last_page_clean = clean_page_content(last_page, common_header)

    # 在第一页寻找起始位置
    start_match = re.search(begin_pattern, first_page_clean)
    if not start_match:
        print("未找到开始模式，返回完整第一页")
        first_content = first_page_clean
    else:
        start_index = start_match.end()
        first_content = first_page_clean[start_index:]

    # 在最后一页寻找结束位置
    end_match = re.search(end_pattern, last_page_clean)
    if not end_match:
        print("未找到结束模式，返回完整最后一页")
        last_content = last_page_clean
    else:
        last_content = last_page_clean[:end_match.start()]

    # 获取中间页面的内容
    middle_content = ""
    if len(pdf_document.pages) > 2:
        for page_num in range(1, len(pdf_document.pages) - 1):
            page_text = pdf_document.pages[page_num].extract_text() or ""
            cleaned_text = clean_page_content(page_text, common_header)
            middle_content += cleaned_text + "\n"

    # 组合所有内容
    relevant_text = first_content + "\n" + middle_content + "\n" + last_content
    relevant_text = re.sub(r'\s+', ' ', relevant_text)

    # print(f"提取的内容范围:\n{relevant_text}")

    # 匹配所需的要求
    matched_requirements = []
    punctuation = r"[，。？！、；：,.?!]*"
    for req in required_keys:
        # required_keys 中的元素本身已包含 \s*，直接作为正则模式
        if re.search(req, relevant_text):
            if req == "服\s*务\s*要\s*求":
                # 提取所有包含"服务要求"的行
                lines = [line for line in relevant_text.split('\n') if re.search(req, line)]
                # 检查是否存在'技术'紧跟在'服务要求'前面（中间只有标点，标点是可选的）
                pattern = "技\s*术" + punctuation + req
                if any(re.search(pattern, line) for line in lines):
                    # 如果存在'技术'紧跟'服务要求'，添加"技术、服务要求"
                    if "技\s*术\s*、\s*服\s*务\s*要\s*求" not in matched_requirements:
                        matched_requirements.append("技\s*术\s*、\s*服\s*务\s*要\s*求")
                else:
                    # 如果不存在'技术'紧跟'服务要求'，正常添加"服务要求"
                    matched_requirements.append(req)
            else:
                matched_requirements.append(req)

    # 去除 \s*，仅返回原始关键词
    clean_requirements = [re.sub(r'\\s\*', '', req) for req in matched_requirements]

    # 判断互斥关系：如果有"技术、服务要求"，删除"技术要求"和"服务要求"
    if "技术、服务要求" in clean_requirements:
        clean_requirements = [req for req in clean_requirements if req not in ["技术要求", "服务要求"]]

    return clean_requirements


def generate_queries(truncate_file, required_keys):
    key_list = find_exists(truncate_file, required_keys)
    queries = []
    user_query_template = "这是一份货物标中采购要求部分的内容，请告诉我\"{}\"是什么，请以json格式返回结果，外层键名是\"{}\"，内层键值对中的键名是原文中的标题或者是你对相关子要求的总结，而键值需要完全与原文保持一致，不可擅自总结删减，注意你无需回答采购清单中具体设备的技术参数要求，仅需从正文部分开始提取，"
    for key in key_list:
        query_base = user_query_template.format(key, key)
        other_keys = [k for k in key_list if k != key]
        if other_keys:
            query_base += "也不需要回答\"{}\"中的内容，".format("\"和\"".join(other_keys))
        query_base += "若相关要求不存在，在键值中填'未知'。"
        queries.append(query_base)
        # print(query_base)
    return queries


def generate_template(required_keys, type=1):
    # 定义每个键对应的示例内容
    example_content1 = {
        "技术要求": ["相关技术要求1", "相关技术要求2"],
        "服务要求": ["服务要求1", "服务要求2", "服务要求3"],
        "商务要求": ["商务要求1", "商务要求2"],
        "其他要求": {
            "子因素名1": ["关于项目采购的其他要求1...", "关于项目采购的其他要求2..."],
            "子因素名2": ["关于项目采购的其他要求3..."]
        },
        "技术、服务要求": ["相关技术、服务要求内容1", "相关技术、服务要求内容2", "相关技术、服务要求内容3"]
    }
    example_content2 = {
        "技术要求": {
            "子因素名1": ["相关技术要求1", "相关技术要求2"]
        },
        "服务要求": {
            "子因素名1": ["服务要求1"],
            "子因素名2": ["服务要求2", "服务要求3"]
        },
        "商务要求": {
            "子因素名1": ["商务要求1"],
            "子因素名2": ["商务要求2"]
        },
        "其他要求": ["关于项目采购的其他要求1...", "关于项目采购的其他要求2...", "关于项目采购的其他要求3..."],
        "技术、服务要求": {
            "子因素名1": ["相关技术、服务要求内容1"],
            "子因素名2": ["相关技术、服务要求内容2", "相关技术、服务要求内容3"]
        }
    }
    # 将 required_keys 转换为集合以便于操作
    keys = set(required_keys)
    type_to_keys_map = {
        1: ["服务要求", "商务要求", "其他要求"],
        2: ["技术要求", "技术、服务要求"]
    }
    # 根据 type 获取对应的 all_possible_keys
    chosen_keys = type_to_keys_map.get(type, [])
    another_keys_list = type_to_keys_map.get(3 - type, [])  # 3 - type 将 type 1 映射到 2，反之亦然
    another_keys_str = ', '.join([f"'{key}'" for key in another_keys_list])
    # 处理互斥关系：如果 "技术要求" 和 "服务要求" 同时存在，则移除 "技术、服务要求"
    if "技术要求" in keys and "服务要求" in keys:
        keys.discard("技术、服务要求")
    # 如果 "技术、服务要求" 存在，则移除 "技术要求" 和 "服务要求"
    elif "技术、服务要求" in keys:
        keys.discard("技术要求")
        keys.discard("服务要求")

    # 确保 keys 中只包含允许的键
    keys = keys.intersection(chosen_keys)
    # 按照预定义的顺序排序键，以保持一致性
    sorted_keys = [key for key in chosen_keys if key in keys]

    # 如果没有任何键被选中，返回""
    if not sorted_keys:
        return ""

    # 生成模板的通用部分
    def generate_prompt_instruction(keys_str, outer_keys_str, another_keys_str, type):
        if type == 1:
            specific_instructions = textwrap.dedent(
                """4. 若章节开头位置或者采购清单中除了需要采购的货物、数量、单位之外，还有带三角▲或五角星★的描述内容（如工期要求、质保要求等商务要求），请将该部分内容提取出来，添加在外层键名为'商务要求'的键值部分。
    5. 在提取'服务要求'的时候，通常包含'售后、维护、培训'等要求，若原文中有这些要求，请一并提取置于'服务要求'的键值中，。
                """
            )
        else:
            specific_instructions = textwrap.dedent(
                """4. 在提取技术要求或技术、服务要求时，你无需从采购清单或表格中提取技术要求以及参数要求，你仅需定位到原文中包含'技术要求'或'技术、服务要求'关键字的标题并提取其后相关内容；若技术要求的内容全在表格中，键值为空列表[]。
    5. 在提取'技术要求'时，注意不要提取有关'售后、维护、运维、培训、质保'等要求，它们不属于'技术要求'。
                """
            )
        return textwrap.dedent(
            f"""请你根据该货物类招标文件中的采购要求部分内容，请告诉我该项目采购的{keys_str}分别是什么，请以json格式返回结果，默认情况下外层键名是{outer_keys_str}，键值为字符串列表，每个字符串表示具体的一条要求，可以按原文中的序号作划分（若有序号的话），请按原文内容回答，保留三角▲、五角星★和序号（若有），不要擅自增添内容及序号。请不要提取{another_keys_str}中的内容。

    要求与指南：
    1. JSON 的结构要求：
        - 默认情况无需嵌套键值对，键值为字符串列表；若存在嵌套结构(即有明确标题表示各子要求)，则嵌套键名是原文中该要求下相应子标题，最多一层嵌套。
        - 每个外层键对应的值可以是：
            a. 一个字符串列表，表示具体的一条条要求。若只有一条要求，也用字符串列表表示。
            b. 一个对象（字典），其键为子因素名，值为字符串列表。
        - 最多只允许一层嵌套。
    2. 请优先且准确定位正文部分包含以下关键字的标题：{outer_keys_str}，在其之后提取'XX要求'相关内容，尽量避免在无关地方提取内容。
    3. 注意请不要返回Markdown语法，必要时使用冒号':'将相关信息拼接在一起。若文档中无符合的相关要求，键值为空列表[] 
    {specific_instructions}
    """)

    # 过滤示例内容
    def filter_example_content(example_content, keys):
        return {k: v for k, v in example_content.items() if k in keys}

    def format_example(example_content):
        return json.dumps(example_content, indent=4, ensure_ascii=False)

    filtered_example_content1 = filter_example_content(example_content1, sorted_keys)
    filtered_example_content2 = filter_example_content(example_content2, sorted_keys)
    tech_json_example1_str = format_example(filtered_example_content1)
    tech_json_example2_str = format_example(filtered_example_content2)
    keys_str = '、'.join(sorted_keys)
    outer_keys_str = ', '.join([f"'{key}'" for key in sorted_keys])
    prompt_instruction = generate_prompt_instruction(keys_str, outer_keys_str, another_keys_str, type)
    # 完整的用户查询模板，包含两份示例输出
    user_query_template = f"""
    {prompt_instruction}
    以下为示例输出，仅供格式参考：
    示例 1：
    {tech_json_example1_str}
    示例 2：
    {tech_json_example2_str}
    """
    return user_query_template

def merge_requirements(input_dict):
    # 初始化一个临时字典，用于存储标准化后的键
    temp_dict = {}
    # 初始化最终字典，只包含指定的四个键
    final_keys = ['技术要求', '商务要求', '服务要求', '其他要求']
    final_dict = {key: "" for key in final_keys}

    # 如果输入字典中有'其他要求'，保留其内容
    if '其他要求' in temp_dict and temp_dict['其他要求'].strip():
        final_dict['其他要求'] = temp_dict['其他要求'].strip()

    # 处理'技术要求', '商务要求', '服务要求'
    for key in ['技术要求', '商务要求', '服务要求']:
        if key in temp_dict:
            final_dict[key] = temp_dict[key].strip()

    # 收集需要合并到'其他要求'的内容
    merge_keys = ['总体要求', '进度要求', '培训要求']
    merged_contents = []
    for key in merge_keys:
        if key in temp_dict and temp_dict[key].strip():
            merged_contents.append(temp_dict[key].strip())

    # 如果有需要合并的内容
    if merged_contents:
        merged_text = " ".join(merged_contents)
        if final_dict['其他要求']:
            final_dict['其他要求'] += " " + merged_text
        else:
            final_dict['其他要求'] = merged_text

    # 移除多余的空格
    for key in final_dict:
        final_dict[key] = final_dict[key].strip()

    return final_dict


# ,"总\s*体\s*要\s*求","进\s*度\s*要\s*求","培\s*训\s*要\s*求"
def get_business_requirements(procurement_path,procurement_docpath):
    file_id = upload_file(procurement_docpath)
    print(file_id)
    required_keys = ["技\s*术\s*要\s*求", "商\s*务\s*要\s*求", "服\s*务\s*要\s*求", "其\s*他\s*要\s*求"]
    contained_keys = find_exists(procurement_path, required_keys)
    print(contained_keys)
    if not contained_keys:
        return {}
    # queries = generate_queries(truncate_file, contained_keys)
    busi_user_query = generate_template(contained_keys, 1)
    tech_user_query = generate_template(contained_keys, 2)
    final_res={}
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        futures = []
        if busi_user_query:
            futures.append(executor.submit(qianwen_long_stream, file_id, busi_user_query, 2, 1))
        if tech_user_query:
            futures.append(executor.submit(qianwen_long_stream, file_id, tech_user_query, 2, 1))
        # 获取结果
        for future in concurrent.futures.as_completed(futures):
            try:
                result = future.result()
                if result:  # 确保结果不为空
                    final_res.update(clean_json_string(result))
            except Exception as e:
                print(f"An error occurred: {e}")
    return final_res


# TODO:改为先判断，再摘取
if __name__ == "__main__":
    # truncate_file = "C:\\Users\\Administrator\\Desktop\\fsdownload\\e4be098d-b378-4126-9c32-a742b237b3b1\\ztbfile_procurement.docx"
    truncate_file = r"C:\Users\Administrator\Desktop\货物标\output1\交警支队机动车查验监管系统项目采购_procurement.pdf"
    docx_path=r'C:\Users\Administrator\Desktop\货物标\output1\交警支队机动车查验监管系统项目采购_procurement.docx'
    # truncate_file=r"C:\Users\Administrator\Desktop\new招标文件\output5\HBDL-2024-0519-001-招标文件_procurement.pdf"
    # file_id = upload_file(truncate_file)
    processed_filepath = pdf2txt(truncate_file)
    final_res= get_business_requirements(truncate_file, docx_path)
    print(json.dumps(final_res, ensure_ascii=False, indent=4))