770 lines
39 KiB
Python
770 lines
39 KiB
Python
import json
|
||
import os
|
||
import re
|
||
import time
|
||
from copy import deepcopy
|
||
|
||
from flask_app.general.format_change import docx2pdf,doc2docx
|
||
from flask_app.general.json_utils import clean_json_string, rename_outer_key
|
||
from flask_app.general.merge_pdfs import merge_pdfs
|
||
from flask_app.general.llm.qianwen_plus import qianwen_plus
|
||
from flask_app.general.通用功能函数 import get_global_logger
|
||
from flask_app.general.截取pdf_main import truncate_pdf_multiple
|
||
from flask_app.货物标.提取采购需求main import fetch_procurement_reqs
|
||
from flask_app.货物标.技术参数要求提取后处理函数 import postprocess_technical_table
|
||
from flask_app.货物标.资格审查main import combine_qualification_review
|
||
from flask_app.general.商务技术评分提取 import combine_evaluation_standards
|
||
import concurrent.futures
|
||
|
||
logger = None
|
||
def get_nested(dic, keys, default=None):
|
||
for key in keys:
|
||
if isinstance(dic, dict):
|
||
dic = dic.get(key, default)
|
||
else:
|
||
return default
|
||
return dic
|
||
|
||
def prepare_for_zige_info(zige_review):
|
||
try:
|
||
zige_info = ""
|
||
fuhe_info = ""
|
||
zigefuhe_info = ""
|
||
# 检查是否存在"资格性和符合性审查"
|
||
if "资格性和符合性审查" in zige_review:
|
||
# 情况3：只有"申请人资格要求"和"资格性和符合性审查"
|
||
if zige_review.get("申请人资格要求", {}) or zige_review.get("资格性和符合性审查", {}):
|
||
zigefuhe_info = json.dumps({
|
||
"申请人资格要求": zige_review.get("申请人资格要求", {}),
|
||
"符合性审查": zige_review.get("资格性和符合性审查", {})
|
||
}, ensure_ascii=False, indent=4)
|
||
else:
|
||
# 情况1和2：存在分开的资格审查和符合性审查
|
||
if zige_review.get("申请人资格要求", {}) or zige_review.get("资格性审查", {}):
|
||
zige_info = json.dumps({
|
||
"申请人资格要求": zige_review.get("申请人资格要求", {}),
|
||
"资格性审查": zige_review.get("资格性审查", {})
|
||
}, ensure_ascii=False, indent=4)
|
||
|
||
# 检查符合性审查的键值是否为空
|
||
fuhe_key = "符合性审查" if "符合性审查" in zige_review else "符合性审查(以下情况不得出现)"
|
||
if zige_review.get(fuhe_key, {}):
|
||
fuhe_info = json.dumps({
|
||
"符合性审查": zige_review.get(fuhe_key, {})
|
||
}, ensure_ascii=False, indent=4)
|
||
|
||
return zige_info, fuhe_info, zigefuhe_info
|
||
except KeyError as e:
|
||
print(f"缺少关键字: {e}")
|
||
# 异常时直接返回空字符串
|
||
return "", "", ""
|
||
def extract_zige_deviation_table(zige_info, fuhe_info, zigefuhe_info):
|
||
prompt_template1 = """
|
||
文本内容：{full_text}
|
||
任务：给出一份文本，根据文本提取资格性检查的具体评审标准。
|
||
输出要求：
|
||
1.以json格式返回结果，不要输出其他内容。
|
||
2.键名为"资格性检查"，键值为字符串列表，每个字符串为一条评审标准，评审标准不分先后，不要有序号标注。
|
||
要求与指南：
|
||
1. 评审标准是具体的内容，不要返回诸如'本项目的特定资格要求：'这种标题性质且不能体现具体评审标准的内容。
|
||
2. 若文本中存在相同或相似的表述，仅需取其中一个作为键值中的一条即可。
|
||
"""
|
||
prompt_template2 = """
|
||
文本内容：{full_text}
|
||
任务：给出一份文本，根据文本提取符合性检查的具体评审标准。
|
||
输出要求：
|
||
1.以json格式返回结果，不要输出其他内容。
|
||
2.键名为"符合性检查"，键值为字符串列表，每个字符串为一条评审标准，评审标准不分先后，不要有序号标注。
|
||
3.仔细检查你所选取的标准，若发现这些标准实际上是在描述不允许出现的符合性审查情况，则将外键替换为'符合性检查(以下情况不得出现)'，并将这些标准写入其中。
|
||
要求与指南：
|
||
1. 评审标准应该是具体的内容，不要返回诸如'本项目的特定符合性要求：'这种标题性质且不能体现具体评审标准的内容。
|
||
2. 若文本中存在相同或相似的表述，仅需取其中一个作为键值中的一条即可。
|
||
输出示例1：
|
||
{{
|
||
"符合性检查": [
|
||
"因素1",
|
||
"因素2",
|
||
...
|
||
]
|
||
}}
|
||
输出示例2：
|
||
{{
|
||
"符合性检查(以下情况不得出现)": [
|
||
"因素1",
|
||
"因素2",
|
||
...
|
||
]
|
||
}}
|
||
"""
|
||
prompt_template3 = """
|
||
文本内容：{full_text}
|
||
任务：给出一份文本，根据文本提取资格性检查和符合性检查的具体评审标准。
|
||
输出要求：
|
||
1.以json格式返回结果，不要输出其他内容。
|
||
2.键名为"资格性和符合性检查"，键值为字符串列表，每个字符串为一条评审标准，评审标准不分先后，不要有序号标注。
|
||
要求与指南：
|
||
1. 评审标准应该是具体的内容，不要返回诸如'本项目的特定符合性要求：'这种标题性质且不能体现具体评审标准的内容。
|
||
2. 若文本中存在相同或相似的表述，仅需取其中一个作为键值中的一条即可。
|
||
"""
|
||
def get_model_response(query):
|
||
return qianwen_plus(query)
|
||
|
||
result = {"资格审查": {}}
|
||
|
||
if zigefuhe_info:
|
||
# 如果zigefuhe_info非空，使用prompt_template3
|
||
user_query3 = prompt_template3.format(full_text=zigefuhe_info)
|
||
model_res3 = get_model_response(user_query3)
|
||
zigefuhe_deviation = clean_json_string(model_res3)
|
||
result["资格审查"] = zigefuhe_deviation
|
||
else:
|
||
zige_deviation = {}
|
||
fuhe_deviation = {}
|
||
# 提交 zige_info 和 fuhe_info 的模型调用
|
||
with concurrent.futures.ThreadPoolExecutor() as executor:
|
||
futures = {}
|
||
if zige_info != "":
|
||
user_query1 = prompt_template1.format(full_text=zige_info)
|
||
futures["zige"] = executor.submit(get_model_response, user_query1)
|
||
if fuhe_info != "":
|
||
user_query2 = prompt_template2.format(full_text=fuhe_info)
|
||
futures["fuhe"] = executor.submit(get_model_response, user_query2)
|
||
# 获取结果
|
||
for key, future in futures.items():
|
||
try:
|
||
model_res = future.result()
|
||
if key == "zige":
|
||
zige_deviation = clean_json_string(model_res)
|
||
elif key == "fuhe":
|
||
fuhe_deviation = clean_json_string(model_res)
|
||
except Exception as e:
|
||
print(f"Error processing {key}: {e}")
|
||
# 合并结果
|
||
result["资格审查"] = {
|
||
"资格性检查": zige_deviation.get("资格性检查", zige_deviation),
|
||
"符合性检查": fuhe_deviation.get("符合性检查", fuhe_deviation),
|
||
}
|
||
return result
|
||
def extract_business_deviation(busi_requirements_dict):
|
||
# 默认返回值
|
||
default_return = ({"商务要求": []}, {"商务要求带星": []})
|
||
|
||
if not busi_requirements_dict:
|
||
# 如果字典为空，返回默认字典
|
||
return default_return
|
||
|
||
def update_keys_and_values(requirements_dict):
|
||
"""
|
||
对输入的要求字典进行处理：
|
||
- 去除以特殊符号开头的键名中的符号。
|
||
- 将该符号添加到对应值（字符串列表）中每个字符串的开头。
|
||
|
||
参数：
|
||
- requirements_dict (dict): 包含服务要求、商务要求、其他要求的大字典。
|
||
|
||
返回：
|
||
- dict: 更新后的要求字典。
|
||
"""
|
||
# 定义特殊符号列表
|
||
special_symbols = ('▲', '★', '●', '■', '◆', '☆', '△', '◇', '○', '□', '#')
|
||
|
||
def process_dict(d, parent_symbol=''):
|
||
"""
|
||
递归处理字典中的键和值，去除特殊符号并在值中添加符号前缀。
|
||
|
||
参数：
|
||
- d (dict): 要处理的字典。
|
||
- parent_symbol (str): 从父级传递下来的符号（如果有）。
|
||
|
||
返回：
|
||
- dict: 处理后的字典。
|
||
"""
|
||
updated = {}
|
||
for key, value in d.items():
|
||
current_symbol = parent_symbol
|
||
new_key = key
|
||
# 检查键名是否以特殊符号开头
|
||
if key and key[0] in special_symbols:
|
||
current_symbol = key[0]
|
||
new_key = key[1:].strip() # 去除符号并去除多余空格
|
||
else:
|
||
# 保持父级符号（如果有）
|
||
current_symbol = parent_symbol
|
||
|
||
# 处理值
|
||
if isinstance(value, dict):
|
||
# 递归处理嵌套字典，传递当前符号
|
||
new_value = process_dict(value, current_symbol)
|
||
elif isinstance(value, list):
|
||
if current_symbol:
|
||
# 为列表中的每个字符串添加符号前缀
|
||
new_value = [f"{current_symbol}{item}" for item in value]
|
||
else:
|
||
new_value = value
|
||
else:
|
||
new_value = value # 保持原样
|
||
|
||
updated[new_key] = new_value
|
||
return updated
|
||
|
||
# 深复制输入字典，避免修改原始数据
|
||
requirements_copy = deepcopy(requirements_dict)
|
||
|
||
# 递归处理整个字典
|
||
updated_requirements = process_dict(requirements_copy)
|
||
|
||
return updated_requirements
|
||
|
||
# 定义辅助函数,应对'★3、服务要求'这种键名，保留'★'
|
||
def find_matching_key(data, target_key):
|
||
for k in data:
|
||
if target_key in k:
|
||
return k
|
||
return None
|
||
|
||
def rename_outer_keys(updated_requirements):
|
||
"""
|
||
重命名 updated_requirements 的顶层键，将其改为 '商务要求-d'，其中 d 是从1开始的自然数。
|
||
忽略值为 [] 的键值对。
|
||
|
||
参数：
|
||
- updated_requirements (dict): 需要重命名顶层键的字典。
|
||
|
||
返回：
|
||
- dict: 重命名后的字典。
|
||
"""
|
||
renamed = {}
|
||
counter = 1 # 计数器从1开始
|
||
|
||
for original_key, value in updated_requirements.items():
|
||
# 检查是否需要忽略该键值对
|
||
if isinstance(value, list) and not value:
|
||
continue # 忽略值为 [] 的键值对
|
||
# 构造新的键名
|
||
new_key = f"商务要求-{counter}"
|
||
renamed[new_key] = value
|
||
counter += 1 # 递增计数器
|
||
|
||
return renamed
|
||
|
||
def process_query(query):
|
||
model_res = qianwen_plus(query)
|
||
return clean_json_string(model_res)
|
||
|
||
# 优化查找逻辑
|
||
target_keys = ["服务要求", "商务要求", "其他要求"]
|
||
requirements_dict = {
|
||
matched_key: busi_requirements_dict[matched_key]
|
||
for key in target_keys
|
||
for matched_key in [find_matching_key(busi_requirements_dict, key)]
|
||
if matched_key is not None
|
||
}
|
||
|
||
# 判断是否提取到了内容
|
||
if all(value == [] for value in requirements_dict.values()):
|
||
# 如果所有值都是 "未提供"，则表示未提取到内容
|
||
print("未提取到任何内容")
|
||
return default_return
|
||
else:
|
||
updated_requirements = update_keys_and_values(requirements_dict)
|
||
print("商务要求已更新！")
|
||
renamed_requirements = rename_outer_keys(updated_requirements)
|
||
business_requirements_string = json.dumps(renamed_requirements, ensure_ascii=False, indent=4)
|
||
# print(business_requirements_string)
|
||
prompt_template1 = """
|
||
文本内容：{full_text}
|
||
以上文本是项目采购需求的商务要求部分，请帮我将信息重新组织，键名为'商务要求'，键值为字符串列表，其中每个字符串为一条商务要求，保留三角▲、五角星★（若有），但是去除开头的序号（若有）。
|
||
**角色**
|
||
你是一个专业的招投标业务专家，擅长从招标文件中总结商务要求的部分，并逐条列出，作为编写商务要求偏离表的前置准备。
|
||
|
||
**要求与指南**：
|
||
1. 每条内容需要有实际的含义、要求，不能光有标题性质的表述如'售后服务期限（质保期）及要求'。
|
||
2. 你的回答内容需从所给文本中整理，尽量不改变原文的表达（除非以下要求与指南3.的特殊情况），请勿擅自添加三角▲、五角星★。
|
||
3. 若输入文本中存在嵌套键值对格式，且键值本身语义完整且符合'商务要求'，可直接将其添加至'商务要求'的键值中；若键值字符串本身语义表达不完整、不明确，可将键值对用冒号'：'拼接之后作为一条商务要求，若键值字符串以★、▲或其他特殊符号开头，将符号移至拼接后的开头，例如'"★交货地点：采购人指定地点"'。
|
||
4. 对于以三角▲或五角星★或其他特殊符号开头的字符串：
|
||
a. 如果该字符串仅为标题性质的表述且不具备实际商务要求的含义，请根据语义关联性将其开头的三角▲或五角星★添加到紧随其后的若干（可为一）内容之后，形成完整的商务要求，并确保整个内容连贯。
|
||
注：默认在该字符串后面的一个字符串开头添加三角▲或五角星★，若有明确的序号或者语义表示了其后若干字符串之间的相关性，那么可在这些字符串开头都添加三角▲或五角星★，作为若干商务要求。
|
||
- 示例输入：
|
||
```
|
||
"★售后服务"
|
||
"提供高质量的售后服务，服务期限不少于两年。"
|
||
```
|
||
- 示例输出：
|
||
```
|
||
"★提供高质量的售后服务，服务期限不少于两年。"
|
||
```
|
||
b. 如果该字符串已经包含实际的商务要求，那么该内容作为一条完整的商务要求，保留开头的三角▲或五角星★。
|
||
- 示例输入：
|
||
```
|
||
"★提供高质量的售后服务，服务期限不少于两年。"
|
||
```
|
||
- 示例输出：
|
||
```
|
||
"★提供高质量的售后服务，服务期限不少于两年。"
|
||
```
|
||
c. 无论哪种情况，都需确保不遗漏任何以三角▲或五角星★开头的重要信息。
|
||
5. 若无商务要求，键值为空列表，即[]
|
||
|
||
**禁止内容**：
|
||
确保所有输出内容均基于提供的实际文本内容，不使用任何预设的示例作为回答。
|
||
|
||
**示例输入如下**：
|
||
{{
|
||
"商务要求-1": ["▲(1)整个平台运行运维服务，须安排人员驻场对平台进行运行维护，采用 4人轮流值班，依照 7×12小时对可视化督察巡控平台进行操作，确保平台稳定运行。","▲ （一） 投标人","1.投标人需要获得 ISO9001 质量管理体系认证 、ISO 14001 环境管理体系认证及 OHSAS18001 职业健康安全管理体系认证。","2.投标人具备网络运营商资格。"]
|
||
"商务要求-2": {{
|
||
"合同履行期限": ["★交货期（工期）：合同签订之日起 15个日历天内完成，并通过项目验收。"],
|
||
"交货地点": ["采购人指定地点"],
|
||
"报价方式": ["（1）本项目报价须为固定总价，包含但不限于：采购、实施、调试、试运行、验收、运维等所有完成本项目相关的一切费用。","（2)因投标人自身原因造成漏报、少报皆由其自行承担责任，采购人不再补偿。"],
|
||
"其他要求": ["无。"]
|
||
}}
|
||
}}
|
||
**对应的参考输出如下**：
|
||
{{
|
||
"商务要求":[
|
||
"▲整个平台运行运维服务，须安排人员驻场对平台进行运行维护，采用 4人轮流值班，依照 7×12小时对可视化督察巡控平台进行操作，确保平台稳定运行。",
|
||
"▲投标人 获得 ISO9001 质量管理体系认证 、ISO 14001 环境管理体系认证及 OHSAS18001 职业健康安全管理体系认证。",
|
||
"▲投标人具备网络运营商资格"
|
||
"★交货期（工期）：合同签订之日起 15个日历天内完成，并通过项目验收。",
|
||
"交货地点：采购人指定地点",
|
||
"本项目报价须为固定总价，包含但不限于：采购、实施、调试、试运行、验收、运维等所有完成本项目相关的一切费用。",
|
||
"因投标人自身原因造成漏报、少报皆由其自行承担责任，采购人不再补偿。"
|
||
]
|
||
}}
|
||
"""
|
||
user_query1 = prompt_template1.format(full_text=business_requirements_string)
|
||
prompt_template2 = """
|
||
文本内容：{full_text}
|
||
以上文本是项目采购需求的商务要求部分。请从中提取以★、▲或其他特殊符号开头的要求项，它们一般是重要的商务要求，需要额外响应。返回结果应仅包含一个键名“重要商务要求”，其键值为字符串列表，每个字符串对应一个以★、▲或特殊符号开头的要求项，但是去除开头的序号（若有）。
|
||
|
||
**要求与指南**：
|
||
1. 每个以★、▲或其他特殊符号开头的要求项应作为单独的字符串。
|
||
2. 每条内容需要有实际的含义、要求，不能光有标题性质的表述如'★售后服务期限（质保期）及要求'；对于类似'技术要求中带★条款项不满足的视为无效投标'这种描述带星★或带三角▲或带特殊符号的响应情况的，它本身不属于具体的★、▲商务要求，因此不需要添加进字符串列表中。
|
||
3. 你的回答内容需从所给文本中提取，尽量不改变原文的表达（除非以下要求与指南3.的特殊情况），请勿擅自添加三角▲、五角星★，请勿返回普通的商务要求。
|
||
4. 若无重要商务要求（即★、▲或其他特殊符号开头的商务要求项），键值为空列表，即[]，无需额外返回说明性文字。
|
||
5. 若输入文本中存在嵌套键值对格式，且键值本身语义完整且符合'重要商务要求'，可直接将其添加至'商务要求'的键值中；若键值字符串本身语义表达不完整，可将键值对用冒号'：'拼接之后作为一条商务要求，若键值字符串以★、▲或其他特殊符号开头，将符号移至拼接后的开头，例如'"★交货地点：采购人指定地点"'。
|
||
6. 对于以三角▲或五角星★或其他特殊符号开头的字符串：
|
||
a. 如果该字符串仅为标题性质的表述且不具备实际商务要求的含义，请根据语义关联性将其开头的三角▲或五角星★添加到紧随其后的若干（可为一）内容之后，形成完整的商务要求，并确保整个内容连贯。
|
||
注：默认在该字符串后面的一个字符串开头添加三角▲或五角星★，若有明确的序号或者语义表示了其后若干字符串之间的相关性，那么可在这些字符串开头都添加三角▲或五角星★，作为若干商务要求。
|
||
- 示例输入：
|
||
```
|
||
"★售后服务期限（质保期）及要求"
|
||
"提供高质量的售后服务，服务期限不少于两年。"
|
||
```
|
||
- 示例输出：
|
||
```
|
||
"★提供高质量的售后服务，服务期限不少于两年。"
|
||
```
|
||
b. 如果该字符串已经包含实际的商务要求，那么该内容作为一条完整的商务要求，保留开头的三角▲或五角星★。
|
||
- 示例输入：
|
||
```
|
||
"★提供高质量的售后服务，服务期限不少于两年。"
|
||
```
|
||
- 示例输出：
|
||
```
|
||
"★提供高质量的售后服务，服务期限不少于两年。"
|
||
```
|
||
c. 无论哪种情况，都需确保不遗漏任何以三角▲或五角星★或其他特殊符号开头的重要信息。
|
||
|
||
**禁止内容**：
|
||
确保所有输出内容均基于提供的实际文本内容，不使用任何预设的示例作为回答，禁止编造回答。
|
||
|
||
**示例输入如下**：
|
||
{{
|
||
"商务要求-1": ["▲(1)整个平台运行运维服务，须安排人员驻场对平台进行运行维护，采用 4人轮流值班，依照 7×12小时对可视化督察巡控平台进行操作，确保平台稳定运行。","▲ （一） 投标人","1.投标人需要获得 ISO9001 质量管理体系认证 、ISO 14001 环境管理体系认证及 OHSAS18001 职业健康安全管理体系认证。","2.投标人具备网络运营商资格。"]
|
||
"商务要求-2": {{
|
||
"合同履行期限": ["★交货期（工期）：合同签订之日起 15个日历天内完成，并通过项目验收。"],
|
||
"交货地点": ["★采购人指定地点"],
|
||
"报价方式": ["（1）本项目报价须为固定总价，包含但不限于：采购、实施、调试、试运行、验收、运维等所有完成本项目相关的一切费用。","（2)因投标人自身原因造成漏报、少报皆由其自行承担责任，采购人不再补偿。"],
|
||
"其他要求": ["无。"]
|
||
}}
|
||
}}
|
||
**对应的参考输出如下**：
|
||
{{
|
||
"重要商务要求":[
|
||
"▲整个平台运行运维服务，须安排人员驻场对平台进行运行维护，采用 4人轮流值班，依照 7×12小时对可视化督察巡控平台进行操作，确保平台稳定运行。",
|
||
"▲投标人 获得 ISO9001 质量管理体系认证 、ISO 14001 环境管理体系认证及 OHSAS18001 职业健康安全管理体系认证。",
|
||
"▲投标人具备网络运营商资格"
|
||
"★交货期（工期）：合同签订之日起 15个日历天内完成，并通过项目验收。",
|
||
"★交货地点：采购人指定地点"
|
||
]
|
||
}}
|
||
"""
|
||
user_query2 = prompt_template2.format(full_text=business_requirements_string)
|
||
queries = [user_query1, user_query2]
|
||
# 初始化结果列表，默认值为 None
|
||
results = [None] * len(queries)
|
||
# 使用 ThreadPoolExecutor 进行多线程处理
|
||
with concurrent.futures.ThreadPoolExecutor() as executor:
|
||
# 提交所有查询任务，并记录对应的索引/
|
||
futures = {executor.submit(process_query, query): idx for idx, query in enumerate(queries)}
|
||
# 遍历完成的任务
|
||
for future in concurrent.futures.as_completed(futures):
|
||
idx = futures[future]
|
||
try:
|
||
result = future.result()
|
||
results[idx] = result
|
||
except Exception as e:
|
||
print(f"查询 {idx + 1} 处理时出错: {e}")
|
||
# 仅将失败的查询结果设为默认值
|
||
results[idx] = default_return[idx]
|
||
# 分配结果到相应的变量，未处理的结果使用默认值
|
||
business_req_deviation = results[0] if results[0] is not None else default_return[0]
|
||
business_star_req_deviation = results[1] if results[1] is not None else default_return[1]
|
||
business_star_req_deviation=rename_outer_key(business_star_req_deviation,"商务要求带星")
|
||
return business_req_deviation, business_star_req_deviation
|
||
|
||
def get_tech_star_deviation(tech_requirements_dict):
|
||
def get_tech_star_deviation_directly(tech_dict):
|
||
"""
|
||
清理要求项开头的序号并python代码提取以特殊符号开头的要求项。
|
||
|
||
参数:
|
||
tech_dict (dict): 输入的技术要求字典。
|
||
|
||
返回:
|
||
dict: 返回清理后的字典，仅包含以特殊符号开头的要求项。
|
||
"""
|
||
# 定义用于清理序号的正则表达式
|
||
clean_pattern = (r'^\s*(?:[（(]\s*\d+\s*[)）)]|'
|
||
r'[A-Za-z]?\d+(?:\.\s*\d+)+[\s*\.、．)）]?|' # 12.1
|
||
r'[一二三四五六七八九十]+、|'
|
||
r'[A-Za-z]\s*[)）\.、．]?\s*)|'
|
||
r'\d+\s*[)）\.、．]?\s*|'
|
||
r'[\s*\.、．)）]?')
|
||
|
||
# 定义特殊符号的正则模式
|
||
special_symbols_pattern = re.compile(r'^[#★▲●■◆☆△◇○□]')
|
||
|
||
# 初始化结果字典
|
||
result = {}
|
||
|
||
for key, requirements in tech_dict.items():
|
||
cleaned_requirements = []
|
||
|
||
# 清理每个要求项的开头序号
|
||
for req in requirements:
|
||
cleaned_req = re.sub(clean_pattern, '', req).strip()
|
||
cleaned_requirements.append(cleaned_req)
|
||
|
||
# 过滤以特殊符号开头的要求项
|
||
special_requirements = [req for req in cleaned_requirements if special_symbols_pattern.match(req)]
|
||
|
||
if special_requirements:
|
||
result[key] = special_requirements
|
||
|
||
return result
|
||
|
||
def get_tech_star_deviation_model(tech_dict):
|
||
"""
|
||
使用大模型提取带有特殊符号开头的要求项。
|
||
|
||
参数:
|
||
tech_requirements_dict (dict): 输入的技术要求字典。
|
||
|
||
返回:
|
||
dict: 返回提取后的技术要求字典。
|
||
"""
|
||
if not tech_dict:
|
||
return {}
|
||
tech_string = json.dumps(tech_dict, ensure_ascii=False, indent=4)
|
||
prompt_template = """
|
||
文本内容：{full_text}
|
||
以上输入文本包含采购标的的技术参数要求或采购要求。请从每个键对应的字符串列表中提取带有星★或三角▲或其他特殊符号开头的的要求项。返回结果仅为符合要求的 JSON 格式对象，每个键名保持不变，键值为包含该键名下的带星或带三角或以其他特殊符合开头的要求项的字符串列表。
|
||
要求与指南：
|
||
1. 仅保留符合条件的键值对：如果某键名下没有任何以星号（★）、三角符号（▲）或其他特殊符号开头的要求项，则该键名及其对应内容不包含在输出结果中。
|
||
2. 逐条提取：每个以星号（★）、三角符号（▲）或其他特殊符号开头的要求项，均作为独立的字符串保存在对应键的值列表中。
|
||
3. 忽略序号：如果星号（★）、三角符号（▲）或其他特殊符号前带有序号（例如：1.、(1) 等），提取时需忽略这些序号，仅保留星号（★）、三角符号（▲）或特殊符号及其后内容。
|
||
4. 返回空字典：如果输入文本中所有键名下均无符合条件的要求项，则直接返回一个空字典 {{}}，无需返回其他任何说明性文本。
|
||
|
||
### 示例输入1如下：
|
||
{{
|
||
"控制键盘": [
|
||
"普通要求xx",
|
||
"#特殊符号要求xx"
|
||
],
|
||
"摄像机": [
|
||
"★带星要求xx",
|
||
"▲带三角要求xx",
|
||
"普通要求xx"
|
||
],
|
||
"交换机": [
|
||
"普通要求xx",
|
||
"普通要求xxx"
|
||
]
|
||
}}
|
||
### 对应的输出如下：
|
||
{{
|
||
"控制键盘": [
|
||
"#特殊符号要求xx"
|
||
],
|
||
"摄像机": [
|
||
"★带星要求xx",
|
||
"▲带三角要求xx"
|
||
]
|
||
}}
|
||
|
||
### 示例输入2如下：
|
||
{{
|
||
"控制键盘": [
|
||
"普通要求xx",
|
||
"普通要求xxx"
|
||
],
|
||
"摄像机": [
|
||
"普通要求xx"
|
||
]
|
||
}}
|
||
### 对应的输出如下：
|
||
{{}}
|
||
"""
|
||
user_query = prompt_template.format(full_text=tech_string)
|
||
# 调用模型接口，假设qianwen_plus是已定义的模型调用函数
|
||
model_res = qianwen_plus(user_query)
|
||
# 假设clean_json_string是已定义的JSON清理函数
|
||
tech_star_deviation = clean_json_string(model_res)
|
||
# 过滤键值为空的键
|
||
filtered_dict = {key: value for key, value in tech_star_deviation.items() if value}
|
||
return filtered_dict
|
||
|
||
if len(tech_requirements_dict) > 20: #暂时设置如果采购货物数量>20 就脚本提取带星要求，暂未考虑总字符串长度
|
||
return get_tech_star_deviation_directly(tech_requirements_dict)
|
||
else:
|
||
return get_tech_star_deviation_model(tech_requirements_dict)
|
||
|
||
def get_proof_materials(all_data_info):
|
||
prompt_template = """
|
||
文本内容：{full_text}
|
||
以上文本是从招标文件中摘取的资格审查、采购需求、商务条款、技术评分相关内容。请根据这些内容，提取并列出投标人需要提交的证明材料。
|
||
格式要求：
|
||
请以 JSON 格式返回结果：
|
||
- 键名为 '证明材料'。
|
||
- 键值为字符串列表，其中每个字符串表示投标人需要提交的一份材料。
|
||
|
||
要求与指南：
|
||
1. 仅提取与投标人需要提交的材料相关的信息，忽略无关内容。
|
||
2. 返回的内容尽量与文本一致。
|
||
3. 不需要包含重复项，每份材料只需列出一次。
|
||
4. 在提取“采购需求”部分的材料时，应明确对应的设备或货物名称，并通过冒号“:”连接，例如：
|
||
- "发射器：内部机构实用新型专利证书及外观专利证书"
|
||
- "发射器：外壳需有正规厂家世标认证"
|
||
示例输出，仅供格式参考：
|
||
{{
|
||
"证明材料":[
|
||
"具有独立承担民事责任的能力；",
|
||
"具有良好的商业信誉和健全的财务会计制度；",
|
||
"发射器：内部机构实用新型专利证书及外观专利证书",
|
||
"发射器：外壳需有正规厂家世标认证"
|
||
]
|
||
}}
|
||
"""
|
||
user_query=prompt_template.format(full_text=all_data_info)
|
||
# print(user_query)
|
||
model_res=qianwen_plus(user_query)
|
||
proof_materials = clean_json_string(model_res)
|
||
return proof_materials
|
||
|
||
def process_functions_in_parallel(tech_requirements_dict, busi_requirements_dict, zige_info, fuhe_info, zigefuhe_info,all_data_info):
|
||
# 准备输入参数
|
||
# 定义任务和对应参数
|
||
tasks = [
|
||
("tech_star_deviation", get_tech_star_deviation, (tech_requirements_dict,)),
|
||
("business_deviation_and_star", extract_business_deviation, (busi_requirements_dict,)),
|
||
("zigefuhe_deviation", extract_zige_deviation_table, (zige_info, fuhe_info, zigefuhe_info)),
|
||
("proof_materials", get_proof_materials, (all_data_info,))
|
||
]
|
||
|
||
results = {}
|
||
|
||
# 执行多线程任务
|
||
with concurrent.futures.ThreadPoolExecutor() as executor:
|
||
future_to_task = {executor.submit(func, *args): name for name, func, args in tasks}
|
||
time.sleep(0.5)
|
||
for future in concurrent.futures.as_completed(future_to_task):
|
||
name = future_to_task[future]
|
||
try:
|
||
result = future.result()
|
||
# 处理返回值（如果任务返回多个值，需要解包）
|
||
if name == "business_deviation_and_star":
|
||
results["business_deviation"], results["business_star_deviation"] = result
|
||
else:
|
||
results[name] = result
|
||
except Exception as e:
|
||
print(f"Task {name} failed with exception: {e}")
|
||
results[name] = None
|
||
|
||
# 返回结果
|
||
return (
|
||
results.get("tech_star_deviation"),
|
||
results.get("business_deviation"),
|
||
results.get("business_star_deviation"),
|
||
results.get("zigefuhe_deviation"),
|
||
results.get("proof_materials")
|
||
)
|
||
def get_tech_and_business_deviation(file_path,file_type,unique_id,output_folder,zb_type=2):
|
||
global logger
|
||
logger = get_global_logger(unique_id)
|
||
# 第一步：根据文件类型进行转换
|
||
if file_type == 1: # docx
|
||
docx_path=file_path
|
||
pdf_path = docx2pdf(file_path) # 将docx转换为pdf以供后续处理
|
||
elif file_type == 2: # pdf
|
||
# docx_path=pdf2docx(file_path)
|
||
docx_path=""
|
||
pdf_path = file_path
|
||
elif file_type == 3: # doc
|
||
docx_path=doc2docx(file_path)
|
||
pdf_path = docx2pdf(file_path)
|
||
else:
|
||
logger.error("不支持的文件类型！")
|
||
return None
|
||
# 第二步：根据zb_type确定选择项和类别，并截取PDF
|
||
if zb_type == 2:
|
||
selections = [1, 2, 3, 5]
|
||
category = 'goods' # 商品类
|
||
else:
|
||
selections = [1, 2, 3]
|
||
category = 'engineering' # 工程类
|
||
try:
|
||
files = truncate_pdf_multiple(pdf_path, output_folder, logger, category, selections)
|
||
except Exception as e:
|
||
logger.error(f"PDF截取过程中出错: {e}")
|
||
return None
|
||
|
||
# 根据zb_type分配路径
|
||
notice_path = files[0] if len(files) > 0 else ""
|
||
evaluation_path = files[1] if len(files) > 1 else ""
|
||
qualification_path = files[2] if len(files) > 2 else ""
|
||
procurement_path = files[3] if zb_type == 2 and len(files) > 3 else ""
|
||
invalid_path = files[-2]
|
||
|
||
# invalid_path=docx_path
|
||
invalid_path = docx_path if docx_path else invalid_path
|
||
if zb_type == 2 and not procurement_path:
|
||
procurement_path = invalid_path
|
||
if not evaluation_path:
|
||
evaluation_path=invalid_path
|
||
if not notice_path:
|
||
notice_path=invalid_path
|
||
if zb_type != 2:
|
||
try:
|
||
qualification_path = merge_pdfs(
|
||
[qualification_path, evaluation_path],
|
||
os.path.join(output_folder, "merged_qualification.pdf")
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"PDF合并过程中出错: {e}")
|
||
qualification_path = ""
|
||
tech_deviation={}
|
||
|
||
with concurrent.futures.ThreadPoolExecutor() as executor:
|
||
# 使用字典存储future对象
|
||
futures = {}
|
||
|
||
# 根据zb_type条件提交fetch_procurement_reqs任务
|
||
if zb_type == 2:
|
||
futures['procurement'] = executor.submit(fetch_procurement_reqs, procurement_path, invalid_path)
|
||
else:
|
||
# 当zb_type != 2时，设置默认空值
|
||
futures['procurement'] = concurrent.futures.Future()
|
||
futures['procurement'].set_result({}) # 设为默认空字典
|
||
|
||
# 提交combine_qualification_review任务
|
||
futures['review'] = executor.submit(combine_qualification_review, invalid_path, qualification_path, notice_path)
|
||
|
||
# 提交combine_evaluation_standards任务
|
||
futures['evaluation'] = executor.submit(combine_evaluation_standards, evaluation_path, invalid_path, zb_type)
|
||
|
||
# 获取并处理任务结果
|
||
try:
|
||
procurement_reqs = futures['procurement'].result()
|
||
except Exception as e:
|
||
logger.error(f'fetch_procurement_reqs 出现异常: {e}')
|
||
procurement_reqs = {} # 根据需要处理
|
||
|
||
try:
|
||
review_standards_res = futures['review'].result()
|
||
except Exception as e:
|
||
logger.error(f'combine_qualification_review 出现异常: {e}')
|
||
review_standards_res = {} # 根据需要处理
|
||
|
||
try:
|
||
evaluation_res = futures['evaluation'].result()
|
||
except Exception as e:
|
||
logger.error(f'combine_evaluation_standards 出现异常: {e}')
|
||
evaluation_res = {
|
||
"技术评分": {},
|
||
"商务评分": {}
|
||
}
|
||
evaluation_info=json.dumps(evaluation_res, ensure_ascii=False, indent=4)
|
||
technical_standards = {"技术评分": evaluation_res.get("技术评分", {})} #技术评议表
|
||
# commercial_standards = {"商务评分": evaluation_res.get("商务评分", {})} #商务评议表
|
||
tech_requirements = get_nested(procurement_reqs, ["采购需求"], {})
|
||
busi_requirements = {k: v for k, v in procurement_reqs.items() if k != "采购需求"}
|
||
busi_requirements_info=json.dumps(busi_requirements,ensure_ascii=False,indent=4)
|
||
if tech_requirements:
|
||
good_list = tech_requirements.pop('货物列表', []) # 如果 '货物列表' 不存在，返回 []
|
||
logger.info("Collected good_list from the processing function: %s", good_list)
|
||
tech_deviation = postprocess_technical_table(tech_requirements, good_list)
|
||
tech_deviation_info = json.dumps(tech_deviation, ensure_ascii=False, indent=4)
|
||
else:
|
||
tech_deviation_info=""
|
||
zige_info, fuhe_info, zigefuhe_info = prepare_for_zige_info(review_standards_res.get("资格审查", {}))
|
||
|
||
all_data_info = '\n'.join([zige_info, fuhe_info, zigefuhe_info, tech_deviation_info,busi_requirements_info, evaluation_info])
|
||
tech_star_deviation, business_deviation, business_star_deviation, zigefuhe_deviation, proof_materials= process_functions_in_parallel(
|
||
tech_requirements_dict=tech_requirements,
|
||
busi_requirements_dict=busi_requirements,
|
||
zige_info=zige_info,
|
||
fuhe_info=fuhe_info,
|
||
zigefuhe_info=zigefuhe_info,
|
||
all_data_info=all_data_info
|
||
)
|
||
return tech_deviation,tech_star_deviation,business_deviation,business_star_deviation,zigefuhe_deviation,proof_materials,technical_standards
|
||
|
||
|
||
prompt_template2_old = """以下文本是项目采购需求的商务要求部分。请从中提取以★、▲或其他特殊符号开头的要求项，它们一般是重要的商务要求，需要额外响应。返回结果应仅包含一个键名“商务要求带星”，其键值为字符串列表，每个字符串对应一个以★、▲或特殊符号开头的要求项。
|
||
**要求与指南**：
|
||
1. 每个以★、▲或特殊符号开头的要求项应作为单独的字符串。
|
||
2. 如果不存在符合条件的要求项，键值为空列表，即[]，无需额外返回其他说明性描述。
|
||
3. 特殊情况处理：
|
||
-对于类似'技术要求中带★条款项不满足的视为无效投标'这种描述带星★或带三角▲或带特殊符号的响应情况的，它本身没有描述具体的★、▲商务要求，因此不需要添加进字符串列表中；仅提取实际的带★、▲或其他特殊符号的商务要求项。
|
||
|
||
**禁止内容**：
|
||
确保所有输出内容均基于提供的实际文本内容，不使用任何预设的示例作为回答。
|
||
|
||
**示例输入如下**：
|
||
{{
|
||
"商务要求": [
|
||
"考虑设备兼容性、项目实施、交付及售后服务",
|
||
"★交货期（工期）：合同签订之日起 15个日历天内完成，并通过项目验收。",
|
||
"▲本项目报价须为固定总价，包含但不限于：采购、实施、调试、试运行、验收、运维等所有完成本项目相关的一切费用。"
|
||
]
|
||
}}
|
||
**对应的输出如下**：
|
||
{{
|
||
"商务要求带星": [
|
||
"★交货期（工期）：合同签订之日起 15个日历天内完成，并通过项目验收。",
|
||
"▲本项目报价须为固定总价，包含但不限于：采购、实施、调试、试运行、验收、运维等所有完成本项目相关的一切费用。"
|
||
]
|
||
}}
|
||
|
||
文本内容：{full_text}
|
||
"""
|
||
|
||
if __name__ == "__main__":
|
||
file_path=r"C:\Users\Administrator\Desktop\new招标文件\工程标\gcHBDL-2024-0017-001-招标文件.pdf"
|
||
file_type=2
|
||
output_folder = r"C:\Users\Administrator\Desktop\new招标文件\工程标\tmp"
|
||
tech_deviation,tech_star_deviation,business_deviation,business_star_deviation,zigefuhe_deviation,proof_materials=get_tech_and_business_deviation(file_path,file_type,"123",output_folder,1)
|
||
print("技术偏离表")
|
||
print(json.dumps(tech_deviation,ensure_ascii=False,indent=4))
|
||
print("技术带星")
|
||
print(json.dumps(tech_star_deviation,ensure_ascii=False,indent=4))
|
||
print("商务偏离表")
|
||
print(json.dumps(business_deviation, ensure_ascii=False, indent=4))
|
||
print("商务带星")
|
||
print(json.dumps(business_star_deviation, ensure_ascii=False, indent=4))
|
||
print("资格审查")
|
||
print(json.dumps(zigefuhe_deviation, ensure_ascii=False, indent=4))
|
||
print("证明材料")
|
||
print(json.dumps(proof_materials,ensure_ascii=False,indent=4))
|