272 lines
14 KiB
Python
272 lines
14 KiB
Python
# -*- encoding:utf-8 -*-
|
||
import json
|
||
import re
|
||
from PyPDF2 import PdfReader
|
||
|
||
from flask_app.general.doubao import read_txt_to_string
|
||
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
|
||
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)
|
||
|
||
# 遍历所有页面，拼接全文
|
||
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,processed_filepath):
|
||
import textwrap
|
||
import json
|
||
|
||
# 定义所有可能的键
|
||
all_possible_keys = ["技术要求", "服务要求", "商务要求", "其他要求", "技术、服务要求","总体要求","进度要求","培训要求"]
|
||
|
||
# 定义每个键对应的示例内容
|
||
example_content1 = {
|
||
"技术要求": ["相关技术要求1", "相关技术要求2"],
|
||
"服务要求": ["服务要求1", "服务要求2"],
|
||
"商务要求": {
|
||
"子因素名1": ["商务要求1"],
|
||
"子因素名2": ["商务要求2"]
|
||
},
|
||
"其他要求": {
|
||
"子因素名1": ["关于项目采购的其他要求1...", "关于项目采购的其他要求2..."],
|
||
"子因素名2": ["关于项目采购的其他要求3...", "关于项目采购的其他要求4..."]
|
||
},
|
||
"技术、服务要求": ["相关技术、服务要求内容1", "相关技术、服务要求内容2"]
|
||
}
|
||
|
||
example_content2 = {
|
||
"技术要求": {
|
||
"子因素名1": ["相关技术要求1", "相关技术要求2"],
|
||
"子因素名2": ["相关技术要求3"]
|
||
},
|
||
"服务要求": {
|
||
"子因素名1": ["相关服务要求1", "相关服务要求2"],
|
||
"子因素名2": ["相关服务要求3", "相关服务要求4"]
|
||
},
|
||
"商务要求": ["商务要求1", "商务要求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. 若章节开头位置或者采购清单中除了需要采购的货物、数量、单位之外，还有带三角▲或五角星★的描述内容（如工期要求、质保要求等商务要求），请将该部分内容提取出来，添加在键名为'商务要求'的字典的键值部分,注意请不要返回Markdown语法，必要时使用冒号':'将相关信息拼接在一起。
|
||
5. 在提取技术要求或技术、服务要求时（若有），你无需从采购清单或表格中提取货物名以及参数要求，你仅需定位到原文中大标题'技术要求'或'技术、服务要求'部分提取正文内容，若内容全在表格中，键值为空列表[]。
|
||
6. 若无相关要求，键值为[]
|
||
""" )
|
||
|
||
# 过滤 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)
|
||
# 从文件中读取内容
|
||
# full_text = read_txt_to_string(processed_filepath)
|
||
# 完整的用户查询模板，包含两份示例输出
|
||
user_query_template = f"""
|
||
{prompt_instruction}
|
||
以下为示例输出，仅供格式参考：
|
||
示例 1：
|
||
{json_example1_str}
|
||
示例 2：
|
||
{json_example2_str}
|
||
|
||
"""
|
||
# 文本内容：{full_text}
|
||
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
|
||
def get_business_requirements(procurement_path,processed_filepath):
|
||
file_id=upload_file(procurement_path)
|
||
required_keys = ["技\s*术\s*要\s*求","商\s*务\s*要\s*求", "服\s*务\s*要\s*求", "其\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,processed_filepath)
|
||
# print(user_query)
|
||
model_res=qianwen_long_stream(file_id,user_query)
|
||
# model_res=doubao_model(user_query)
|
||
# Combine and fill missing keys with default values
|
||
final_res = clean_json_string(model_res)
|
||
# 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\fsdownload\5901b181-b55f-4107-9f30-c85d607b1fa0\ztbfile_procurement.pdf"
|
||
processed_filepath=""
|
||
# file_id = upload_file(truncate_file)
|
||
res=get_business_requirements(truncate_file,"")
|
||
print(json.dumps(res, ensure_ascii=False, indent=4))
|