utils/excel_utils.py

import pandas as pd
import io
from typing import List, Optional

def merge_excel_files(files: List, merge_type: str = "rows") -> pd.DataFrame:
    """合并多个Excel文件"""
    all_dfs = []
    for file in files:
        if file.name.endswith('.csv'):
            df = pd.read_csv(file)
        else:
            df = pd.read_excel(file)
        all_dfs.append(df)

    if merge_type == "rows":
        return pd.concat(all_dfs, ignore_index=True)
    else:
        return pd.concat(all_dfs, axis=1)

def remove_duplicates(df: pd.DataFrame, columns: Optional[List] = None) -> pd.DataFrame:
    """去除重复行"""
    return df.drop_duplicates(subset=columns, keep='first')

def clean_data(df: pd.DataFrame, remove_empty_rows: bool = True, remove_empty_cols: bool = True) -> pd.DataFrame:
    """清洗数据"""
    if remove_empty_rows:
        df = df.dropna(how='all')
    if remove_empty_cols:
        df = df.dropna(axis=1, how='all')
    return df.reset_index(drop=True)

def convert_format(df: pd.DataFrame, target_format: str) -> bytes:
    """转换数据格式"""
    buffer = io.BytesIO()
    if target_format == 'csv':
        df.to_csv(buffer, index=False, encoding='utf-8-sig')
    elif target_format == 'excel':
        df.to_excel(buffer, index=False, engine='openpyxl')
    elif target_format == 'json':
        df.to_json(buffer, orient='records', force_ascii=False, indent=2)
    return buffer.getvalue()

def create_pivot_table(df: pd.DataFrame, index_col: str, columns_col: str, values_col: str, aggfunc: str = 'sum') -> pd.DataFrame:
    """创建透视表"""
    return pd.pivot_table(df, index=index_col, columns=columns_col, values=values_col, aggfunc=aggfunc)

def vlookup_merge(left_df: pd.DataFrame, right_df: pd.DataFrame, key_col: str, value_col: str) -> pd.DataFrame:
    """类似VLOOKUP的合并"""
    right_subset = right_df[[key_col, value_col]]
    return left_df.merge(right_subset, on=key_col, how='left')