上传文件

equipment_calculation/calculation_strategy.py

@@ -0,0 +1,364 @@
+from abc import ABC, abstractmethod
+from typing import Dict, List, Any, Optional, Set, Tuple
+
+
+class CalculationStrategy(ABC):
+    """
+    计算策略接口
+
+    定义了计算过程中可以自定义的各种方法。
+    每种软件类型可以提供自己的实现，以修改计算过程中的任何规则。
+    """
+
+    @abstractmethod
+    def calculate_fee_base(
+        self,
+        fee_base: str,
+        cost_table: Dict[str, Any],
+        project_node: Dict[str, Any],
+        context: Any,
+        in_calculation: Optional[Set[str]] = None,
+    ) -> float:
+        """
+        计算取费基数
+
+        Args:
+            fee_base: 取费基数表达式
+            cost_table: 取费表
+            project_node: 项目节点
+            context: 计算上下文
+            in_calculation: 正在计算中的费用代码集合，用于检测循环依赖
+
+        Returns:
+            float: 计算结果
+        """
+        pass
+
+    @abstractmethod
+    def calculate_internal_fee(
+        self,
+        fee_code: str,
+        cost_table: Dict[str, Any],
+        project_node: Dict[str, Any],
+        context: Any,
+        in_calculation: Optional[Set[str]] = None,
+    ) -> float:
+        """
+        计算表内费用
+
+        Args:
+            fee_code: 费用代码
+            cost_table: 取费表
+            project_node: 项目节点
+            context: 计算上下文
+            in_calculation: 正在计算中的费用代码集合，用于检测循环依赖
+
+        Returns:
+            float: 计算结果
+        """
+        pass
+
+    @abstractmethod
+    def calculate_external_variable(self, var_name: str, context: Any) -> float:
+        """
+        计算表外变量
+
+        Args:
+            var_name: 变量名
+            context: 计算上下文
+
+        Returns:
+            float: 计算结果
+        """
+        pass
+
+    @abstractmethod
+    def find_fee_item_by_code(self, cost_table: Dict[str, Any], code: str) -> Optional[Dict[str, Any]]:
+        """
+        在取费表中查找指定代码的费用项
+
+        Args:
+            cost_table: 取费表
+            code: 费用代码
+
+        Returns:
+            Optional[Dict[str, Any]]: 找到的费用项，如果未找到则返回None
+        """
+        pass
+
+    @abstractmethod
+    def find_fee_base_nodes(self, node: Dict[str, Any], result: Optional[List] = None) -> List[Dict[str, Any]]:
+        """
+        递归查找取费表中包含"取费基数"的节点或有子节点的费用项
+
+        Args:
+            node: 取费表节点
+            result: 结果列表
+
+        Returns:
+            List[Dict[str, Any]]: 包含"取费基数"的节点或有子节点的费用项列表
+        """
+        pass
+
+    @abstractmethod
+    def calculate_all_fees(
+        self,
+        project_node: Dict[str, Any],
+        cost_table: Dict[str, Any],
+        json_file_path: Optional[str] = None,
+        engineering_type: Optional[str] = None,
+    ) -> Dict[str, float]:
+        """
+        计算所有费用
+
+        Args:
+            project_node: 项目节点
+            cost_table: 取费表
+            json_file_path: JSON文件路径
+            engineering_type: 工程类型
+
+        Returns:
+            Dict[str, float]: 计算结果，键为费用名称，值为计算结果
+        """
+        pass
+
+    @abstractmethod
+    def calculate_rcj_count(
+        self,
+        rcj_nodes: List[Tuple[Dict[str, Any], str]],
+        project_children: List[Dict[str, Any]],
+        json_file_path: Optional[str] = None,
+    ) -> List[Dict[str, Any]]:
+        """
+        计算人材机节点的数量，考虑父级消耗量
+
+        Args:
+            rcj_nodes: 人材机节点列表，每个元素是(节点, 父级ID)元组
+            project_children: 项目划分级别下的所有工程量节点
+            json_file_path: JSON文件路径，用于获取工程信息
+
+        Returns:
+            List[Dict[str, Any]]: 计算数量后的人材机节点列表
+        """
+        pass
+
+    @abstractmethod
+    def cat_rcj_count(self, rcj_nodes: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """
+        汇总人材机节点的数量
+
+        Args:
+            rcj_nodes: 人材机节点列表
+
+        Returns:
+            List[Dict[str, Any]]: 汇总后的人材机节点列表
+        """
+        pass
+
+    @abstractmethod
+    def calc_rcj_fee(self, rcj_nodes: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """
+        计算人材机节点的费用
+
+        Args:
+            rcj_nodes: 人材机节点列表
+
+        Returns:
+            List[Dict[str, Any]]: 计算费用后的人材机节点列表
+        """
+        pass
+
+    @abstractmethod
+    def format_rcj_output(self, rcj_nodes: List[Dict[str, Any]], node_type: str) -> List[Dict[str, Any]]:
+        """
+        格式化人材机节点输出
+
+        Args:
+            rcj_nodes: 人材机节点列表
+            node_type: 节点类型
+
+        Returns:
+            List[Dict[str, Any]]: 格式化后的人材机节点列表
+        """
+        pass
+
+    @abstractmethod
+    def calculate_rcj_fees(
+        self,
+        json_file_path: str,
+        project_name: str,
+        engineering_type: str,
+        project_guid: str = None,
+    ) -> Dict[str, Any]:
+        """
+        计算人材机合价
+
+        Args:
+            json_file_path: JSON文件路径
+            project_name: 项目名称
+            engineering_type: 工程类型
+            project_guid: 项目GUID，用于区分同名项目
+
+        Returns:
+            Dict[str, Any]: 计算结果
+        """
+        pass
+
+    @abstractmethod
+    def post_process_quantity_fees(self, output_file: str, project_name: str) -> None:
+        """
+        对工程量取费表进行后处理
+
+        Args:
+            output_file: 输出文件路径
+            project_name: 项目名称
+        """
+        pass
+
+    @abstractmethod
+    def preprocess_project_nodes(self, project_nodes: List[Dict[str, Any]]) -> None:
+        """
+        对工程量节点进行预处理
+
+        Args:
+            project_nodes: 工程量节点列表
+        """
+        pass
+
+
+class DefaultCalculationStrategy(CalculationStrategy):
+    """
+    默认计算策略实现
+
+    实现了所有计算方法的默认行为，与原有代码逻辑相同。
+    其他计算策略可以继承此类，只重写需要修改的方法。
+    """
+
+    def __init__(self):
+        # 导入必要的模块
+        from expressioncalculator import ExpressionCalculator
+        from bcl_utils import calculator
+        from item_acquisition import get_quantity_nodes, get_classified_resource_nodes
+
+        # 缓存已计算过的费用
+        self.calculated_fees = {}
+
+    def calculate_fee_base(
+        self,
+        fee_base: str,
+        cost_table: Dict[str, Any],
+        project_node: Dict[str, Any],
+        context: Any,
+        in_calculation: Optional[Set[str]] = None,
+    ) -> float:
+        """计算取费基数"""
+        from quantity_fee_calculator import calculate_fee_base as original_calculate_fee_base
+
+        # 传递自身作为计算策略
+        return original_calculate_fee_base(fee_base, cost_table, project_node, context, in_calculation, self)
+
+    def calculate_internal_fee(
+        self,
+        fee_code: str,
+        cost_table: Dict[str, Any],
+        project_node: Dict[str, Any],
+        context: Any,
+        in_calculation: Optional[Set[str]] = None,
+    ) -> float:
+        """计算表内费用"""
+        from quantity_fee_calculator import calculate_internal_fee as original_calculate_internal_fee
+
+        # 传递自身作为计算策略
+        return original_calculate_internal_fee(fee_code, cost_table, project_node, context, in_calculation, self)
+
+    def calculate_external_variable(self, var_name: str, context: Any) -> float:
+        """计算表外变量"""
+        # 直接使用 calculator.calculate，而不是调用 quantity_fee_calculator.py 中的函数
+        from bcl_utils import calculator
+
+        result = calculator.calculate(var_name, context)
+        return float(result) if result is not None else 0.0
+
+    def find_fee_item_by_code(self, cost_table: Dict[str, Any], code: str) -> Optional[Dict[str, Any]]:
+        """在取费表中查找指定代码的费用项"""
+        from quantity_fee_calculator import find_fee_item_by_code as original_find_fee_item_by_code
+
+        return original_find_fee_item_by_code(cost_table, code)
+
+    def find_fee_base_nodes(self, node: Dict[str, Any], result: Optional[List] = None) -> List[Dict[str, Any]]:
+        """递归查找取费表中包含"取费基数"的节点或有子节点的费用项"""
+        from quantity_fee_calculator import find_fee_base_nodes as original_find_fee_base_nodes
+
+        return original_find_fee_base_nodes(node, result)
+
+    def calculate_all_fees(
+        self,
+        project_node: Dict[str, Any],
+        cost_table: Dict[str, Any],
+        json_file_path: Optional[str] = None,
+        engineering_type: Optional[str] = None,
+    ) -> Dict[str, float]:
+        """计算所有费用"""
+        from quantity_fee_calculator import calculate_all_fees as original_calculate_all_fees
+
+        # 传递自身作为计算策略
+        return original_calculate_all_fees(project_node, cost_table, json_file_path, engineering_type, self)
+
+    def calculate_rcj_count(
+        self,
+        rcj_nodes: List[Tuple[Dict[str, Any], str]],
+        project_children: List[Dict[str, Any]],
+        json_file_path: Optional[str] = None,
+    ) -> List[Dict[str, Any]]:
+        """计算人材机节点的数量"""
+        from resource_fee_calculator import calc_rcj_count as original_calc_rcj_count
+
+        return original_calc_rcj_count(rcj_nodes, project_children, json_file_path)
+
+    def cat_rcj_count(self, rcj_nodes: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """汇总人材机节点的数量"""
+        from resource_fee_calculator import cat_rcj_count as original_cat_rcj_count
+
+        return original_cat_rcj_count(rcj_nodes)
+
+    def calc_rcj_fee(self, rcj_nodes: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """计算人材机节点的费用"""
+        from resource_fee_calculator import calc_rcj_fee as original_calc_rcj_fee
+
+        return original_calc_rcj_fee(rcj_nodes)
+
+    def format_rcj_output(self, rcj_nodes: List[Dict[str, Any]], node_type: str) -> List[Dict[str, Any]]:
+        """格式化人材机节点输出"""
+        from resource_fee_calculator import format_rcj_output as original_format_rcj_output
+
+        return original_format_rcj_output(rcj_nodes, node_type)
+
+    # 修复 calculate_rcj_fees 未定义的错误
+    def calculate_rcj_fees(self, json_file_path, project_name, project_guid=None):
+        """
+        计算项目划分节点下所有人材机节点的合价
+
+        Args:
+            json_file_path: JSON文件路径
+            project_name: 项目名称
+            project_guid: 项目GUID，用于区分同名项目
+
+        Returns:
+            Dict[str, Any]: 计算结果
+        """
+        # 导入原始函数
+        from resource_fee_calculator import calculate_rcj_fees as original_calculate_rcj_fees
+
+        # 调用原始函数
+        return original_calculate_rcj_fees(json_file_path, project_name, project_guid)
+
+    def post_process_quantity_fees(self, output_file: str, project_name: str) -> None:
+        """对工程量取费表进行后处理"""
+        # 默认实现不做任何操作
+        pass
+
+    def preprocess_project_nodes(self, project_nodes: List[Dict[str, Any]]) -> None:
+        """
+        默认实现不做任何预处理
+        """
+        pass