# -*- encoding:utf-8 -*- import json import re from PyPDF2 import PdfReader 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 #正则表达式判断原文中是否有商务、服务、其他要求 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) # 遍历所有页面，拼接全文 text = "" for page in pdf_document.pages: page_text = page.extract_text() or "" cleaned_text = clean_page_content(page_text, common_header) text += cleaned_text + "\n" # 匹配起始位置 start_match = re.search(begin_pattern, text) if not start_match: print("未找到开始模式") return [] start_index = start_match.end() # 匹配结束位置 end_match = re.search(end_pattern, text[start_index:]) if end_match: end_index = start_index + end_match.start() relevant_text = text[start_index:end_index] else: relevant_text = text[start_index:] # 保留换行，避免结构丢失 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_user_query_template(required_keys): import textwrap import json # 定义所有可能的键 all_possible_keys = ["技术要求", "服务要求", "商务要求", "其他要求", "技术、服务要求"] # 定义每个键对应的示例内容 example_content1 = { "技术要求": ["相关技术要求以及服务要求1", "相关技术要求以及服务要求2"], "服务要求": ["服务要求1", "服务要求2"], "商务要求": { "★产品质保期": ["所投 LED 整屏不低于 3 年，健身器材整套不低于 2 年"], "售后服务方案": ["包含产品配送、安装保障方案"] }, "其他要求": { "子因素名1": ["关于项目采购的其他要求1...", "关于项目采购的其他要求2..."], "子因素名2": ["关于项目采购的其他要求3...", "关于项目采购的其他要求4..."] }, "技术、服务要求": ["相关技术、服务要求内容1", "相关技术、服务要求内容2"] } example_content2 = { "技术要求": { "子因素名1": ["相关技术要求1", "相关技术要求2"], "子因素名2": ["相关技术要求3"] }, "服务要求": { "子因素名1": ["相关服务要求1", "相关服务要求2"], "子因素名2": ["相关服务要求3", "相关服务要求4"] }, "商务要求": ["所投 LED 整屏不低于 3 年，健身器材整套不低于 2 年", "包含产品配送、安装保障方案"], "其他要求": ["关于项目采购的其他要求1..."], "技术、服务要求": { "子因素名1": ["相关技术、服务要求内容1"], "子因素名2": ["相关技术、服务要求内容2", "相关技术、服务要求内容3"] } } # 将 required_keys 转换为集合以便于操作 keys = set(required_keys) # 处理互斥关系：如果 "技术要求" 和 "服务要求" 同时存在，则移除 "技术、服务要求" if "技术要求" in keys and "服务要求" in keys: keys.discard("技术、服务要求") # 如果 "技术、服务要求" 存在，则移除 "技术要求" 和 "服务要求" elif "技术、服务要求" in keys: keys.discard("技术要求") keys.discard("服务要求") # 确保 keys 中只包含允许的键 keys = keys.intersection(all_possible_keys) # 按照预定义的顺序排序键，以保持一致性 sorted_keys = [key for key in all_possible_keys if key in keys] # 如果没有任何键被选中，返回一个默认的模板或抛出异常 if not sorted_keys: raise ValueError("required_keys 中没有有效的键。") # 生成提示部分，根据 sorted_keys 动态构建 keys_str = '、'.join(sorted_keys) outer_keys_str = ', '.join([f"'{key}'" for key in sorted_keys]) # 使用三引号定义多行字符串，便于编辑和维护 prompt_instruction = textwrap.dedent(f"""请你根据该货物类招标文件中的采购要求部分内容，请告诉我该项目采购的{keys_str}分别是什么，请以json格式返回结果，默认情况下外层键名是{outer_keys_str}，键值为字符串列表，每个字符串表示具体的一条要求，内容需要与原文保持一致，不可擅自总结删减。 要求与指南： 1. 默认情况无需嵌套，键值为字符串列表；若存在嵌套结构，嵌套键名是原文中该要求下相应子标题，最多一层嵌套。 2. JSON 的结构要求： - 外层键名为 {outer_keys_str} 中的各项。 - 每个外层键对应的值可以是： a. 一个对象（字典），其键为子因素名，值为字符串列表。 b. 一个字符串列表，表示具体的一条条要求。若只有一条要求，也用字符串列表表示。 - 最多只允许一层嵌套。 3. 请优先定位正文部分的大标题'xx要求'，在其之后提取'xx要求'相关内容，由于要求的位置比较集中，请尽量避免在全文各处寻找。 4. 在提取技术要求或技术、服务要求时（若有），你无需从采购清单或表格中提取货物名以及参数要求，你仅需定位到原文中相应位置（正文部分、而非表格中）并提取正文内容，通常一类要求写在一块大标题下，否则，键值为空列表。 5. 若无相关要求，键值为[] """) # 过滤 example_content1 和 example_content2 以仅包含 sorted_keys def filter_content(example_content, keys): return {k: v for k, v in example_content.items() if k in keys} filtered_example_content1 = filter_content(example_content1, sorted_keys) filtered_example_content2 = filter_content(example_content2, sorted_keys) # 将过滤后的示例转换为格式化的 JSON 字符串 json_example1_str = json.dumps(filtered_example_content1, indent=4, ensure_ascii=False) json_example2_str = json.dumps(filtered_example_content2, indent=4, ensure_ascii=False) # 完整的用户查询模板，包含两份示例输出 user_query_template = f""" {prompt_instruction} 以下为示例输出，仅供格式参考： 示例 1： {json_example1_str} 示例 2： {json_example2_str} """ return user_query_template def get_business_requirements(procurement_path): file_id=upload_file(procurement_path) 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) # queries = generate_queries(truncate_file, contained_keys) user_query=generate_user_query_template(contained_keys) business_requirements=qianwen_long_stream(file_id,user_query) # Combine and fill missing keys with default values final_res = clean_json_string(business_requirements) # final_res.update({key: final_res.get(key, "") for key in required_keys}) 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\2-招标文件（广水市教育局封闭管理）_procurement.pdf" # file_id = upload_file(truncate_file) res=get_business_requirements(truncate_file) print(json.dumps(res, ensure_ascii=False, indent=4))