
每个月最后两天,工位上都会堆满打印出来的Excel表格。
上个月28号,财务总监在群里发了条消息:“各部门请在30号之前提交本月费用明细表,格式按模板来。”然后群里开始陆续回复“收到”。
到了30号下午,作为运营部的数据对接人,收到了12个部门发来的Excel文件。每个文件的列名都不太一样,有的写“费用金额”,有的写“金额(元)”,还有写“总额”的。有的部门加了合计行,有的没加。有个部门甚至把月份写在了第一行当标题,数据从第三行才开始。
要把这12份表合成一份汇总表,还要跟上个月的数据做对比,标出差异超过10%的项。
手动搞?上次搞完眼睛花了,还漏了行政部的两张表,被点名批评。
那天晚上加班到九点,对着第12份Excel的“金额(元)”列发了会儿呆,决定不能再这么搞了。这个场景是不是很熟悉?
pandas读Excel本来就快,12个文件几秒钟读完。列名不统一的问题做个映射表就行。对比上月数据直接用DataFrame的merge。openpyxl用来输出带格式的汇总Excel,比pandas自带的to_excel好看很多。
三个库各管一段,读取、处理、输出都齐了。
定义列名映射表,把各部门五花八门的列名统一到标准名
遍历文件夹读取所有Excel,跳过表头行不一致的问题
合并为一张大表,按部门加标签
跟上月数据做对比,算变化百分比
输出一份汇总Excel,差异大的标红
先装依赖:
pip install pandas openpyxl然后是主脚本:
import pandas as pdimport osimport globfrom datetime import datetimefrom openpyxl import Workbookfrom openpyxl.styles import Font, PatternFill, Alignment, Border, Sidefrom openpyxl.utils.dataframe import dataframe_to_rows# ========== 配置区 ==========# 文件所在目录INPUT_DIR = "./reports/2024_08"# 当月各部门报表LAST_MONTH_FILE = "./reports/summary_2024_07.xlsx"# 上月汇总OUTPUT_FILE = f"./reports/summary_{datetime.now().strftime('%Y_%m')}.xlsx"# 列名映射 —— 把各种奇葩列名统一成标准名COLUMN_MAP = {# 部门"部门名称": "部门","所属部门": "部门","dept": "部门",# 费用类别"费用类别": "类别","费用类型": "类别","类别名称": "类别","category": "类别",# 金额"费用金额": "金额","金额(元)": "金额","总额": "金额","总金额": "金额","amount": "金额",# 备注"备注说明": "备注","remarks": "备注","说明": "备注",}# 需要保留的标准列STANDARD_COLUMNS = ["部门", "类别", "金额", "备注"]# 差异阈值(超过这个比例就标红)DIFF_THRESHOLD = 0.10# ========== 读取与清洗 ==========def read_single_excel(filepath):""" 读取单个Excel文件,处理表头不一致的问题。 自动跳过空行和非数据的行。 """filename = os.path.basename(filepath)# 先读前几行看看结构df_preview = pd.read_excel(filepath, nrows=5, header=None)# 找到真正的表头行(第一行包含"部门"或"类别"或"金额"相关文字的行)header_row = 0for i in range(min(5, len(df_preview))):row_text = " ".join(str(x) for x in df_preview.iloc[i].values)if any(kw in row_text for kw in ["部门", "费用", "金额", "类别", "dept", "amount"]):header_row = ibreak# 用找到的表头行读取数据df = pd.read_excel(filepath, header=header_row)# 统一列名df.rename(columns=COLUMN_MAP, inplace=True)# 删除不在标准列中的多余列(备注可选)keep_cols = [c for c in STANDARD_COLUMNS if c in df.columns]df = df[keep_cols]# 补充缺失的列for col in STANDARD_COLUMNS:if col not in df.columns:df[col] = ""# 如果没有"部门"列,用文件名当部门名if"部门" not in keep_cols or df["部门"].isna().all():dept_name = os.path.splitext(filename)[0]df["部门"] = dept_name# 清洗金额列:转成数字,无法转换的设为0df["金额"] = pd.to_numeric(df["金额"], errors="coerce").fillna(0)# 去掉全空的行df.dropna(how="all", subset=["类别", "金额"], inplace=True)# 添加来源标记df["来源文件"] = filenamereturn dfdef read_all_reports(input_dir):"""读取目录下所有Excel文件并合并"""excel_files = glob.glob(os.path.join(input_dir, "*.xlsx")) +\glob.glob(os.path.join(input_dir, "*.xls"))print(f"在 {input_dir} 中找到 {len(excel_files)} 个Excel文件")all_data = []errors = []for filepath in excel_files:filename = os.path.basename(filepath)try:df = read_single_excel(filepath)all_data.append(df)print(f"{filename}: {len(df)} 行数据")except Exception as e:errors.append((filename, str(e)))print(f"{filename}: 读取失败 - {e}")if errors:print(f"\n有 {len(errors)} 个文件读取失败,请检查格式")if not all_data:print("没有成功读取任何文件")return pd.DataFrame()# 合并所有数据merged = pd.concat(all_data, ignore_index=True)print(f"\n合并完成,共 {len(merged)} 行数据")return merged# ========== 对比分析 ==========def compare_with_last_month(current_df, last_month_file):"""与上月数据对比,计算变化"""if not os.path.exists(last_month_file):print(f"上月汇总文件不存在: {last_month_file},跳过对比")return current_dflast_df = pd.read_excel(last_month_file)# 确保上月数据也有标准列if "部门" not in last_df.columns or "类别" not in last_df.columns:print("上月数据缺少标准列,跳过对比")return current_df# 按部门+类别汇总当月和上月金额current_summary = current_df.groupby(["部门", "类别"])["金额"].sum().reset_index()current_summary.rename(columns={"金额": "本月金额"}, inplace=True)last_summary = last_df.groupby(["部门", "类别"])["金额"].sum().reset_index()last_summary.rename(columns={"金额": "上月金额"}, inplace=True)# 合并对比comparison = pd.merge(current_summary, last_summary,on=["部门", "类别"],how="outer" ).fillna(0)# 计算变化comparison["变化金额"] = comparison["本月金额"] -comparison["上月金额"]# 避免除以零comparison["变化比例"] = comparison.apply(lambda row: row["变化金额"] /row["上月金额"]if row["上月金额"] != 0 else (float("inf") if row["变化金额"] != 0 else0),axis=1 )# 标记异常变化comparison["是否异常"] = comparison["变化比例"].abs() >DIFF_THRESHOLDreturncomparison# ========== 输出Excel ==========def write_summary_excel(merged_df, comparison_df, output_file):"""输出带格式的汇总Excel"""wb = Workbook()# 定义样式header_font = Font(bold=True, color="FFFFFF", size=11)header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")header_alignment = Alignment(horizontal="center", vertical="center")alert_fill = PatternFill(start_color="FFC7CE", end_color="FFC7CE", fill_type="solid")alert_font = Font(color="9C0006")thin_border = Border(left=Side(style="thin"),right=Side(style="thin"),top=Side(style="thin"),bottom=Side(style="thin"), )# ===== Sheet1: 明细数据 =====ws1 = wb.activews1.title = "明细数据"# 写表头for col_idx, col_name in enumerate(merged_df.columns, 1):cell = ws1.cell(row=1, column=col_idx, value=col_name)cell.font = header_fontcell.fill = header_fillcell.alignment = header_alignmentcell.border = thin_border# 写数据for row_idx, row in enumerate(merged_df.itertuples(index=False), 2):for col_idx, value in enumerate(row, 1):cell = ws1.cell(row=row_idx, column=col_idx, value=value)cell.border = thin_border# 金额列右对齐并加千分位if merged_df.columns[col_idx-1] == "金额":cell.number_format = '#,##0.00'cell.alignment = Alignment(horizontal="right")# 自动调整列宽for col in ws1.columns:max_len = max(len(str(cell.valueor"")) forcellincol)ws1.column_dimensions[col[0].column_letter].width = min(max_len+4, 35)# ===== Sheet2: 对比分析 =====if isinstance(comparison_df, pd.DataFrame) and not comparison_df.empty:ws2 = wb.create_sheet("对比分析")# 格式化百分比列display_df = comparison_df.copy()display_df["变化比例"] = display_df["变化比例"].apply(lambda x: f"{x:.1%}"if x!= float("inf") else"新增" )# 写表头headers = ["部门", "类别", "本月金额", "上月金额", "变化金额", "变化比例", "是否异常"]for col_idx, col_name in enumerate(headers, 1):cell = ws2.cell(row=1, column=col_idx, value=col_name)cell.font = header_fontcell.fill = header_fillcell.alignment = header_alignmentcell.border = thin_border# 写数据for row_idx, (_, row) in enumerate(display_df.iterrows(), 2):is_alert = row.get("是否异常", False)for col_idx, col_name in enumerate(headers, 1):value = row[col_name]# 布尔值转文字if col_name == "是否异常":value = "异常" if value else "正常"cell = ws2.cell(row=row_idx, column=col_idx, value=value)cell.border = thin_border# 金额列格式化if col_name in ["本月金额", "上月金额", "变化金额"]:cell.number_format = '#,##0.00'cell.alignment = Alignment(horizontal="right")# 异常行标红if is_alert:cell.fill = alert_fillif col_name == "是否异常":cell.font = alert_font# 列宽for col in ws2.columns:max_len = max(len(str(cell.valueor"")) forcellincol)ws2.column_dimensions[col[0].column_letter].width = min(max_len+4, 30)# 在对比表顶部加一行汇总total_current = comparison_df["本月金额"].sum()total_last = comparison_df["上月金额"].sum()alert_count = comparison_df["是否异常"].sum()ws2.insert_rows(1)ws2.cell(row=1, column=1, value=f"汇总:本月合计 ¥{total_current:,.2f} | 上月合计 ¥{total_last:,.2f} | 异常项 {alert_count} 个")ws2.cell(row=1, column=1).font = Font(bold=True, size=12)ws2.merge_cells(start_row=1, start_column=1, end_row=1, end_column=7)# 保存os.makedirs(os.path.dirname(output_file) or".", exist_ok=True)wb.save(output_file)print(f"\n汇总文件已保存: {output_file}")# ========== 主流程 ==========def main():print("="*50)print("Excel批量合并与对比工具")print(f"运行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")print("="*50)# 第一步:读取并合并所有报表print("\n>>> 读取当月各部门报表...")merged_df = read_all_reports(INPUT_DIR)if merged_df.empty:print("没有数据,退出")return# 第二步:与上月对比print("\n>>> 与上月数据对比...")comparison_df = compare_with_last_month(merged_df, LAST_MONTH_FILE)# 第三步:输出汇总Excelprint("\n>>> 生成汇总报表...")write_summary_excel(merged_df, comparison_df, OUTPUT_FILE)# 输出统计if isinstance(comparison_df, pd.DataFrame) and"是否异常"in comparison_df.columns:alert_count = comparison_df["是否异常"].sum()print(f"\n发现 {alert_count} 个异常项(变化超过 {DIFF_THRESHOLD:.0%})")print("完成。")if__name__ == "__main__":main()
==================================================Excel批量合并与对比工具运行时间: 2024-08-30 18:32:15==================================================>>> 读取当月各部门报表...在 ./reports/2024_08 中找到 12 个Excel文件 ✓ 市场部_8月费用.xlsx: 34 行数据 ✓ 技术部_8月费用明细.xlsx: 56 行数据 ✓ 行政部费用表.xlsx: 28 行数据 ✓ 人事部8月报表.xlsx: 19 行数据 ✓ 财务部_费用明细_202408.xlsx: 41 行数据 ✓ 销售部8月费用.xlsx: 63 行数据 ✓ 产品部费用.xlsx: 22 行数据 ✓ 运营部_8月费用表.xlsx: 37 行数据 ✓ 客服部8月报表.xlsx: 15 行数据 ✓ 法务部费用.xlsx: 8 行数据 ✓ 采购部8月费用.xlsx: 45 行数据 ✓ 研发二部费用明细.xlsx: 31 行数据合并完成,共 399 行数据>>> 与上月数据对比...>>> 生成汇总报表...汇总文件已保存: ./reports/summary_2024_08.xlsx发现 7 个异常项(变化超过 10%)完成。
打开生成的Excel,两个sheet。第一个是全部明细数据,列名整齐划一,金额带千分位格式。第二个是对比分析表,顶部一行汇总信息,异常项整行标红,一眼就能看到哪里不对。
7个异常项。我点开一看,市场部“推广费”涨了35%,技术部“服务器费用”涨了22%。这些变化手动对比的时候根本注意不到,因为数字散落在不同文件的不同位置。
这段代码里最值钱的其实是那个COLUMN_MAP字典。每个公司的部门报表格式都不一样,你不可能让12个部门统一格式。与其抱怨,不如做个映射。
碰到新的奇怪列名,往字典里加一行就行。用了几个月之后,那个字典基本上覆盖了所有情况。
有些部门喜欢在表格前面插几行空行,或者放个标题“XX部门X月费用明细”。read_single_excel函数里那段预览前5行的逻辑就是干这个的。它会自己找到真正的表头在哪里,然后从那里开始读数据。
这个设计救了我很多次。以前用pandas直接read_excel的时候,碰到表头不在第一行的就会读出一堆NaN,然后各种报错。
以前手动合并12个部门的表,加上对比上月数据,至少要两个小时。还经常出错,错了就得重来。
现在跑一次脚本,10秒出结果。我只要花几分钟看看异常项,确认哪些变化是合理的就行。每个月省下来的时间差不多够我多喝两杯咖啡了。
https://ima.qq.com/wiki/?shareId=f2628818f0874da17b71ffa0e5e8408114e7dbad46f1745bbd1cc1365277631c
