Python办公自动化:Excel数据处理与PDF合并脚本模板
pandas/openpyxl 和 pypdf 库,拆解列对齐、类型容错、大纲构建及异常防护等核心实现。
一、执行环境与依赖准备
本文脚本运行于 Python 3.9 及以上环境。核心依赖库主要有两个:用于表格数据读取与变换的 pandas(附带 openpyxl 用于读写 .xlsx 格式),以及用于底层 PDF 流组装与目录树生成的 pypdf。需要注意,早期的 PyPDF2 已在 3.0 版本后合并回归为官方推荐的 pypdf,新代码应避免引用废弃的 PyPDF2 包名。
# 在终端或虚拟环境中安装核心依赖
pip install pandas openpyxl pypdf
工程目录建议采用清晰的职责隔离结构。输入源文件与输出归档文件保持分离,避免因脚本反复执行导致已生成文件被作为源文件二次加载:
automation_project/
├── data_input/ # 待合并的原始 Excel 文件夹
├── pdf_input/ # 待排序与合并的 PDF 单页或分册
├── output/ # 清洗聚合后的结果表及成品 PDF
└── run_pipeline.py # 整合调度执行入口
二、Excel 批量数据处理脚本模板
企业报表汇总最常见的痛点是:文件名格式不规范、字段顺序存在微小漂移、数值列混杂空行或非法字符、日期格式不统一。处理批量表格时,关键在于建立“元数据校验 — 字段重映射 — 统一类型转换 — 纵向合并 — 统计汇总”的标准流水线。
1. Excel 批处理通用脚本模板
以下代码实现了安全扫描目录、自动忽略 Office 临时隐藏文件(如 ~$filename.xlsx)、映射列名、填补空缺并输出合并汇总报表的完整过程:
import sys
from pathlib import Path
import pandas as pd
def batch_process_excel(
input_folder: str,
output_filepath: str,
column_mapping: dict = None,
required_columns: list = None
) -> pd.DataFrame:
"""
扫描输入目录下所有 Excel 表格,执行字段对齐与清洗,并输出统一报表。
"""
input_path = Path(input_folder)
output_path = Path(output_filepath)
output_path.parent.mkdir(parents=True, exist_ok=True)
# 1. 筛选合法 xlsx 文件,过滤 ~$ 开头的临时锁定文件
excel_files = [
f for f in input_path.glob("*.xlsx")
if not f.name.startswith("~$") and f.is_file()
]
if not excel_files:
print(f"[警告] 目录 {input_folder} 下未找到合法的 .xlsx 文件。")
return pd.DataFrame()
data_frames = []
for file in sorted(excel_files):
try:
# 读取首个工作表,自动跳过完全为空的行和列
df = pd.read_excel(file, sheet_name=0)
if df.empty:
print(f"[跳过] 文件内容为空: {file.name}")
continue
# 2. 列名空格清理与别名映射
df.columns = [str(c).strip() for c in df.columns]
if column_mapping:
df.rename(columns=column_mapping, inplace=True)
# 3. 校验必填列;若缺失则补充 NaN 列,避免 concat 时对齐报错
if required_columns:
for col in required_columns:
if col not in df.columns:
df[col] = None
# 只保留业务需要的核心列
df = df[required_columns]
# 4. 追加元数据溯源字段
df["_源文件名"] = file.name
# 5. 清理完全重复项和无主键数据的空行
df.dropna(how="all", inplace=True)
data_frames.append(df)
print(f"[成功加载] {file.name} - 读取有效行数: {len(df)}")
except Exception as e:
print(f"[读取失败] 解析文件 {file.name} 出现异常: {e}", file=sys.stderr)
if not data_frames:
print("[错误] 未能成功解析任何表格数据。")
return pd.DataFrame()
# 6. 纵向合并所有分表
master_df = pd.concat(data_frames, ignore_index=True)
# 7. 数据格式收敛与清洗示例:去重与数值转化
if "金额" in master_df.columns:
master_df["金额"] = pd.to_numeric(master_df["金额"], errors="coerce").fillna(0.0)
# 8. 导出至 Excel:包含主表与简易透视汇总表
with pd.ExcelWriter(output_path, engine="openpyxl") as writer:
master_df.to_excel(writer, sheet_name="明细合集", index=False)
# 若包含业务分类字段,顺带生成汇总 Sheet
if "部门" in master_df.columns and "金额" in master_df.columns:
summary = master_df.groupby("部门", as_index=False)["金额"].agg(["count", "sum"])
summary.columns = ["部门", "记录笔数", "汇总总额"]
summary.to_excel(writer, sheet_name="部门统计", index=False)
print(f"[完成] 合并完成,共计 {len(master_df)} 行数据,已保存至: {output_path}")
return master_df
if __name__ == "__main__":
# 配置业务列名映射字典(将各部门叫法不一的字段统一为标准名)
mapping = {
"部门名称": "部门",
"所属部门": "部门",
"交易金额": "金额",
"报销金额": "金额",
"申请日期": "日期"
}
target_cols = ["部门", "姓名", "金额", "日期"]
batch_process_excel(
input_folder="./data_input",
output_filepath="./output/合并清洗汇总表.xlsx",
column_mapping=mapping,
required_columns=target_cols
)
1. 字符串日期与数字串的隐式转换:工号或身份证号若直接调用 to_numeric 会破坏前导零或精度丢失。对于标识型列,在 pd.read_excel 中应显式指定 dtype={"工号": str}。
2. 内存防范:若待合并表格数量达到数千个或单表超 50 万行,不宜无脑将全部 DataFrame 保存在内存列表中,应分批写入 CSV 或 SQLite 中间层。
三、PDF 文档合并与书签构建脚本模板
单将多个 PDF 拼接成一个大文件并不困难,真正决定交付文件可用性的是书签大纲(Outline / Bookmarks)的编排。没有大纲的长 PDF 文件在移动端或阅读器中检索极为低效。理想的合并脚本必须在依次写入各章节内容时,精准记录当前的页码偏移量(Page Offset),并自动生成清晰的树状目录导航。
1. PDF 合并与书签注入脚本模板
现代 pypdf 库的 PdfWriter 提供了极其高效的 append() 方法,该方法支持传入文件对象、路径,并且原生支持 outline_item 参数直接在一级目录添加书签。针对复杂的二级嵌套书签,可以使用 add_outline_item() 配合父级书签指针实现。
import sys
from pathlib import Path
from pypdf import PdfReader, PdfWriter
def merge_pdfs_with_bookmarks(
file_configs: list,
output_pdf_path: str,
doc_title: str = "合并业务报告"
) -> bool:
"""
按顺序合并多个 PDF 文件,并构建具有层级结构的书签导航。
:param file_configs: 包含文件路径及书签配置的列表,例如:
[
{"path": "cover.pdf", "title": "一、报告封面"},
{"path": "main.pdf", "title": "二、主要经营分析", "children": [
{"title": "2.1 营收增长概览", "page_offset": 0},
{"title": "2.2 成本控制分析", "page_offset": 4}
]},
{"path": "appendix.pdf", "title": "三、详细附录清单"}
]
:param output_pdf_path: 合成后的 PDF 输出路径
:param doc_title: 注入到 PDF 元数据的文档标题
"""
writer = PdfWriter()
current_page_index = 0
valid_file_count = 0
output_path = Path(output_pdf_path)
output_path.parent.mkdir(parents=True, exist_ok=True)
for item in file_configs:
file_path = Path(item["path"])
if not file_path.exists():
print(f"[跳过] 找不到指定 PDF 文件: {file_path}", file=sys.stderr)
continue
try:
reader = PdfReader(file_path)
# 处理受密码保护的文件检查
if reader.is_encrypted:
print(f"[错误] 文件受到加密保护,请先解密: {file_path.name}", file=sys.stderr)
continue
num_pages = len(reader.pages)
if num_pages == 0:
print(f"[跳过] PDF 文件页数为空: {file_path.name}")
continue
# 1. 将该 PDF 所有页面追加到总写入流中
writer.append(reader)
# 2. 插入当前文件对应的一级书签,直接跳转到合并后对应的起始页面
parent_bookmark = writer.add_outline_item(
title=item.get("title", file_path.stem),
page_number=current_page_index
)
# 3. 递归或迭代插入二级子书签(如果有)
if "children" in item and isinstance(item["children"], list):
for child in item["children"]:
# 子书签页码 = 当前文件的基准页码 + 子章节相对页码
sub_offset = child.get("page_offset", 0)
target_page = min(current_page_index + sub_offset, current_page_index + num_pages - 1)
writer.add_outline_item(
title=child.get("title", "子章节"),
page_number=target_page,
parent=parent_bookmark
)
print(f"[已合并] {file_path.name} | 页数: {num_pages} | 起始页码: {current_page_index + 1}")
current_page_index += num_pages
valid_file_count += 1
except Exception as e:
print(f"[异常] 处理文件 {file_path.name} 时出错: {e}", file=sys.stderr)
if valid_file_count == 0:
print("[错误] 未能合并任何有效页面,终止生成。")
return False
# 4. 写入元数据
writer.add_metadata({
"/Title": doc_title,
"/Producer": "Python Office Automation Pipeline (pypdf)"
})
# 5. 持久化输出
with open(output_path, "wb") as f_out:
writer.write(f_out)
print(f"\n[成功] 最终合成文档生成完毕!总页数: {current_page_index},路径: {output_path}")
return True
if __name__ == "__main__":
# 待合并文件清单及书签配置示例
tasks = [
{
"path": "./pdf_input/01_封面.pdf",
"title": "一、封面与目录"
},
{
"path": "./pdf_input/02_主要报告.pdf",
"title": "二、业务运行总览",
"children": [
{"title": "2.1 关键运营指标", "page_offset": 0},
{"title": "2.2 风险排查清单", "page_offset": 2}
]
},
{
"path": "./pdf_input/03_附件.pdf",
"title": "三、数据附录"
}
]
merge_pdfs_with_bookmarks(
file_configs=tasks,
output_pdf_path="./output/2026年度业务综合汇编报告.pdf",
doc_title="2026年度综合运营报告"
)
1. 页码索引是从 0 开始的:在 add_outline_item 中,参数 page_number=0 代表跳转至文档第 1 页。如果传入的数值大于等于总页数,将会导致阅读器跳转失效或在保存时抛出 IndexError。
2. 文件句柄生命周期:直接使用 writer.append(reader) 会在内存中维持引用,应确保文件流直到 writer.write() 结束前未被意外关闭或移动。
四、自动化脚本端到端执行流程
在实际生产环境中,Excel 数据清洗和 PDF 合并往往不是孤立的,而是前后衔接的流水线任务。例如:先合并 10 个业务组的 Excel 报表,清洗后生成汇总分析,再结合已有的模板文件生成附录,最后连同说明页全部打包为交付 PDF。
2. 一体化调度主程序实现
通过统一的流程编排函数,使数据流在处理完表格后直接触发后续拼装动作。以下代码演示了完整的集成模式,带有规范的异常捕获与执行日志输出:
import logging
from pathlib import Path
# 配置结构化日志输出
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
def run_pipeline(workspace_root: str):
root = Path(workspace_root).resolve()
excel_dir = root / "data_input"
pdf_dir = root / "pdf_input"
out_dir = root / "output"
out_dir.mkdir(parents=True, exist_ok=True)
logging.info(f"工作空间初始化完成: {root}")
# 阶段一:Excel 数据合并与清洗
logging.info(">>> 启动阶段一:Excel 批处理与清洗...")
mapping = {"项目名称": "项目", "结算金额": "金额", "申报人": "责任人"}
req_cols = ["项目", "金额", "责任人"]
merged_excel_path = out_dir / "汇总清洗明细.xlsx"
df = batch_process_excel(
input_folder=str(excel_dir),
output_filepath=str(merged_excel_path),
column_mapping=mapping,
required_columns=req_cols
)
if df.empty:
logging.warning("阶段一未清洗出有效数据,请检查源数据表。")
else:
logging.info(f"阶段一完成:生成表格包含 {len(df)} 条明细。")
# 阶段二:PDF 合并与书签构建
logging.info(">>> 启动阶段二:PDF 章节拼接与目录编排...")
# 构造有序章节队列(确保物理文件存在)
pdf_queue = [
{"path": str(pdf_dir / "01_封面与目录.pdf"), "title": "1. 封面与引言"},
{"path": str(pdf_dir / "02_业务主报告.pdf"), "title": "2. 主体分析报告", "children": [
{"title": "2.1 统计报表", "page_offset": 0},
{"title": "2.2 异常分析", "page_offset": 2}
]},
{"path": str(pdf_dir / "03_附录明细.pdf"), "title": "3. 数据附录与凭证"}
]
final_pdf_path = out_dir / "最终交付汇编文档.pdf"
success = merge_pdfs_with_bookmarks(
file_configs=pdf_queue,
output_pdf_path=str(final_pdf_path),
doc_title="季度自动化整合业务报告"
)
if success:
logging.info(f"流水线执行成功!最终产物生成完毕: {final_pdf_path}")
else:
logging.error("阶段二 PDF 合并存在错误,请排查输入文件。")
if __name__ == "__main__":
run_pipeline(".")
五、生产环境稳定性与自检清单
在从个人本机脚本走向部门协作或定时调度任务时,必须考虑跨平台路径处理、进程文件锁定以及脏数据容错等防御性机制。
核心排错检查表
| 现象 / 报错 | 可能原因 | 定位与解决方式 |
|---|---|---|
PermissionError: [Errno 13] |
目标 Excel 或 PDF 文件已被用户在 Office / WPS 中打开锁定 | 在执行前通过 open(path, 'a') 试探文件是否被锁定,或在写入失败时提示用户关闭占用程序。 |
KeyError: '列名' |
某业务员提交的表格列名多了首尾不可见空格或换行符 | 读取后第一时间执行 df.columns = [c.strip() for c in df.columns],或使用映射字典。 |
| 合并后的 PDF 书签点击跳转错误 | 子章节页码采用了绝对页码,未累加前序各文件的页面总偏移量 | 严格维护 current_page_index += len(reader.pages) 并在添加书签时累加该基准偏移。 |
| 合并后总行数对不上 | 源表底部包含合并单元格的备注行或制表人签名行 | 使用 df.dropna(subset=['核心主键字段']) 准确剔除尾部注释非数据行。 |
借助上述两套模板与组合编排方式,常规的多文件表格整理与 PDF 装配任务即可脱离人工重复劳动,在数十秒内实现规范化、零差错的自动化批量产出。
免责声明:本文内容仅供参考,不代表本站观点。转载或引用请注明出处。









