hello-agents/code/chapter9/codebase/data_processor.py

"""
数据处理模块
用于处理和转换数据
"""

import pandas as pd
from typing import List, Dict, Any


def process_data(data: List[Dict[str, Any]]) -> pd.DataFrame:
    """
    处理原始数据并返回DataFrame
    
    Args:
        data: 原始数据列表
        
    Returns:
        处理后的DataFrame
    """
    # TODO: 添加数据验证逻辑
    df = pd.DataFrame(data)
    df = clean_data(df)
    df = transform_data(df)
    return df


def clean_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    清理数据中的空值和异常值
    
    Args:
        df: 原始DataFrame
        
    Returns:
        清理后的DataFrame
    """
    # TODO: 实现更复杂的清理逻辑
    df = df.dropna()
    df = df.drop_duplicates()
    return df


def transform_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    转换数据格式
    
    Args:
        df: 输入DataFrame
        
    Returns:
        转换后的DataFrame
    """
    # TODO: 添加更多转换规则
    df['processed_date'] = pd.to_datetime(df['date'])
    return df


def aggregate_data(df: pd.DataFrame, group_by: List[str]) -> pd.DataFrame:
    """
    聚合数据
    
    Args:
        df: 输入DataFrame
        group_by: 分组字段列表
        
    Returns:
        聚合后的DataFrame
    """
    return df.groupby(group_by).agg({
        'value': ['sum', 'mean', 'count']
    })


def export_data(df: pd.DataFrame, output_path: str) -> None:
    """
    导出数据到文件
    
    Args:
        df: 要导出的DataFrame
        output_path: 输出文件路径
    """
    # TODO: 支持更多输出格式
    df.to_csv(output_path, index=False)
    print(f"Data exported to {output_path}")