288 lines
12 KiB
Python
288 lines
12 KiB
Python
import json
|
||
import re
|
||
|
||
|
||
# 对于每个target_value元素，如果有完美匹配json_data中的键，那就加入这个完美匹配的键名，否则，把全部模糊匹配到的键名都加入
|
||
def find_keys_by_value(target_value, json_data):
|
||
matched_keys = [k for k, v in json_data.items() if v == target_value] # 首先检查 JSON 中的每个键值对，如果值完全等于目标值，则将这些键收集起来。
|
||
if not matched_keys:
|
||
matched_keys = [k for k, v in json_data.items() if
|
||
isinstance(v, str) and v.startswith(target_value)] # 如果没有找到完全匹配的键，它会检查字符串类型的值是否以目标值开头，并收集这些键。
|
||
return matched_keys # eg:[3.1,3.1.1,3.1.2,3.2...]
|
||
|
||
|
||
# 定义查找以特定前缀开始的键的函数，eg:若match_keys中有3.1，那么以3.1为前缀的键都会被找出来，如3.1.1 3.1.2...
|
||
def find_keys_with_prefix(key_prefix, json_data):
|
||
subheadings = [k for k in json_data if k.startswith(key_prefix)]
|
||
return subheadings
|
||
|
||
|
||
# 从完整的json文件中读取所需数据，eg:投标、评标
|
||
# def extract_json(data, target_values):
|
||
# results = {}
|
||
#
|
||
# # 遍历所有目标值
|
||
# for target_value in target_values:
|
||
# # 找到所有与目标值匹配的键
|
||
# matched_keys = find_keys_by_value(target_value, data)
|
||
#
|
||
# for key in matched_keys:
|
||
# # 查找所有以该键为前缀的子键，限制只提取直接子项
|
||
# key_and_subheadings = find_keys_with_prefix(key, data)
|
||
#
|
||
# for subkey in key_and_subheadings:
|
||
# # 如果子键有多级结构（比如 '7.2.1'），并且是直接子项
|
||
# if "." in subkey:
|
||
# parent_key = subkey.rsplit('.', 1)[0]
|
||
# top_level_key = parent_key.split('.')[0] + '.'
|
||
#
|
||
# # 确保顶级键不会重复添加
|
||
# if top_level_key not in results:
|
||
# results[top_level_key] = data[top_level_key]
|
||
#
|
||
# # 添加或更新父级键
|
||
# if parent_key not in results:
|
||
# if parent_key in data:
|
||
# results[parent_key] = data[parent_key]
|
||
#
|
||
# # 添加当前子键和它的值
|
||
# if subkey in data:
|
||
# results[subkey] = data[subkey]
|
||
#
|
||
# return results
|
||
def extract_json(data, target_values):
|
||
results = {}
|
||
for target_value in target_values:
|
||
matched_keys = find_keys_by_value(target_value, data)
|
||
for key in matched_keys:
|
||
key_and_subheadings = find_keys_with_prefix(key, data)
|
||
for subkey in key_and_subheadings:
|
||
if "." in subkey:
|
||
parent_key = subkey.rsplit('.', 1)[0]
|
||
top_level_key = parent_key.split('.')[0] + '.'
|
||
# 特别处理定标相关的顶级键，确保不会重复添加其他键
|
||
if top_level_key not in results:
|
||
results[top_level_key] = target_value
|
||
# 添加或更新父级键
|
||
if parent_key not in results:
|
||
if parent_key in data:
|
||
results[parent_key] = data[parent_key]
|
||
# 添加当前键
|
||
results[subkey] = data[subkey]
|
||
return results
|
||
|
||
def sort_clean_data_keys(data):
|
||
# 预处理：删除键名中的空格
|
||
def preprocess_key(key):
|
||
return re.sub(r'\s+', '', key)
|
||
|
||
# 将键转换成由整数构成的元组，作为排序依据
|
||
def key_func(key):
|
||
return tuple(int(part) for part in re.split(r'\D+', key) if part)
|
||
|
||
# 创建一个新的字典，键名经过预处理
|
||
preprocessed_data = {preprocess_key(key): value for key, value in data.items()}
|
||
|
||
# 对预处理后的字典键进行排序
|
||
sorted_keys = sorted(preprocessed_data.keys(), key=key_func)
|
||
|
||
# 创建一个新的字典，按照排序后的键添加键值对
|
||
sorted_data = {key: preprocessed_data[key] for key in sorted_keys}
|
||
|
||
return sorted_data
|
||
|
||
|
||
# 转换结构化的JSON数据
|
||
def transform_json(data):
|
||
result = {}
|
||
temp = {0: result} # 初始化根字典
|
||
|
||
# 首先,创建一个临时字典用于检查是否存在三级标题
|
||
has_subkey = {}
|
||
for key in data.keys():
|
||
parts = key.split('.')
|
||
if len(parts) > 2 and parts[1]:
|
||
parent_key = parts[0] + '.' + parts[1]
|
||
has_subkey[parent_key] = True
|
||
|
||
for key, value in data.items():
|
||
match = re.match(r'(\d+)(?:\.(\d+))?(?:\.(\d+))?', key)
|
||
if match:
|
||
levels = [int(l) for l in match.groups() if l is not None]
|
||
if (len(levels) - 1) in temp:
|
||
parent = temp[len(levels) - 1]
|
||
else:
|
||
print(f"No parent found at level {len(levels) - 1} for key '{key}'. Check the data structure.")
|
||
continue
|
||
|
||
if len(levels) == 1: # 一级标题
|
||
# 优先按 '\n' 拆分
|
||
if '\n' in value:
|
||
new_key, *new_value = value.split('\n', 1)
|
||
new_key = new_key.strip()
|
||
new_value = new_value[0].strip() if new_value else ""
|
||
# 如果没有 '\n'，再检查 ':' 或 '：'，并进行拆分
|
||
elif ':' in value or '：' in value:
|
||
delimiter = ':' if ':' in value else '：'
|
||
new_key, new_value = value.split(delimiter, 1)
|
||
new_key = new_key.strip()
|
||
new_value = new_value.strip()
|
||
else:
|
||
new_key = value.strip()
|
||
new_value = ""
|
||
|
||
parent[new_key] = {}
|
||
if new_value:
|
||
parent[new_key][new_key] = new_value # 使用 new_key 作为键名，而不是固定的 "content"
|
||
temp[len(levels)] = parent[new_key]
|
||
elif len(levels) == 2: # 二级标题
|
||
new_key, *new_value = value.split('\n', 1)
|
||
new_key = new_key.strip()
|
||
new_value = new_value[0].strip() if new_value else ""
|
||
|
||
if f"{levels[0]}.{levels[1]}" in has_subkey:
|
||
parent[new_key] = [new_value] if new_value else []
|
||
else:
|
||
parent[new_key] = new_value
|
||
|
||
temp[len(levels)] = parent[new_key]
|
||
else: # 三级标题
|
||
if isinstance(parent, dict):
|
||
parent_key = list(parent.keys())[-1]
|
||
if isinstance(parent[parent_key], list):
|
||
parent[parent_key].append(value)
|
||
elif parent[parent_key]:
|
||
parent[parent_key] = [parent[parent_key], value]
|
||
else:
|
||
parent[parent_key] = [value]
|
||
elif isinstance(parent, list):
|
||
parent.append(value)
|
||
|
||
def remove_single_item_lists(node):
|
||
if isinstance(node, dict):
|
||
for key in list(node.keys()):
|
||
node[key] = remove_single_item_lists(node[key])
|
||
if isinstance(node[key], list) and len(node[key]) == 1:
|
||
node[key] = node[key][0]
|
||
return node
|
||
|
||
return remove_single_item_lists(result)
|
||
|
||
|
||
# 主要是处理键值中若存在若干序号且每个序号块的内容>=50字符的时候，用列表表示。
|
||
def post_process(value):
|
||
# 如果传入的是非字符串值，直接返回原值
|
||
if not isinstance(value, str):
|
||
return value
|
||
|
||
# 定义可能的分割模式及其正则表达式
|
||
patterns = [
|
||
(r'\d+、', r'(?=\d+、)'), # 匹配 '1、'
|
||
(r'[（(]\d+[）)]', r'(?=[（(]\d+[）)])'), # 匹配 '(1)' 或 '（1）'
|
||
(r'\d+\.', r'(?=\d+\.)'), # 匹配 '1.'
|
||
(r'[一二三四五六七八九十]、', r'(?=[一二三四五六七八九十]、)'), # 匹配 '一、'、'二、' 等
|
||
(r'[一二三四五六七八九十]\.', r'(?=[一二三四五六七八九十]\.)') # 匹配 '一.'、'二.' 等
|
||
]
|
||
|
||
# 初始化用于保存最早匹配到的模式及其位置
|
||
first_match = None
|
||
first_match_position = len(value) # 初始值设为文本长度，确保任何匹配都会更新它
|
||
|
||
# 遍历所有模式，找到第一个出现的位置
|
||
for search_pattern, split_pattern_candidate in patterns:
|
||
match = re.search(search_pattern, value)
|
||
if match:
|
||
# 如果这个匹配的位置比当前记录的更靠前，更新匹配信息
|
||
if match.start() < first_match_position:
|
||
first_match = split_pattern_candidate
|
||
first_match_position = match.start()
|
||
|
||
# 如果找到了最早出现的匹配模式，使用它来分割文本
|
||
if first_match:
|
||
blocks = re.split(first_match, value)
|
||
else:
|
||
# 如果没有匹配的模式，保留原文本
|
||
blocks = [value]
|
||
|
||
processed_blocks = []
|
||
for block in blocks:
|
||
if not block:
|
||
continue
|
||
# 计算中英文字符总数，如果大于50，则加入列表
|
||
if block and len(re.findall(r'[\u4e00-\u9fff\w]', block)) >= 50:
|
||
processed_blocks.append(block.strip())
|
||
else:
|
||
# 如果发现有块长度小于50，返回原数据
|
||
return value
|
||
|
||
# 如果所有的块都符合条件，返回分割后的列表
|
||
return processed_blocks
|
||
|
||
|
||
"""
|
||
递归处理嵌套的数据结构（字典和列表）。
|
||
对最内层的字符串值应用 post_process 函数。
|
||
post_process 函数尝试将长字符串按特定模式分割成块，每块至少包含50个中英文字符。
|
||
如果字典中所有值都是 ""、"/" 或空列表，则返回'键'的列表。
|
||
"""
|
||
def process_nested_data(data):
|
||
# 先检查是否所有值都是 ""、"/" 或空列表
|
||
if isinstance(data, dict) and all(v == "" or v == "/" or (isinstance(v, list) and not v) for v in data.values()):
|
||
return list(data.keys())
|
||
# 递归遍历字典，处理最内层的字符串
|
||
if isinstance(data, dict):
|
||
# 如果当前项是字典，继续递归遍历其键值对
|
||
result = {}
|
||
for key, value in data.items():
|
||
processed_value = process_nested_data(value)
|
||
# 如果处理后的值是只有一个元素的列表，就直接使用该元素
|
||
if isinstance(processed_value, list) and len(processed_value) == 1:
|
||
result[key] = processed_value[0]
|
||
else:
|
||
result[key] = processed_value
|
||
return result
|
||
elif isinstance(data, list):
|
||
# 如果是列表，直接返回列表，保持原样
|
||
return data
|
||
else:
|
||
# 到达最内层，处理非字典和非列表的元素（字符串）
|
||
return post_process(data)
|
||
|
||
|
||
|
||
|
||
# 读取JSON数据，提取内容，转换结构，并打印结果
|
||
def extract_from_notice(clause_path, type):
|
||
if type == 1:
|
||
target_values = ["投标","投标文件","响应文件"]
|
||
elif type == 2:
|
||
target_values = ["开标", "评标", "定标","磋商程序","中标"]
|
||
elif type == 3:
|
||
target_values = ["重新招标、不再招标和终止招标", "重新招标", "不再招标", "终止招标"]
|
||
elif type == 4:
|
||
target_values = ["评标"] # 测试
|
||
else:
|
||
raise ValueError(
|
||
"Invalid type specified. Use 1 for '投标文件, 投标' or 2 for '开标, 评标, 定标'or 3 for '重新招标'")
|
||
with open(clause_path, 'r', encoding='utf-8') as file:
|
||
data = json.load(file)
|
||
extracted_data = extract_json(data, target_values) # 读取json
|
||
# print(json.dumps(extracted_data,ensure_ascii=False,indent=4))
|
||
sorted_data = sort_clean_data_keys(extracted_data) # 对输入的字典 data 的键进行预处理和排序
|
||
transformed_data = transform_json(sorted_data)
|
||
# print(json.dumps(transformed_data,ensure_ascii=False,indent=4))
|
||
final_result = process_nested_data(transformed_data)
|
||
return final_result
|
||
|
||
|
||
# TODO: extract_json新版本仍有问题，未知。
|
||
if __name__ == "__main__":
|
||
# file_path = 'C:\\Users\\Administrator\\Desktop\\fsdownload\\3bffaa84-2434-4bd0-a8ee-5c234ccd7fa0\\clause1.json'
|
||
file_path="C:\\Users\\Administrator\\Desktop\\招标文件\\special_output\\clause1.json"
|
||
try:
|
||
res = extract_from_notice(file_path, 1) # 可以改变此处的 type 参数测试不同的场景
|
||
res2 = json.dumps(res, ensure_ascii=False, indent=4)
|
||
print(res2)
|
||
except ValueError as e:
|
||
print(e)
|