Python自动化实现对多个Excel工作簿中的工作表进行分类汇总

pip install pandas xlwings

import os
import pandas as pd
import xlwings as xw


def clean_to_number(series: pd.Series) -> pd.Series:
    """
    将可能带有货币符号、逗号、空格、文本前缀的金额列清洗为数值。
    例如：
    ￥12,345.67 -> 12345.67
    profit: 6543.21 -> 6543.21
    """
    series = series.astype(str).str.strip()
    series = series.str.replace(",", "", regex=False)
    series = series.str.replace(r"[¥￥$ ]", "", regex=True)
    series = series.str.replace(r"[^0-9\.\-]", "", regex=True)
    return pd.to_numeric(series, errors="coerce")


def summarize_one_sheet(
    df: pd.DataFrame,
    group_col: str = "销售区域",
    value_col: str = "销售利润"
) -> pd.DataFrame:
    """
    对单张工作表数据进行分类汇总。
    按 group_col 分组，对 value_col 求和。
    """
    if group_col not in df.columns:
        raise KeyError(f"缺少分组列：{group_col}")

    if value_col not in df.columns:
        raise KeyError(f"缺少汇总列：{value_col}")

    temp = df.copy()

    # 先清洗成数值，再汇总，避免字符串求和或排序错误
    temp[value_col] = clean_to_number(temp[value_col]).fillna(0)

    result = (
        temp.groupby(group_col, dropna=False)[value_col]
        .sum()
        .reset_index()
        .rename(columns={
            group_col: "销售区域",
            value_col: "销售利润汇总"
        })
        .sort_values("销售利润汇总", ascending=False)
    )

    return result


def batch_summary_workbooks(
    input_folder: str,
    output_folder: str,
    group_col: str = "销售区域",
    value_col: str = "销售利润",
    start_cell: str = "A1",
    write_cell: str = "J1"
) -> None:
    """
    批量处理多个工作簿中的所有工作表：
    1. 扫描 input_folder 下的 Excel 文件
    2. 遍历每个工作簿中的所有工作表
    3. 读取表格数据为 DataFrame
    4. 按指定字段分类汇总
    5. 将结果写回每张工作表的 write_cell 位置
    6. 保存到 output_folder
    """
    os.makedirs(output_folder, exist_ok=True)

    app = xw.App(visible=False, add_book=False)
    app.display_alerts = False
    app.screen_updating = False

    try:
        for file_name in os.listdir(input_folder):
            # 跳过 Excel 临时文件和非 xlsx 文件
            if file_name.startswith("~$"):
                continue

            if not file_name.lower().endswith(".xlsx"):
                continue

            input_path = os.path.join(input_folder, file_name)
            output_path = os.path.join(output_folder, file_name)

            print(f"\n[OPEN] 正在处理工作簿：{input_path}")

            wb = app.books.open(input_path)
            success_count = 0
            skip_count = 0

            try:
                for sht in wb.sheets:
                    try:
                        rng = sht.range(start_cell).expand("table")

                        if rng.value is None:
                            print(f"  [SKIP] {sht.name}：空表")
                            skip_count += 1
                            continue

                        df = rng.options(pd.DataFrame, header=1, index=False).value

                        if df is None or df.empty:
                            print(f"  [SKIP] {sht.name}：无有效数据")
                            skip_count += 1
                            continue

                        summary_df = summarize_one_sheet(
                            df,
                            group_col=group_col,
                            value_col=value_col
                        )

                        # 清理旧汇总区，避免上一次结果残留
                        sht.range(write_cell).resize(100, 3).clear_contents()

                        # 写回汇总结果，不写入 DataFrame 索引
                        sht.range(write_cell).options(index=False).value = summary_df
                        sht.autofit()

                        print(f"  [OK] {sht.name}：已汇总到 {write_cell}")
                        success_count += 1

                    except Exception as e:
                        print(f"  [SKIP] {sht.name}：{e}")
                        skip_count += 1

                wb.save(output_path)
                print(f"[DONE] 已保存：{output_path}，成功 {success_count} 张表，跳过 {skip_count} 张表")

            finally:
                wb.close()

    finally:
        app.quit()
        print("\n[ALL DONE] 所有工作簿处理完成")


if __name__ == "__main__":
    batch_summary_workbooks(
        input_folder=r"销售表",
        output_folder=r"输出结果",
        group_col="销售区域",
        value_col="销售利润",
        start_cell="A1",
        write_cell="J1"
    )

df.groupby("销售区域")["销售利润"].sum()