Files
KG_generation/equipment_calculation/find_project_nodes.py
T
chentianrui 9609bb67b4 上传文件
2025-08-01 15:31:56 +08:00

149 lines
5.1 KiB
Python

import json
import os
from typing import Dict, List, Any, Tuple
def read_json_file(file_path: str) -> Dict:
"""读取JSON文件"""
try:
with open(file_path, "r", encoding="utf-8") as f:
data = json.load(f)
return data
except Exception as e:
print(f"读取JSON文件时出错: {e}")
return {}
def find_leaf_project_divisions(
node: Any, project_name: str = None, path: List = None
) -> List[Tuple[str, str, str, List[str]]]:
"""
递归查找最子集的项目划分节点
Args:
node: 当前节点
project_name: 当前工程名称
path: 当前路径
Returns:
List[Tuple[str, str, str, List[str]]]: 项目划分节点名称、调差类型、GUID和路径的列表
"""
results = []
if path is None:
path = []
# 如果节点是字典
if isinstance(node, dict):
# 判断是否是项目划分节点
is_project_division = node.get("type") == "项目划分"
has_children = "children" in node and isinstance(node["children"], list) and node["children"]
if is_project_division:
# 获取GUID
node_guid = node.get("GUID") or node.get("guid", "")
# 保存当前节点的名称到路径
current_path = path + [node.get("项目名称", "未命名")]
# 如果没有子节点或子节点不是项目划分节点,则认为是最子集的项目划分节点
if not has_children or not any(child.get("type") == "项目划分" for child in node["children"]):
results.append((node.get("项目名称", "未命名"), node_guid, current_path))
# 递归处理子节点
if has_children:
for child in node["children"]:
results.extend(find_leaf_project_divisions(child, project_name, current_path))
else:
# 遍历所有键值对
for key, value in node.items():
# 如果是工程名称
if isinstance(value, dict) and key != "children":
new_project_name = key
# 遍历调差类型
for adj_type, adj_value in value.items():
if isinstance(adj_value, list):
for item in adj_value:
results.extend(
find_leaf_project_divisions(item, new_project_name, path + [new_project_name])
)
# 处理其他可能的字典或列表
elif isinstance(value, (dict, list)):
results.extend(find_leaf_project_divisions(value, project_name, path))
# 如果节点是列表
elif isinstance(node, list):
for item in node:
results.extend(find_leaf_project_divisions(item, project_name, path))
return results
def find_all_project_divisions(json_file_path: str) -> List[Tuple[str, str, List[str]]]:
"""
查找JSON文件中所有最子集的项目划分节点和对应的调差类型
Args:
json_file_path: JSON文件路径
Returns:
List[Tuple[str, str, List[str]]]: 项目划分节点名称、调差类型和路径的列表
"""
# 读取JSON文件
data = read_json_file(json_file_path)
# 获取projectDivision
project_division = data.get("projectData", {}).get("projectDivision", {})
# 查找所有最子集的项目划分节点
results = find_leaf_project_divisions(project_division)
return results
def process_project_and_save(json_file_path: str, project_name: str, output_dir: str = "计算结果") -> str:
"""
处理指定的项目划分和调差类型,计算费用并保存结果
Args:
json_file_path: JSON文件路径
project_name: 项目划分名称
output_dir: 输出目录
Returns:
str: 输出文件路径
"""
# 确保输出目录存在
os.makedirs(output_dir, exist_ok=True)
# 设置输出文件名
output_file = os.path.join(output_dir, f"{project_name}.json")
# 返回输出文件路径
return output_file
def print_project_divisions(results: List[Tuple[str, str, List[str]]]):
"""打印项目划分节点和调差类型"""
print(f"找到 {len(results)} 个最子集项目划分节点:")
for i, (name, adj_type, path) in enumerate(results, 1):
path_str = " > ".join(path)
print(f"{i}. 项目划分: {name}")
print(f" 调差类型: {adj_type}")
print(f" 路径: {path_str}")
print()
def get_project_divisions_list(json_file_path: str) -> List[Tuple[str, str, str]]:
"""
获取JSON文件中所有项目划分节点和调差类型的列表,返回名称、调差类型和GUID
Args:
json_file_path: JSON文件路径
Returns:
List[Tuple[str, str, str]]: 项目划分节点名称、调差类型和GUID的列表
"""
results = find_all_project_divisions(json_file_path)
# 返回项目名称、调差类型和GUID,不包括路径
return [(name, guid) for name, guid, _ in results]