"""
数据计算处理器

提供各种数据计算功能，如数量计算、价格计算、汇总统计等
"""

import pandas as pd
import numpy as np
from typing import Dict, Any, Optional, List, Union
from app.core.utils.log_utils import get_logger

logger = get_logger(__name__)


class DataCalculator:
    """数据计算处理器
    
    提供标准化的数据计算功能，支持各种业务计算规则
    """
    
    def __init__(self, config: Optional[Dict[str, Any]] = None):
        """初始化数据计算器
        
        Args:
            config: 计算配置
        """
        self.config = config or {}
        self.calculation_rules = []
    
    def add_rule(self, rule_type: str, **kwargs):
        """添加计算规则
        
        Args:
            rule_type: 规则类型
            **kwargs: 规则参数
        """
        rule = {'type': rule_type, **kwargs}
        self.calculation_rules.append(rule)
        logger.debug(f"添加计算规则: {rule_type}")
    
    def calculate(self, df: pd.DataFrame) -> pd.DataFrame:
        """执行数据计算
        
        Args:
            df: 输入数据
            
        Returns:
            计算后的数据
        """
        logger.info(f"开始数据计算，原始数据形状: {df.shape}")
        
        result_df = df.copy()
        
        for i, rule in enumerate(self.calculation_rules):
            try:
                logger.debug(f"执行计算规则 {i+1}/{len(self.calculation_rules)}: {rule['type']}")
                result_df = self._apply_rule(result_df, rule)
                logger.debug(f"规则执行完成，数据形状: {result_df.shape}")
            except Exception as e:
                logger.error(f"计算规则执行失败: {rule}, 错误: {e}")
                # 继续执行下一个规则，而不是中断整个流程
                continue
        
        logger.info(f"数据计算完成，最终数据形状: {result_df.shape}")
        return result_df
    
    def _apply_rule(self, df: pd.DataFrame, rule: Dict[str, Any]) -> pd.DataFrame:
        """应用单个计算规则
        
        Args:
            df: 数据
            rule: 规则配置
            
        Returns:
            处理后的数据
        """
        rule_type = rule.get('type')
        
        if rule_type == 'multiply':
            return self._multiply(df, rule)
        elif rule_type == 'divide':
            return self._divide(df, rule)
        elif rule_type == 'add':
            return self._add(df, rule)
        elif rule_type == 'subtract':
            return self._subtract(df, rule)
        elif rule_type == 'formula':
            return self._formula(df, rule)
        elif rule_type == 'round':
            return self._round(df, rule)
        elif rule_type == 'sum':
            return self._sum(df, rule)
        elif rule_type == 'aggregate':
            return self._aggregate(df, rule)
        else:
            logger.warning(f"未知的计算规则类型: {rule_type}")
            return df
    
    def _multiply(self, df: pd.DataFrame, rule: Dict[str, Any]) -> pd.DataFrame:
        """乘法计算
        
        Args:
            df: 数据
            rule: 规则配置
            
        Returns:
            处理后的数据
        """
        source_column = rule.get('source_column')
        target_column = rule.get('target_column')
        factor = rule.get('factor', 1)
        
        if source_column and target_column:
            if source_column in df.columns:
                df[target_column] = df[source_column] * factor
                logger.debug(f"乘法计算: {source_column} * {factor} -> {target_column}")
            else:
                logger.warning(f"源列不存在: {source_column}")
        
        return df
    
    def _divide(self, df: pd.DataFrame, rule: Dict[str, Any]) -> pd.DataFrame:
        """除法计算
        
        Args:
            df: 数据
            rule: 规则配置
            
        Returns:
            处理后的数据
        """
        source_column = rule.get('source_column')
        target_column = rule.get('target_column')
        divisor = rule.get('divisor', 1)
        
        if source_column and target_column and divisor != 0:
            if source_column in df.columns:
                df[target_column] = df[source_column] / divisor
                logger.debug(f"除法计算: {source_column} / {divisor} -> {target_column}")
            else:
                logger.warning(f"源列不存在: {source_column}")
        elif divisor == 0:
            logger.error("除数不能为0")
        
        return df
    
    def _add(self, df: pd.DataFrame, rule: Dict[str, Any]) -> pd.DataFrame:
        """加法计算
        
        Args:
            df: 数据
            rule: 规则配置
            
        Returns:
            处理后的数据
        """
        columns = rule.get('columns', [])
        target_column = rule.get('target_column')
        constant = rule.get('constant', 0)
        
        if target_column:
            if isinstance(columns, str):
                columns = [columns]
            
            if columns:
                # 列相加
                valid_columns = [col for col in columns if col in df.columns]
                if valid_columns:
                    df[target_column] = df[valid_columns].sum(axis=1) + constant
                    logger.debug(f"加法计算: {valid_columns} + {constant} -> {target_column}")
                else:
                    logger.warning(f"没有有效的列用于加法计算: {columns}")
            else:
                # 只加常数
                if target_column in df.columns:
                    df[target_column] = df[target_column] + constant
                    logger.debug(f"加法计算: {target_column} + {constant}")
                else:
                    logger.warning(f"目标列不存在: {target_column}")
        
        return df
    
    def _subtract(self, df: pd.DataFrame, rule: Dict[str, Any]) -> pd.DataFrame:
        """减法计算
        
        Args:
            df: 数据
            rule: 规则配置
            
        Returns:
            处理后的数据
        """
        minuend = rule.get('minuend')  # 被减数列
        subtrahend = rule.get('subtrahend')  # 减数列
        target_column = rule.get('target_column')
        constant = rule.get('constant', 0)
        
        if target_column and minuend and minuend in df.columns:
            if subtrahend and subtrahend in df.columns:
                df[target_column] = df[minuend] - df[subtrahend] - constant
                logger.debug(f"减法计算: {minuend} - {subtrahend} - {constant} -> {target_column}")
            else:
                df[target_column] = df[minuend] - constant
                logger.debug(f"减法计算: {minuend} - {constant} -> {target_column}")
        else:
            logger.warning(f"减法计算参数不完整或列不存在")
        
        return df
    
    def _formula(self, df: pd.DataFrame, rule: Dict[str, Any]) -> pd.DataFrame:
        """公式计算
        
        Args:
            df: 数据
            rule: 规则配置
            
        Returns:
            处理后的数据
        """
        formula = rule.get('formula')
        target_column = rule.get('target_column')
        
        if formula and target_column:
            try:
                df[target_column] = df.eval(formula)
                logger.debug(f"公式计算: {formula} -> {target_column}")
            except Exception as e:
                logger.error(f"公式计算失败: {formula}, 错误: {e}")
        else:
            logger.warning("公式计算缺少公式或目标列")
        
        return df
    
    def _round(self, df: pd.DataFrame, rule: Dict[str, Any]) -> pd.DataFrame:
        """四舍五入
        
        Args:
            df: 数据
            rule: 规则配置
            
        Returns:
            处理后的数据
        """
        columns = rule.get('columns', [])
        decimals = rule.get('decimals', 0)
        
        if isinstance(columns, str):
            columns = [columns]
        
        target_columns = columns or df.select_dtypes(include=[np.number]).columns
        
        for col in target_columns:
            if col in df.columns and pd.api.types.is_numeric_dtype(df[col]):
                df[col] = df[col].round(decimals)
                logger.debug(f"四舍五入: {col} 保留 {decimals} 位小数")
        
        return df
    
    def _sum(self, df: pd.DataFrame, rule: Dict[str, Any]) -> pd.DataFrame:
        """求和计算
        
        Args:
            df: 数据
            rule: 规则配置
            
        Returns:
            处理后的数据
        """
        columns = rule.get('columns', [])
        target_column = rule.get('target_column')
        group_by = rule.get('group_by')
        
        if isinstance(columns, str):
            columns = [columns]
        
        if group_by and group_by in df.columns:
            # 分组求和
            if columns:
                for col in columns:
                    if col in df.columns:
                        sum_result = df.groupby(group_by)[col].sum()
                        logger.debug(f"分组求和: {col} 按 {group_by} 分组")
            else:
                # 所有数值列分组求和
                numeric_columns = df.select_dtypes(include=[np.number]).columns
                sum_result = df.groupby(group_by)[numeric_columns].sum()
                logger.debug(f"分组求和: 所有数值列 按 {group_by} 分组")
        else:
            # 总体求和
            if columns:
                valid_columns = [col for col in columns if col in df.columns]
                if valid_columns and target_column:
                    df[target_column] = df[valid_columns].sum(axis=1)
                    logger.debug(f"求和计算: {valid_columns} -> {target_column}")
            else:
                # 所有数值列求和
                numeric_columns = df.select_dtypes(include=[np.number]).columns
                if target_column and len(numeric_columns) > 0:
                    df[target_column] = df[numeric_columns].sum(axis=1)
                    logger.debug(f"求和计算: {list(numeric_columns)} -> {target_column}")
        
        return df
    
    def _aggregate(self, df: pd.DataFrame, rule: Dict[str, Any]) -> pd.DataFrame:
        """聚合计算
        
        Args:
            df: 数据
            rule: 规则配置
            
        Returns:
            处理后的数据
        """
        group_by = rule.get('group_by')
        aggregations = rule.get('aggregations', {})
        
        if group_by and group_by in df.columns:
            # 构建聚合函数字典
            agg_dict = {}
            for column, func in aggregations.items():
                if column in df.columns:
                    if isinstance(func, str):
                        agg_dict[column] = func
                    elif isinstance(func, list):
                        agg_dict[column] = func
            
            if agg_dict:
                result = df.groupby(group_by).agg(agg_dict)
                logger.debug(f"聚合计算: 按 {group_by} 分组, 聚合: {agg_dict}")
                return result.reset_index()
        
        return df
    
    # 便捷方法
    def multiply(self, source_column: str, target_column: str, factor: float):
        """乘法计算"""
        self.add_rule('multiply', source_column=source_column, 
                     target_column=target_column, factor=factor)
        return self
    
    def divide(self, source_column: str, target_column: str, divisor: float):
        """除法计算"""
        self.add_rule('divide', source_column=source_column, 
                     target_column=target_column, divisor=divisor)
        return self
    
    def add(self, columns: Union[str, List[str]], target_column: str, constant: float = 0):
        """加法计算"""
        self.add_rule('add', columns=columns, target_column=target_column, constant=constant)
        return self
    
    def subtract(self, minuend: str, target_column: str, 
                  subtrahend: Optional[str] = None, constant: float = 0):
        """减法计算"""
        self.add_rule('subtract', minuend=minuend, target_column=target_column,
                     subtrahend=subtrahend, constant=constant)
        return self
    
    def formula(self, formula: str, target_column: str):
        """公式计算"""
        self.add_rule('formula', formula=formula, target_column=target_column)
        return self
    
    def round_columns(self, columns: Optional[Union[str, List[str]]] = None, decimals: int = 0):
        """四舍五入"""
        self.add_rule('round', columns=columns, decimals=decimals)
        return self
    
    def sum_columns(self, columns: Optional[Union[str, List[str]]] = None, 
                   target_column: Optional[str] = None, group_by: Optional[str] = None):
        """求和计算"""
        self.add_rule('sum', columns=columns, target_column=target_column, group_by=group_by)
        return self
    
    def aggregate(self, group_by: str, aggregations: Dict[str, Union[str, List[str]]]):
        """聚合计算"""
        self.add_rule('aggregate', group_by=group_by, aggregations=aggregations)
        return self