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}")