在自动化报表和数据分析场景中,经常需要把多个CSV或Excel文件中的记录合并、清洗后再输出。Python凭借丰富的标准库和第三方包,能够非常高效地完成这类批量读写任务。下面以实际目录下的多文件处理为例,逐步展开实现方式。

一、环境准备与文件遍历
批量处理的第一步是定位目标文件。我们假设当前工作目录有一个叫做 data 的子文件夹,里面混合存放了 .csv 和 .xlsx 文件。可以使用 os 模块或者 pathlib 来列举文件,后者语法更现代清晰。
在处理之前,建议先确认已安装 pandas 以及对应的 Excel 引擎。读取 .xlsx 通常需要 openpyxl,而写入带格式的文件也依赖它。可以通过 pip install pandas openpyxl 来完成安装。下面的代码展示了如何筛选出指定类型的文件:
from pathlib import Path
data_dir = Path('./data')
csv_files = list(data_dir.glob('*.csv'))
xlsx_files = list(data_dir.glob('*.xlsx'))
print(f'找到CSV文件: {len(csv_files)}个')
print(f'找到Excel文件: {len(xlsx_files)}个')
这种写法比 os.listdir 配合字符串判断更直观,而且 Path 对象自带跨平台路径处理能力。拿到文件列表后,就可以针对不同类型调用不同的读取函数。
二、使用csv模块批量读取CSV
Python标准库中的 csv 模块适合处理结构简单、数据量中等的CSV文件。它不依赖第三方库,在受限环境中也能运行。核心思路是用 csv.DictReader 按行读取,将每行映射为字典,便于后续字段对齐。
当多个CSV文件的列顺序不同但列名一致时,DictReader 能自动按表头归位。如果某些文件缺失列,可以在累加时给默认值。下面示例把多个CSV合并到一个列表里:
import csv
from pathlib import Path
rows = []
for f in Path('./data').glob('*.csv'):
with open(f, newline='', encoding='utf-8-sig') as fp:
reader = csv.DictReader(fp)
for r in reader:
# 补充可能缺失的字段
r['source_file'] = f.name
rows.append(r)
print(f'共读取记录: {len(rows)}条')
这里使用 utf-8-sig 编码是为了兼容带BOM的Windows导出文件,避免首列名出现不可见字符。虽然 csv 模块轻量,但面对上百万行数据时,纯Python循环效率偏低,此时应转向 pandas。
三、用pandas批量读取与合并
pandas 提供了 read_csv 和 read_excel 两个高层接口,能够直接返回 DataFrame 结构,并自带类型推断、缺失值处理等能力。批量读取时,通常先逐个读入再使用 concat 纵向拼接。
实际业务中常见问题是不同文件表头大小写不一致或多余空格。可以在读取后统一清洗列名。以下代码演示了混合读取CSV与Excel并合并:
import pandas as pd
from pathlib import Path
frames = []
for f in Path('./data').iterdir():
if f.suffix == '.csv':
df = pd.read_csv(f, encoding='utf-8-sig')
elif f.suffix in ('.xlsx', '.xls'):
df = pd.read_excel(f, engine='openpyxl')
else:
continue
df.columns = [c.strip().lower() for c in df.columns]
df['source_file'] = f.name
frames.append(df)
merged = pd.concat(frames, ignore_index=True)
print(merged.shape)
concat 时设置 ignore_index=True 可重置索引,防止原文件行号残留造成混淆。如果某些Excel sheet不止一个,还可以通过 sheet_name=None 读取全部工作表再展开。pandas的优势在于后续过滤、分组、透视都能一行搞定。
四、批量写入CSV与Excel
合并完成后,输出环节同样重要。写CSV用 to_csv,写Excel用 to_excel。若需将一个总表按某个字段拆分回多个文件,可以结合 groupby 实现。
下面示例将合并数据按 source_file 来源标记导出为单一Excel,并演示了如何控制Excel写入参数以避免索引列:
# 写出为单一Excel文件,不包含行索引
merged.to_excel('./output/merged.xlsx', index=False, engine='openpyxl')
# 按某分类字段拆分写入多个CSV
for key, group in merged.groupby('category'):
out_name = f'./output/cat_{key}.csv'
group.to_csv(out_name, index=False, encoding='utf-8-sig')
使用 openpyxl 引擎时,还可以先获取 workbook 对象来设置列宽、冻结首行等。不过要注意,频繁打开关闭Excel文件会有磁盘开销,批量拆分建议先缓存在内存再统一写。
五、常见坑与处理建议
编码问题是CSV批量处理的高频故障点。Linux生成的UTF-8无BOM文件在Windows Excel中打开可能乱码,此时可改存为 utf-8-sig 或在写入时指定 encoding='gbk' 适配老版Excel。
另一个坑是Excel单元格类型。数字被存为文本会导致求和错误,read_excel 默认会推断,但若源文件格式不规范,可传入 dtype 参数强制转换。日期字段建议使用 parse_dates 显式解析,避免后续排序时出现字符串比较异常。
| 问题现象 | 可能原因 | 解决办法 |
|---|---|---|
| 首列名带问号 | UTF-8 BOM未处理 | 用utf-8-sig读取 |
| Excel数字无法计算 | 文本型数字 | dtype指定或清洗 |
| 合并后列错位 | 表头空格大小写 | 统一lower和strip |
只要预先规范化列名与类型,批量管道就能稳定跑通。对于超大规模数据,可改用分块读取 chunksize 参数,降低内存占用。
六、小结与扩展
通过上述步骤,我们用标准库与pandas覆盖了从遍历、读取、合并到写出的完整链路。对于定时任务,可将脚本放入 cron 或 Windows计划任务,实现每日报表自动归集。
如果下游系统要求特定JSON或数据库落地,只需把 to_excel 换成 to_json 或 SQLAlchemy 的 to_sql。理解文件IO与DataFrame转换原理后,无论格式如何组合,都能快速写出健壮的批处理脚本。