chentianrui
332 lines
13 KiB
Python
332 lines
13 KiB
Python
import os
|
||
import json
|
||
import logging
|
||
from memory_profiler import profile
|
||
from typing import Dict, List, Any, Optional, Tuple
|
||
from equipment_calculation.software_types import (
|
||
get_software_type,
|
||
SoftwareCategory,
|
||
EngineeringType,
|
||
ALL_SOFTWARE_TYPES,
|
||
)
|
||
from equipment_calculation.software_calculators import get_calculator
|
||
|
||
|
||
# 软件类别名称映射字典，将各种变体映射到标准类别
|
||
CATEGORY_MAPPING = {
|
||
# 主网及其变体
|
||
"主网": "主网",
|
||
"主网工程": "主网",
|
||
"主网项目": "主网",
|
||
# 配网及其变体
|
||
"配网": "配网",
|
||
"配网造价": "配网",
|
||
"配网清单": "配网",
|
||
# 技改及其变体
|
||
"技改": "技改",
|
||
"技改工程": "技改",
|
||
"技改项目": "技改",
|
||
"技改造价": "技改",
|
||
"技改清单": "技改",
|
||
}
|
||
|
||
|
||
# 项目类型名称映射字典，将各种变体映射到标准类型（预算/清单）
|
||
PROJECT_TYPE_MAPPING = {
|
||
"概预算工程": "预算",
|
||
"初步设计概算": "预算",
|
||
"可行性研究投资估算": "预算",
|
||
"施工图预算": "预算",
|
||
"配网定额计价": "预算",
|
||
"招标控制价": "清单",
|
||
"投标报价": "清单",
|
||
"招投标工程": "清单",
|
||
"配网清单招投标计价": "清单",
|
||
}
|
||
|
||
|
||
####应该是加在json后没释放
|
||
def parse_json_content(json_file_path: str) -> Tuple[Optional[str], Optional[str], Optional[str]]:
|
||
"""
|
||
从JSON文件内容中解析：
|
||
- 软件类别: 来自 basicData["软件类别"]（若无则尝试 basicData["软件名称"] 作为兜底）
|
||
- 项目类型: 来自 basicData["项目类型"]（期望为 "预算" 或 "清单"）
|
||
- 工程类型: 来自 projectData.projectInfo["工程类型"]
|
||
|
||
:param json_file_path: JSON文件路径
|
||
:return: (category, project_type, engineering_type) 元组，解析失败对应位置返回 None
|
||
"""
|
||
try:
|
||
with open(json_file_path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
|
||
# 提取 basicData
|
||
basic_data = data.get("basicData", {}) if isinstance(data, dict) else {}
|
||
# 软件类别（优先 软件类别，其次 软件名称）
|
||
category = basic_data.get("软件类别") or basic_data.get("软件名称")
|
||
engineering_type = basic_data.get("项目类型") or basic_data.get("工程类型") or basic_data.get("工程类别")
|
||
# 规范化项目类型为 预算/清单
|
||
if engineering_type:
|
||
mapped_pt = PROJECT_TYPE_MAPPING.get(engineering_type)
|
||
if mapped_pt:
|
||
engineering_type = mapped_pt
|
||
else:
|
||
print(f"警告: basicData.项目类型 '{engineering_type}' 不是有效值，将使用默认值 '清单'")
|
||
engineering_type = "清单"
|
||
|
||
# 规范化软件类别
|
||
if category:
|
||
if category in CATEGORY_MAPPING:
|
||
category = CATEGORY_MAPPING[category]
|
||
else:
|
||
print(f"警告: basicData中的软件名称/名称 '{category}' 不是有效值，将使用默认值 '主网'")
|
||
category = "主网"
|
||
|
||
# 提取工程类型：projectData.projectInfo.工程类型
|
||
project_info = data.get("projectData", {}).get("projectInfo", {}) if isinstance(data, dict) else {}
|
||
project_type = project_info.get("工程类型")
|
||
|
||
# 打印解析结果（便于调试）
|
||
print(f"解析完成: 软件类别={category}, 项目类型={engineering_type}, 工程类型={project_type}")
|
||
|
||
return category, engineering_type, project_type
|
||
|
||
except Exception as e:
|
||
print(f"解析JSON文件内容时出错: {str(e)}")
|
||
return None, None, None
|
||
|
||
|
||
def process_json_file(
|
||
json_file_path: str,
|
||
output_dir: str,
|
||
calculate_type: str,
|
||
project_name: str = None,
|
||
bcl_dir_path: str | None = None,
|
||
) -> bool:
|
||
"""
|
||
处理单个JSON文件
|
||
|
||
Args:
|
||
json_file_path: JSON文件路径
|
||
output_dir: 输出目录
|
||
calculate_type: 计算类型（all: 全部, quantity: 工程量取费, resource: 人材机合价）
|
||
project_name: 项目名称，如果不指定则处理所有项目
|
||
|
||
Returns:
|
||
bool: 处理是否成功
|
||
"""
|
||
try:
|
||
# 从JSON文件内容中解析软件类别、项目类型(预算/清单) 和 工程类型(如 变电/线路/配网)
|
||
# parse_json_content 返回顺序为: (category, project_type, engineering_type)
|
||
category, project_type, engineering_type = parse_json_content(json_file_path)
|
||
|
||
# 如果解析失败，使用默认值
|
||
if category is None:
|
||
category = "主网"
|
||
print(f"无法从文件中解析软件类别，使用默认值: {category}")
|
||
|
||
# 项目类型(预算/清单) 兜底
|
||
if project_type is None:
|
||
project_type = "预算"
|
||
print(f"无法从文件中解析项目类型(预算/清单)，使用默认值: {project_type}")
|
||
|
||
# 可选：记录工程类型
|
||
if engineering_type:
|
||
print(f"工程类型: {engineering_type}")
|
||
|
||
# 获取软件类型
|
||
# 这里的第二个参数应为项目类型(预算/清单)
|
||
software_type = get_software_type(category, project_type)
|
||
print(f"使用软件类型: {software_type.name}")
|
||
|
||
# 获取计算器
|
||
calculator = get_calculator(software_type)
|
||
# 若提供了 bcl 目录，优先使用新接口
|
||
if hasattr(calculator, "set_bcl_dir") and bcl_dir_path:
|
||
calculator.set_bcl_dir(bcl_dir_path)
|
||
if not calculator:
|
||
print(f"错误: 未找到软件类型 {software_type.name} 的计算器")
|
||
return False
|
||
|
||
# 获取文件名（不含扩展名），并替换空格和特殊字符
|
||
base_filename = os.path.basename(json_file_path).replace(".json", "")
|
||
# 创建安全的目录名（替换空格和特殊字符）
|
||
safe_dirname = base_filename.replace(" ", "_").replace("\\", "_").replace("/", "_")
|
||
|
||
# 设置自定义输出目录
|
||
custom_output_dir = os.path.join(output_dir, safe_dirname)
|
||
|
||
# 确保输出目录存在
|
||
try:
|
||
os.makedirs(custom_output_dir, exist_ok=True)
|
||
print(f"创建输出目录: {custom_output_dir}")
|
||
except Exception as e:
|
||
print(f"创建输出目录失败: {str(e)}")
|
||
# 尝试使用更安全的目录名
|
||
safe_dirname = f"project_{hash(base_filename) % 10000}"
|
||
custom_output_dir = os.path.join(output_dir, safe_dirname)
|
||
os.makedirs(custom_output_dir, exist_ok=True)
|
||
print(f"使用备用输出目录: {custom_output_dir}")
|
||
|
||
# 检查计算器是否有set_output_dir方法，如果有则直接设置输出目录
|
||
if hasattr(calculator, "set_output_dir"):
|
||
calculator.set_output_dir(custom_output_dir)
|
||
print(f"已设置计算器输出目录: {custom_output_dir}")
|
||
|
||
# 根据计算类型执行计算
|
||
if calculate_type in ["all", "quantity"]:
|
||
print("开始计算工程量取费表...")
|
||
calculator.calculate_quantity_fee_tables(
|
||
json_file_path=json_file_path,
|
||
project_name=project_name,
|
||
# 这里的 project_type 传递给计算器用于BCL筛选，应为 工程类型，例如"变电/线路/配网"
|
||
project_type=engineering_type,
|
||
)
|
||
|
||
if calculate_type in ["all", "resource"]:
|
||
print("开始计算人材机合价...")
|
||
calculator.calculate_resource_fee_tables(json_file_path=json_file_path, project_name=project_name)
|
||
else:
|
||
# 创建一个新的计算器，复制原始计算器的属性
|
||
class CustomOutputCalculator:
|
||
def __init__(self, original_calculator, custom_dir):
|
||
self.original_calculator = original_calculator
|
||
self.custom_dir = custom_dir
|
||
|
||
def get_output_dir(self):
|
||
return self.custom_dir
|
||
|
||
def __getattr__(self, name):
|
||
return getattr(self.original_calculator, name)
|
||
|
||
# 创建自定义输出目录的计算器
|
||
custom_calculator = CustomOutputCalculator(calculator, custom_output_dir)
|
||
# 自定义计算器同样继承 bcl_dir 设置
|
||
if hasattr(custom_calculator, "set_bcl_dir") and bcl_dir_path:
|
||
custom_calculator.set_bcl_dir(bcl_dir_path)
|
||
|
||
# 根据计算类型执行计算
|
||
if calculate_type in ["all", "quantity"]:
|
||
print("开始计算工程量取费表...")
|
||
custom_calculator.calculate_quantity_fee_tables(
|
||
json_file_path=json_file_path,
|
||
project_name=project_name,
|
||
# 这里的 project_type 传递给计算器用于BCL筛选，应为 工程类型，例如"变电/线路/配网"
|
||
project_type=engineering_type,
|
||
)
|
||
|
||
if calculate_type in ["all", "resource"]:
|
||
print("开始计算人材机合价...")
|
||
custom_calculator.calculate_resource_fee_tables(
|
||
json_file_path=json_file_path, project_name=project_name
|
||
)
|
||
|
||
print(f"文件 {json_file_path} 处理完成")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"处理文件 {json_file_path} 时出错: {str(e)}")
|
||
import traceback
|
||
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
def process_directory(
|
||
input_dir: str, output_dir: str, calculate_type: str = "quantity", bcl_dir_path: str | None = None
|
||
) -> None:
|
||
"""
|
||
批量处理目录中的JSON文件
|
||
|
||
Args:
|
||
input_dir: 输入目录路径
|
||
output_dir: 输出目录路径
|
||
calculate_type: 计算类型（all: 全部, quantity: 工程量取费, resource: 人材机合价）
|
||
"""
|
||
# 确保输出目录存在
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
|
||
# 获取目录中的所有JSON文件
|
||
json_files = [f for f in os.listdir(input_dir) if f.lower().endswith(".json")]
|
||
|
||
if not json_files:
|
||
print(f"警告: 在目录 {input_dir} 中未找到JSON文件")
|
||
return
|
||
|
||
print(f"找到 {len(json_files)} 个JSON文件，开始处理...")
|
||
|
||
# 处理每个JSON文件
|
||
success_count = 0
|
||
for i, json_file in enumerate(json_files, 1):
|
||
json_file_path = os.path.join(input_dir, json_file)
|
||
print(f"处理文件 {i}/{len(json_files)}: {json_file}")
|
||
|
||
if process_json_file(json_file_path, output_dir, calculate_type, bcl_dir_path=bcl_dir_path):
|
||
success_count += 1
|
||
|
||
print(f"处理完成，成功: {success_count}/{len(json_files)}")
|
||
|
||
|
||
def bcl_calculate(
|
||
input_dir: str, output_dir: str, calculate_type: str = "quantity", bcl_dir_path: str | None = None
|
||
) -> None:
|
||
"""
|
||
主函数，处理指定目录中的所有JSON文件
|
||
|
||
Args:
|
||
input_dir: 输入目录路径
|
||
output_dir: 输出目录路径
|
||
calculate_type: 计算类型（all: 全部, quantity: 工程量取费, resource: 人材机合价）
|
||
"""
|
||
# 将日志写入本次工程的 bclresults 目录
|
||
_configure_bcl_logging(output_dir)
|
||
|
||
print(f"开始处理目录: {input_dir}")
|
||
print(f"输出目录: {output_dir}")
|
||
print(f"计算类型: {calculate_type}")
|
||
|
||
process_directory(input_dir, output_dir, calculate_type, bcl_dir_path=bcl_dir_path)
|
||
|
||
print("所有文件处理完成")
|
||
|
||
|
||
def _configure_bcl_logging(log_dir: str):
|
||
"""将 bcl 计算日志写入指定目录下的 bcl_calculator.log。
|
||
|
||
- 清理先前的 FileHandler，避免多次叠加/重复输出。
|
||
- 保留/添加一个 StreamHandler 以在控制台输出。
|
||
"""
|
||
try:
|
||
os.makedirs(log_dir, exist_ok=True)
|
||
log_path = os.path.join(log_dir, "bcl_calculator.log")
|
||
|
||
logger = logging.getLogger()
|
||
logger.setLevel(logging.INFO)
|
||
|
||
# 移除已有的 FileHandler，避免写到旧位置或重复写
|
||
for h in list(logger.handlers):
|
||
if isinstance(h, logging.FileHandler):
|
||
logger.removeHandler(h)
|
||
try:
|
||
h.close()
|
||
except Exception:
|
||
pass
|
||
|
||
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
|
||
|
||
file_handler = logging.FileHandler(log_path, mode="w", encoding="utf-8")
|
||
file_handler.setLevel(logging.INFO)
|
||
file_handler.setFormatter(formatter)
|
||
logger.addHandler(file_handler)
|
||
|
||
# 确保有一个控制台输出
|
||
if not any(isinstance(h, logging.StreamHandler) for h in logger.handlers):
|
||
stream_handler = logging.StreamHandler()
|
||
stream_handler.setLevel(logging.INFO)
|
||
stream_handler.setFormatter(formatter)
|
||
logger.addHandler(stream_handler)
|
||
|
||
print(f"日志输出重定向到: {log_path}")
|
||
except Exception as e:
|
||
print(f"配置日志输出失败，将继续使用默认日志配置: {e}")
|