编写自动化办公Python3脚本：Excel批量处理实例

在日常日常运营与财务工作中，跨部门的数据汇总往往需要面对几十甚至上百个结构相近却细微有别的 Excel 工作簿。手动逐一打开、复制粘贴不仅耗时费力，而且极易引入遗漏或格式错乱。本文通过一个典型的业务销售报表批量合并场景，讲解如何编写一个健壮、高容错的 Python 3 自动化脚本，完成从目录扫描、表头校验、数据清洗到带样式透视导出的完整过程。

一、任务目标与环境准备

本实例设定的具体业务场景是：公司每月初会从全国 12 个分部收到当月的销售流水表格（存放于 ./data/monthly_reports/ 文件夹中）。这些表格的文件名类似 北京分部_10月.xlsx、华东中心_10月.xlsx。由于提报人员习惯差异，部分表格可能存在多余的空行、首尾带空格的文本型数字，甚至在编辑时产生了隐藏的临时锁文件（以 ~$ 开头）。

我们的目标是编写一段 Python 3 脚本，一次性实现以下目标：

• 安全遍历文件：仅读取有效的 .xlsx 工作簿，主动识别并跳过所有 Office 临时锁文件。
• 字段映射与规范化：解决个别表格中“订单ID”与“订单编号”、“销售金额”与“实收金额”列名轻微不一致的问题。
• 元数据增强：在合并后的总表中新增一列“来源分部”，自动从文件名称中提取。
• 输出双 Sheet 成果表：Sheet 1 存储清洗后的全量明细合并库；Sheet 2 输出按分部汇总的经营透视表，并为表头和单元格应用专业的表格样式。

处理 Excel 数据推荐使用 pandas（负责高速数据对齐、清洗与统计）与 openpyxl（负责底层的 Excel 读取与单元格样式渲染）。在终端或命令行中执行如下安装命令：

pip install pandas openpyxl

二、批量处理的生命周期设计

批量处理并不是简单地将所有表格用循环加在一起。在工程实践中，必须将流程拆解为清晰的四个操作环节：扫描发现、结构提取、数据加工与成果持久化。

每个环节承担明确的职责：

1. 路径扫描过滤：利用标准库 pathlib.Path 检索目标文件夹，确保跨平台（Windows / macOS / Linux）路径兼容，并严防系统临时文件干扰。
2. 结构校验提取：单表读取时预先验证必要的核心字段。如果某张表格缺少关键的“销售金额”列，则记录警报并跳过，避免中断整体任务。
3. 清洗对齐汇总：使用 pd.concat 进行纵向拼接，统一处理缺失值与数据格式，并计算分组透视指标。
4. 格式化落盘：使用 pd.ExcelWriter 生成包含多个 Sheet 的工作簿，并通过 openpyxl.styles 自动设定列宽和高对比度表头。

三、核心管道与数据流转架构

不同分部的报表可能存在脏数据，例如业务员在输入金额时误输入了货币符号 ￥ 或逗号分隔符 1,250.00，导致整列被误判为字符串对象（object）。数据转换管道必须具备容错清洗能力。

数据流转的关键环节包括：

• 别名归一化（Alias Mapping）：通过 Python 字典定义别名映射规则。例如 {"订单ID": "订单编号", "实收金额": "销售金额"}，在数据读入后立刻调用 rename(columns=...)，保证进入合并池的字段完全一致。
• 类型强制脱水（Coercion）：将金额字段中的多余字符剔除后使用 pd.to_numeric(..., errors='coerce') 转换，并将无法解析的脏数据填充为 0 或合理均值。
• 透视聚合（Pivot Analysis）：基于清洗后的统一明细，以“所属分部”为聚合轴，生成 订单总数、销售总额 与 客单均价 的经营报表。

四、模块化脚本实现与函数职责

为了让脚本便于维护和扩展，我们将功能解耦为四个函数，并在主执行块中组织串联。

1. 目录扫描与文件过滤：scan_input_files

使用标准库中的 pathlib.Path 能够轻松处理绝对路径与相对路径。注意利用正则表达式或简单前缀过滤掉 Office 隐藏锁文件。

from pathlib import Path
from typing import List

def scan_input_files(data_dir: str) -> List[Path]:
    """
    扫描目标目录下所有有效的 .xlsx 文件，排除 ~$ 开头的临时锁文件。
    """
    folder = Path(data_dir)
    if not folder.exists() or not folder.is_dir():
        raise FileNotFoundError(f"指定的目录不存在: {folder.resolve()}")

    valid_files = [
        f for f in folder.glob("*.xlsx")
        if not f.name.startswith("~$")
    ]
    
    if not valid_files:
        print(f"[提示] 在 {data_dir} 中未找到符合条件的 .xlsx 文件。")
    else:
        print(f"[扫描完成] 找到 {len(valid_files)} 个待处理的 Excel 文件。")

    return valid_files

2. 单表读取、字段对齐与清洗：clean_single_file

该函数封装了单表的容错逻辑：读取表格、重命名同义列、验证必要字段、注入文件名元数据，并返回处理完毕的 DataFrame。如果文件格式损毁或关键列缺失，函数将打印警告并安全返回 None。

import pandas as pd
from typing import Optional, Dict

COLUMN_ALIASES: Dict[str, str] = {
    "订单ID": "订单编号",
    "销售流水号": "订单编号",
    "实收金额": "销售金额",
    "总额": "销售金额",
    "成交时间": "销售日期",
    "日期": "销售日期",
    "经手人": "销售员"
}

REQUIRED_COLUMNS = ["订单编号", "销售金额", "销售日期", "销售员"]

def clean_single_file(file_path: Path) -> Optional[pd.DataFrame]:
    """
    读取并清洗单个 Excel 文件，返回标准化的 DataFrame。
    """
    try:
        # 读取第一张工作表，将所有列名转为字符串去除空白
        df = pd.read_excel(file_path, sheet_name=0)
        df.columns = [str(col).strip() for col in df.columns]

        # 映射同义列名
        df.rename(columns=COLUMN_ALIASES, inplace=True)

        # 检查必要字段是否存在
        missing = [col for col in REQUIRED_COLUMNS if col not in df.columns]
        if missing:
            print(f"[跳过] 文件 {file_path.name} 缺少必要列: {missing}")
            return None

        # 只保留目标字段并复制
        df = df[REQUIRED_COLUMNS].copy()

        # 剔除订单编号为空的废行
        df.dropna(subset=["订单编号"], inplace=True)

        # 清洗销售金额：去除货币符号和逗号，强转为浮点型
        if df["销售金额"].dtype == object:
            df["销售金额"] = df["销售金额"].astype(str).str.replace(r"[¥,$, ]", "", regex=True)
        df["销售金额"] = pd.to_numeric(df["销售金额"], errors="coerce").fillna(0.0)

        # 统一日期格式为 YYYY-MM-DD
        df["销售日期"] = pd.to_datetime(df["销售日期"], errors="coerce").dt.strftime("%Y-%m-%d")

        # 注入来源元数据（例如从 '北京分部_10月.xlsx' 提取 '北京分部'）
        dept_name = file_path.stem.split("_")[0]
        df.insert(0, "所属分部", dept_name)

        return df

    except Exception as e:
        print(f"[错误] 解析文件 {file_path.name} 发生异常: {e}")
        return None

3. 纵向汇总与经营透视：aggregate_data

将各个分部的 DataFrame 清单使用 pd.concat 拼接在一起，重置全局行索引，并利用 groupby 生成汇总透视表。

from typing import Tuple

def aggregate_data(df_list: List[pd.DataFrame]) -> Tuple[pd.DataFrame, pd.DataFrame]:
    """
    汇总所有子表，返回 (全量明细表, 部门透视统计表)。
    """
    if not df_list:
        raise ValueError("没有可供汇总的数据帧。")

    # 1. 纵向合并所有分部数据
    master_df = pd.concat(df_list, ignore_index=True)

    # 2. 生成部门透视分析表
    pivot_df = master_df.groupby("所属分部").agg(
        订单数量=("订单编号", "count"),
        销售总额=("销售金额", "sum"),
        平均客单价=("销售金额", "mean")
    ).reset_index()

    # 格式化客单价保留 2 位小数
    pivot_df["平均客单价"] = pivot_df["平均客单价"].round(2)
    pivot_df["销售总额"] = pivot_df["销售总额"].round(2)

    return master_df, pivot_df

4. 格式化输出与样式美化：export_styled_report

为了使导出的 Excel 具备直接汇报交付的专业质感，我们利用 openpyxl 的样式模块，给表头添加钢灰与稳重蓝色底色，统一居中对齐、设置细网格边框，并自动根据单元格字符长度调节列宽。

from openpyxl.styles import PatternFill, Font, Alignment, Border, Side
from openpyxl.utils import get_column_letter

def export_styled_report(master_df: pd.DataFrame, pivot_df: pd.DataFrame, output_file: str):
    """
    将明细与汇总数据导出至同一个 Excel 文件的两个 Sheet 中，并渲染样式。
    """
    out_path = Path(output_file)
    out_path.parent.mkdir(parents=True, exist_ok=True)

    # 样式定义
    font_header = Font(name="微软雅黑", size=11, bold=True, color="FFFFFF")
    font_body = Font(name="微软雅黑", size=10, color="1F2937")
    fill_header_gray = PatternFill(start_color="4F6272", end_color="4F6272", fill_type="solid")
    fill_header_blue = PatternFill(start_color="477DAD", end_color="477DAD", fill_type="solid")
    
    thin_side = Side(border_style="thin", color="D8DEE4")
    border_all = Border(left=thin_side, right=thin_side, top=thin_side, bottom=thin_side)
    
    align_center = Alignment(horizontal="center", vertical="center")
    align_right = Alignment(horizontal="right", vertical="center")

    with pd.ExcelWriter(out_path, engine="openpyxl") as writer:
        master_df.to_excel(writer, sheet_name="全国明细汇总", index=False)
        pivot_df.to_excel(writer, sheet_name="部门经营分析", index=False)

        workbook = writer.book

        # 美化 Sheet 1 (全国明细汇总)
        ws_detail = workbook["全国明细汇总"]
        _format_sheet(ws_detail, fill_header_gray, font_header, font_body, border_all, align_center, align_right)

        # 美化 Sheet 2 (部门经营分析)
        ws_pivot = workbook["部门经营分析"]
        _format_sheet(ws_pivot, fill_header_blue, font_header, font_body, border_all, align_center, align_right)

    print(f"[导出成功] 报表已保存至: {out_path.resolve()}")


def _format_sheet(ws, header_fill, font_header, font_body, border, align_center, align_right):
    """
    辅助函数：统一设置单个工作表的行高、列宽、表头底色与单元格边框。
    """
    ws.row_dimensions[1].height = 28

    # 格式化表头行
    for col_idx in range(1, ws.max_column + 1):
        cell = ws.cell(row=1, column=col_idx)
        cell.fill = header_fill
        cell.font = font_header
        cell.alignment = align_center
        cell.border = border

    # 格式化数据行并计算自适应列宽
    for col in ws.columns:
        max_len = 0
        col_letter = get_column_letter(col[0].column)
        
        for cell in col:
            cell.border = border
            if cell.row > 1:
                cell.font = font_body
                # 数值列右对齐，其他居中
                if isinstance(cell.value, (int, float)):
                    cell.alignment = align_right
                else:
                    cell.alignment = align_center

            val_str = str(cell.value or "")
            # 处理中文字符宽度（中文算 2 字符）
            str_len = sum(2 if ord(c) > 127 else 1 for c in val_str)
            if str_len > max_len:
                max_len = str_len

        # 适当增加边距
        ws.column_dimensions[col_letter].width = max(max_len + 4, 12)

五、端到端完整运行示例

下面将上述四个函数组装为可以直接执行的完整自动化脚本。你可以直接将本段代码存为 batch_excel_processor.py。

# -*- coding: utf-8 -*-
"""
batch_excel_processor.py
自动化办公 Python3 脚本：Excel 批量处理实例
"""

import sys
from pathlib import Path
import pandas as pd
from openpyxl.styles import PatternFill, Font, Alignment, Border, Side
from openpyxl.utils import get_column_letter

# 1. 字段同义别名表
COLUMN_ALIASES = {
    "订单ID": "订单编号",
    "销售流水号": "订单编号",
    "实收金额": "销售金额",
    "总额": "销售金额",
    "成交时间": "销售日期",
    "日期": "销售日期",
    "经手人": "销售员"
}

REQUIRED_COLUMNS = ["订单编号", "销售金额", "销售日期", "销售员"]

def process_batch_excel(source_dir: str, output_filepath: str):
    """
    主批处理入口函数
    """
    input_path = Path(source_dir)
    if not input_path.exists():
        print(f"[错误] 目录不存在: {source_dir}")
        return

    # 步骤一：安全扫描文件
    excel_files = [
        f for f in input_path.glob("*.xlsx")
        if not f.name.startswith("~$")
    ]
    if not excel_files:
        print("[提示] 未发现待处理的 Excel 文件。")
        return

    print(f"开始批量处理 {len(excel_files)} 个文件...")

    # 步骤二：逐表读取与清洗
    cleaned_frames = []
    for file in excel_files:
        try:
            df = pd.read_excel(file, sheet_name=0)
            df.columns = [str(c).strip() for c in df.columns]
            df.rename(columns=COLUMN_ALIASES, inplace=True)

            if not all(col in df.columns for col in REQUIRED_COLUMNS):
                print(f"[警告] 文件 {file.name} 缺少关键列，已略过。")
                continue

            df = df[REQUIRED_COLUMNS].copy()
            df.dropna(subset=["订单编号"], inplace=True)

            # 金额清洗
            if df["销售金额"].dtype == object:
                df["销售金额"] = df["销售金额"].astype(str).str.replace(r"[¥,$, ]", "", regex=True)
            df["销售金额"] = pd.to_numeric(df["销售金额"], errors="coerce").fillna(0.0)
            df["销售日期"] = pd.to_datetime(df["销售日期"], errors="coerce").dt.strftime("%Y-%m-%d")

            # 标记分部
            dept_name = file.stem.split("_")[0]
            df.insert(0, "所属分部", dept_name)
            cleaned_frames.append(df)
            print(f"  + 成功解析: {file.name} ({len(df)} 行)")

        except Exception as e:
            print(f"[错误] 读取 {file.name} 失败: {e}")

    if not cleaned_frames:
        print("[错误] 没有有效的数据被提取，退出处理。")
        return

    # 步骤三：合并与透视分析
    master_df = pd.concat(cleaned_frames, ignore_index=True)
    pivot_df = master_df.groupby("所属分部").agg(
        订单总量=("订单编号", "count"),
        销售总额=("销售金额", "sum"),
        客单均价=("销售金额", "mean")
    ).reset_index()

    pivot_df["销售总额"] = pivot_df["销售总额"].round(2)
    pivot_df["客单均价"] = pivot_df["客单均价"].round(2)

    # 步骤四：格式化保存
    out_file = Path(output_filepath)
    out_file.parent.mkdir(parents=True, exist_ok=True)

    with pd.ExcelWriter(out_file, engine="openpyxl") as writer:
        master_df.to_excel(writer, sheet_name="全国明细汇总", index=False)
        pivot_df.to_excel(writer, sheet_name="分部经营透视", index=False)

        wb = writer.book
        _style_sheet(wb["全国明细汇总"], header_color="4F6272")
        _style_sheet(wb["分部经营透视"], header_color="477DAD")

    print(f"\n[任务完成] 汇总报表已成功生成：{out_file.resolve()}")
    print(f"总计合并记录: {len(master_df)} 行 | 涉及分部: {len(pivot_df)} 个")


def _style_sheet(ws, header_color: str):
    header_fill = PatternFill(start_color=header_color, end_color=header_color, fill_type="solid")
    font_h = Font(name="微软雅黑", size=11, bold=True, color="FFFFFF")
    font_b = Font(name="微软雅黑", size=10, color="1F2937")
    thin = Side(border_style="thin", color="D8DEE4")
    border = Border(left=thin, right=thin, top=thin, bottom=thin)

    ws.row_dimensions[1].height = 26

    for cell in ws[1]:
        cell.fill = header_fill
        cell.font = font_h
        cell.alignment = Alignment(horizontal="center", vertical="center")
        cell.border = border

    for col in ws.columns:
        max_l = 0
        letter = get_column_letter(col[0].column)
        for cell in col:
            cell.border = border
            if cell.row > 1:
                cell.font = font_b
                cell.alignment = Alignment(
                    horizontal="right" if isinstance(cell.value, (int, float)) else "center",
                    vertical="center"
                )
            length = sum(2 if ord(c) > 127 else 1 for c in str(cell.value or ""))
            max_l = max(max_l, length)
        ws.column_dimensions[letter].width = max(max_l + 4, 12)

if __name__ == "__main__":
    # 示例调用
    SRC_DIR = "./data/monthly_reports"
    DST_FILE = "./output/2026年10月_全国销售汇总分析.xlsx"
    process_batch_excel(SRC_DIR, DST_FILE)

六、总结与生产环境防御清单

编写批量处理脚本时，业务逻辑往往只占代码量的一半，另一半则是围绕各种异常边界所构建的防御机制。

在将脚本部署到自动化定时任务（如 crontab 或 Windows 任务计划程序）之前，请对照以下检查项：

通过上述结构化设计，Python 3 脚本不仅能够稳定处理成百上千张 Excel 报表，还能够形成标准化的流水线，为后续引入定时调度、数据库写入以及邮件自动推送奠定坚实的基础。