最近在帮一个做运营的朋友处理数据他丢过来一堆Excel和CSV文件说是从不同后台导出的零零散散加起来有上百个命名也不规范有叫导出_0512.xlsx的也有直接叫data.csv的还有几个打开就乱码。他想要的无非是把这些文件合并成一个总表再顺便把日期格式、电话号码那几列收拾干净方便后面做透视分析。这种场景说实话太常见了。只要你的工作跟数据沾边迟早会遇到这类批量处理Excel和CSV文件的需求。用Python做这件事核心思路并不复杂无非是定位文件、读取解析、清洗逻辑、写回结果这四步。真正磨人的不是代码本身而是文件编码、数据类型、Excel公式残留这些细节。这篇文章我把整个流程拆开讲从环境准备到完整代码再到我踩过的几个坑全部摆出来你照着改改就能用。1. 先理清需求这批文件到底要怎么处理1.1 常见批处理场景拆解动手写代码之前先把需求拆明白。大部分人遇到的批量处理场景其实就那么几类第一类是合并汇总。多个门店的销售报表、多个后台的投放数据、多个月的财务明细结构基本相同需要纵向堆叠成一个总表。这是最常见的需求。第二类是清洗规整。日期字段有的是2024/5/1有的是20240501字符串里掺杂全角空格手机号带一个个位数缺失的脏数据数值列被Excel存成了文本。这些脏数据如果不处理后续做透视表或者导入数据库全是问题。第三类是格式转换。客户要CSV系统只给Excel或者反过来把一堆CSV转成Excel发给别人。还有的是把一个大CSV按行数拆成多个小文件方便分批导入。第四类是抽取汇总。从几十个格式各异的文件里把符合条件的若干行或若干字段抽出来形成一个新表。你可以对照自己的实际需求看属于哪一类。大部分时候是上面几种的组合。明确需求的目的是为了选择合适的库和读取策略——比如你只需要转格式那pandas其实是杀鸡用牛刀用csv模块就够了要做复杂清洗和合并pandas是最稳妥的选择。1.2 选对轮子pandas、openpyxl还是csv模块Python处理Excel和CSV的库不少但真正派得上用场的主要就三个层次。最底层的是标准库里的csv模块。它只负责解析CSV文本速度极快内存占用小适合处理超大文件或者做简单的格式转换。但它不关心数据类型读出来全是字符串也不支持Excel文件。中间层是openpyxl专门操作xlsx文件。能读取单元格、修改样式、写公式、插入图表但它是一个单元格级别的API处理逻辑要自己写对几百兆的大数据量效率一般。顶层是pandas数据科学生态的核心库。它基于numpy读CSV用C解析器读Excel时内部调用openpyxl或xlrd。它最强大的地方是把文件读成DataFrame这种二维表结构之后筛选、分组、去重、合并都是几个方法的事。我的建议很直接只要涉及到多文件合并、字段清洗、类型转换直接用pandas不要自己造轮子。如果文件特别大超过1GB的CSV或者只需要简单转码再考虑用csv模块分块处理。2. 环境准备一劳永逸的Python数据处理环境2.1 安装Python与关键库如果你电脑上还没有Python环境先去官网下载安装包。Windows用户注意一件事安装时务必勾选Add Python to PATH否则后面在命令行里敲python会提示找不到命令。装好Python之后建议建一个独立的虚拟环境避免不同项目之间依赖冲突。Windows下在项目目录里执行python -m venv venv venv\Scripts\activatemacOS或者Linux下执行python3 -m venv venv source venv/bin/activate虚拟环境激活之后再安装依赖库pip install pandas openpyxlpandas是核心openpyxl是pandas读写xlsx文件的后端引擎。如果你还要处理xls格式的老文件加装一个xlrdpip install xlrd这里有一个容易踩的版本坑xlrd2.0以上的版本不再支持xls格式只支持xlsx。如果你确实需要处理老版的.xls文件要指定安装xlrd1.2.0。不过现在绝大多数系统导出的都是xlsx了装不装看你的实际情况。2.2 文件组织处理前先把目录结构理顺代码写得好不好是一回事文件放得乱不乱是另一回事。我见过不少朋友把待处理文件散落在桌面、下载文件夹、微信传输助手里结果脚本一跑要么找不到文件要么把源文件覆盖了。建议你建立这样的目录结构batch_process/ ├── input/ # 待处理文件全部放这里 ├── output/ # 处理结果统一输出到这里 ├── processed/ # 已经处理过的源文件归档 └── script.py # 处理脚本这样做的原因有三个第一脚本只扫描input目录不碰其他位置避免误读无关文件。第二结果写到output目录不会覆盖源文件。第三processed目录做归档处理完的文件挪过去下次再跑脚本不会重复处理同一批文件。这个习惯看起来很简单但能帮你省掉大量卧槽怎么处理了两遍的麻烦。特别是自动化脚本配合定时任务运行时归档目录就是你的操作日志。3. 批量读取核心实现的第一步3.1 用glob或pathlib找到所有目标文件写代码的第一步是把所有待处理的文件路径找出来。glob模块是最直接的方式import glob import os # 找到input目录下所有的xlsx文件 xlsx_files glob.glob(input/*.xlsx) print(f找到 {len(xlsx_files)} 个xlsx文件) # 找到所有csv文件包含子目录用**递归匹配 csv_files glob.glob(input/**/*.csv, recursiveTrue)如果你用的是Python 3.4以上我更推荐用pathlib它返回的是Path对象链式操作更方便from pathlib import Path input_dir Path(input) xlsx_files list(input_dir.glob(*.xlsx)) csv_files list(input_dir.rglob(*.csv)) # 递归搜索所有子目录注意到glob()只匹配当前目录rglob()会递归匹配所有子目录。如果文件分散在多层目录里用rglob。如果只是平铺在一个目录里glob就够了。还有一个细节文件后缀有的是.xlsx有的是.XLSXWindows不区分大小写但Linux区分。稳妥起见匹配时用i后缀或者干脆同时匹配两种excel_files list(input_dir.glob(*.xlsx)) list(input_dir.glob(*.XLSX))如果文件数量特别多上面的方式可能会因为列表过大占一点内存。更优雅的做法是直接用生成器逐个处理for file in input_dir.glob(*.xlsx): process_one(file)这样每处理一个文件就释放一个文件句柄内存占用更平稳。3.2 读取Excel和CSV的正确姿势拿到文件路径之后读取本身不难但有几个参数非常关键用不好就会出现各种诡异问题。读取CSVimport pandas as pd # 基础读取 df pd.read_csv(input/data.csv) # 如果遇到乱码通常是编码问题 df pd.read_csv(input/data.csv, encodingutf-8-sig) # 如果列名是中文或者有特殊字符可以指定列名 df pd.read_csv(input/data.csv, dtype{phone: str})读取Excel# 读取第一个工作表 df pd.read_excel(input/report.xlsx) # 读取指定工作表 df pd.read_excel(input/report.xlsx, sheet_name5月) # 读取所有工作表返回一个字典 all_sheets pd.read_excel(input/report.xlsx, sheet_nameNone)读取Excel时我通常建议把表头所在的行数明确指定。很多表格前面还有一两行标题说明如果不指定表头行pandas会把第一行数据当成列名df pd.read_excel(input/report.xlsx, header0) # 默认第一行是列名 df pd.read_excel(input/report.xlsx, header2) # 跳过前两行第三行是列名这个参数没有固定答案你必须先打开文件看一眼表头实际在哪一行。我自己的习惯是遇到那种表头不在第一行的文件先用headerNone读一遍看看原始结构再决定从哪一行开始解析。3.3 迅捷的批量合并代码把读取封装好之后批量合并的代码其实很短import glob import pandas as pd from pathlib import Path input_dir Path(input) output_file Path(output/merged.xlsx) all_data [] failed_files [] # 处理Excel文件 for file in input_dir.glob(*.xlsx): try: df pd.read_excel(file) # 增加一列记录数据来源 df[来源文件] file.name all_data.append(df) print(f成功读取: {file.name}, 形状: {df.shape}) except Exception as e: failed_files.append((file.name, str(e))) print(f读取失败: {file.name}, 错误: {e}) # 处理CSV文件 for file in input_dir.glob(*.csv): try: df pd.read_csv(file, encodingutf-8-sig) df[来源文件] file.name all_data.append(df) print(f成功读取: {file.name}, 形状: {df.shape}) except Exception as e: failed_files.append((file.name, str(e))) print(f读取失败: {file.name}, 错误: {e}) # 合并所有DataFrame if all_data: merged pd.concat(all_data, ignore_indexTrue, sortFalse) print(f合并完成总行数: {len(merged)}) # 写回Excel merged.to_excel(output_file, indexFalse) print(f已保存到: {output_file}) else: print(没有读取到任何数据)这段代码有几个地方值得解释pd.concat(all_data, ignore_indexTrue)的作用是把多个DataFrame纵向堆叠。ignore_indexTrue表示重新生成连续的索引否则你合完之后索引是乱的后面切片容易踩坑。sortFalse表示合并列名时不排序。如果不同文件的列顺序不一致pandas对齐的依据是列名而不是位置。这既是好事也是隐患好事是只要列名一样顺序错乱也能正确对齐隐患是如果某个文件多了一列合并后会出现NaN填充你必须检查一轮。增加来源文件列是我特别推荐的一个习惯。等你合完数据发现某一行数据有明显问题时可以通过来源文件快速定位是哪个原始文件出了问题不用翻遍几百个文件。3.4 列对齐问题不同文件结构不一致怎么办上面这段代码有一个理想前提所有文件的列名基本一致。但现实中经常遇到的情况是文件A有8列文件B有10列文件A叫销售额文件B叫销售金额。遇到这种情况你有两个选择第一个选择是只保留核心列。先定一个标准列清单读取每个文件之后通过列名筛选出需要的列required_columns [日期, 门店, 销售额, 订单量] def normalize_columns(df): # 只保留标准列中存在于当前df的列 existing [col for col in required_columns if col in df.columns] return df[existing] df normalize_columns(df)第二个选择是统一列名映射。建立一个字典把不同叫法映射到标准列名column_mapping { 销售金额: 销售额, 成交额: 销售额, 订单数: 订单量, 单量: 订单量, } def normalize_columns(df): # 重命名列 df df.rename(columnscolumn_mapping) return df这两种方式可以结合使用。先重命名统一叫法再筛选固定列。这样不管来多少种格式的文件最后合并出来的表结构一定是完全统一的后面做透视和汇总才不会出幺蛾子。4. 批量清洗数据规整的关键环节4.1 处理空值和重复行读进来之后第一批要收拾的是空值和重复行。空值处理没有一个万能方案要看数据类型和分析场景# 删除全部为空的行 df df.dropna(howall) # 删除某一列为空的行 df df.dropna(subset[订单号]) # 数值列缺失填充为0 df[销售额] df[销售额].fillna(0) # 文本列缺失填充为未知 df[门店名称] df[门店名称].fillna(未知) # 日期列缺失填充为指定的默认日期 df[订单日期] df[订单日期].fillna(1970-01-01)我处理数据时的判断标准是这个字段缺失了会导致它所在的那一行数据失去意义吗如果订单号缺失那这一行根本没有跟踪价值直接删如果备注缺失只是个别记录没有备注不影响主数据就让它空着或者填无。drop_duplicates去重也有讲究# 全行完全重复才去重 df df.drop_duplicates() # 根据指定列去重保留第一条 df df.drop_duplicates(subset[订单号]) # 根据指定列去重保留最后一条 df df.drop_duplicates(subset[订单号], keeplast)实际场景里订单号重复往往意味着同一条记录被导出了两次。用subset[订单号]去重是比较稳妥的做法。但有一个例外如果一个订单号下面包含了多个子商品那就是合法的多行不能去重。所以去重前先看清楚数据的粒度单位。4.2 日期、文本和数值的格式统一日期格式统一大概是清洗里最让人头疼的一环。同一个2024/5/1、20240501、2024-05-01、2024年5月1日并存的情况我见过太多次了。pandas的pd.to_datetime是强大的工具但也不是万能的# 尝试自动解析多种格式 df[日期] pd.to_datetime(df[日期], errorscoerce) # 统一输出为指定格式 df[日期] df[日期].dt.strftime(%Y-%m-%d)errorscoerce的意思是解析失败时置为NaTNot a Time而不是抛异常让程序死掉。这样既能保住程序运行又能让你之后检查哪些日期是异常值。如果原始日期列里混有20240501这种数字形态需要先转成字符串再解析df[日期] pd.to_datetime(df[日期].astype(str), format%Y%m%d, errorscoerce)数值列清洗比较常见的问题是数字被存储成文本或者字符串里带千分位逗号。前者在pandas里表现为dtype为object后者表现为1,234.56这种字符串。处理方式# 去掉千分位逗号再转数值 df[销售额] df[销售额].astype(str).str.replace(,, ) df[销售额] pd.to_numeric(df[销售额], errorscoerce) # 把纯粹的文本型数字转数值 df[销售额] pd.to_numeric(df[销售额], errorscoerce)文本列清洗最常见的坑是首尾空格、全角空格和不可见字符# 去掉首尾空格 df[门店名称] df[门店名称].str.strip() # 把全角空格和全角字符转半角 df[门店名称] df[门店名称].str.replace( , ).str.replace(, ,) # 替换空字符串为NaN方便后续统一处理 df[门店名称] df[门店名称].replace(, pd.NA)字符串方法.str.strip()只能处理首尾空格如果中间不小心混入了空格比如北京 朝阳店那需要str.replace( , )把所有空格去掉或者按实际情况保留。4.3 手机号和身份证号别让pandas吃了你的前导零这是一个所有新手都会踩的坑。如果你的表格里有一列是手机号或者身份证号直接read_excel或者read_csv读进来你会发现号码变成了科学计数法或者前导零消失了。比如010-12345678变成1012345678001234变成1234。正确解法是读取的时候就指定dtype为字符串# 读取CSV时指定 df pd.read_csv(input/data.csv, dtype{手机号: str, 身份证号: str}) # 读取Excel时指定 df pd.read_excel(input/report.xlsx, dtype{手机号: str, 身份证号: str}) # 已经读进来了再补救 df[手机号] df[手机号].astype(str).str.replace(.0, , regexFalse)更稳妥的方案是如果你发现号码列被读成了带小数点的科学计数法比如1.38234E10用上述的str.replace(.0, , regexFalse)配合astype(str)就能还原。最保险的办法还是建一个dtype映射字典任何可能被误判的列都显式指定为字符串string_columns [订单号, 手机号, 身份证号, 银行卡号, 门店编码] dtype_dict {col: str for col in string_columns if col in df.columns} df pd.read_excel(file, dtypedtype_dict)5. 批量输出写回Excel和CSV5.1 Excel输出多表写入和格式控制合并清洗完之后输出是最简单的一步但也有几个细节。基本输出df.to_excel(output/result.xlsx, indexFalse)indexFalse几乎必加否则pandas会把默认的0、1、2索引写成第一列打开文件的时候多出一列不知道哪来的数字。多表写入如果想把多个结果写入同一个Excel的不同Sheetwith pd.ExcelWriter(output/多表结果.xlsx) as writer: df_sales.to_excel(writer, sheet_name销售明细, indexFalse) df_summary.to_excel(writer, sheet_name汇总, indexFalse) df_bad.to_excel(writer, sheet_name异常数据, indexFalse)这个能力在生成日报或者数据检查报告的时候特别方便。我经常把正常数据异常数据处理日志写在同一个工作簿的不同Sheet里一次交付对方不用到处找文件。冻结首行如果导出的表有几万行用户往下滚动时看不到表头会很痛苦。pandas原生写Excel时不支持冻结窗格但可以用openpyxl二次加工with pd.ExcelWriter(output/result.xlsx, engineopenpyxl) as writer: df.to_excel(writer, indexFalse) # 获取工作表然后冻结首行 ws writer.sheets[Sheet1] ws.freeze_panes A2freeze_panes A2代表以A2为左上角区域固定上面的第一行滚动时表头始终可见。这是个很小的操作但对使用体验的提升非常明显。5.2 CSV输出编码是绕不过去的坎CSV输出的核心问题是编码。用Excel打开CSV文件时如果中文乱码大概率是编码没对。这里是一套经过多次验证的经验# 如果需要用Excel打开就用utf-8-sig df.to_csv(output/result.csv, indexFalse, encodingutf-8-sig) # 如果只是给程序读普通utf-8即可 df.to_csv(output/result.csv, indexFalse, encodingutf-8) # 如果对接的是老系统指定gbk编码 df.to_csv(output/result_gbk.csv, indexFalse, encodinggbk)utf-8-sig和utf-8的区别在于前者会在文件开头写入一个BOM头Excel识别到BOM就知道这是一个UTF-8编码的文件不乱码。这也解释了为什么很多程序员用纯utf-8生成的CSV用户用Excel打开却乱码。如果写入gbk时遇到某些字符GBK不认识比如瀞这种生僻字可以加一个errorsignore主动丢掉无法编码的字符但这有风险最好是换成如下方式df.to_csv(output/result.csv, indexFalse, encodinggb18030)gb18030是GBK的超集能覆盖几乎所有的中文字符兼容性更好。6. 实操心得效率翻倍的进阶技巧6.1 用dtype字典和converters提升读取速度批量处理几百个文件时读取速度可能成为瓶颈。三个立竿见影的提速技巧第一只读取需要的列。如果文件有50列但你只需要5列用usecols指定df pd.read_csv(file, usecols[日期, 门店, 销售额])这个参数能显著减少内存占用。同理pd.read_excel也支持usecols比如usecolsA:C表示只读A到C列。第二读取前先设置好dtype让pandas不要做类型推断。类型推断本身是有成本的而且可能推错。你直接告诉它每列是什么类型读取速度反而更快dtype_dict {手机号: str, 金额: float, 日期: str} df pd.read_csv(file, dtypedtype_dict)第三用CSV的C引擎而不是Python引擎。pd.read_csv默认就是C引擎但如果你传了enginepython就会变慢。某些正则表达式参数会强制走Python引擎尽量避免。处理上千万行的大CSV时还可以分块读取chunk_size 100000 chunks [] for chunk in pd.read_csv(big_file.csv, chunksizechunk_size): # 每块单独清洗 chunk[销售额] pd.to_numeric(chunk[销售额], errorscoerce) chunks.append(chunk) result pd.concat(chunks, ignore_indexTrue)分块的核心不在于省时间而在于控制峰值内存。一次性读入1GB的CSVpandas可能需要额外2-3GB内存来完成类型转换小内存机器直接就崩了。分块处理后单次峰值内存被限制在几十MB级别。6.2 异常文件的处理别让一个坏文件毁掉整个批次几十个文件里有个别文件结构异常、编码特殊、甚至损坏是批量处理时最常见的翻车原因。我的做法是永远用try-except把单个文件的处理包起来并且把失败信息记录到单独的失败清单里results [] errors [] for file in all_files: try: df pd.read_excel(file) results.append(df) except Exception as e: errors.append({file: str(file), error: str(e)}) # 处理完所有文件后输出失败清单 if errors: error_df pd.DataFrame(errors) error_df.to_csv(output/read_errors.csv, indexFalse, encodingutf-8-sig)这样设计的好处是一个文件读不了不会影响其他文件的处理联调的时候也只需要看错误清单而不是盯着终端日志一行行滚。另外异常的类型值得留意。文件已被占用这个错误在Windows下很常见——Excel还没关掉pandas就去读它就会抛PermissionError。处理这类文件前可以用一个小函数检测import os def is_file_locked(file_path): try: with open(file_path, a): return False except (PermissionError, OSError): return True测试发现文件被占用就提示用户先关闭Excel或者跳过并记录而不是让程序崩溃退出。6.3 写日志跑了那么多文件怎么知道发生了什么批量处理脚本不像交互程序跑完了也没有界面告诉你这边处理成功那边失败。所以给自己留一份日志非常必要。最简单的方式是Python自带的logging模块import logging logging.basicConfig( filenameoutput/process.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, encodingutf-8 ) logging.info(开始处理共发现 %d 个文件, len(all_files)) logging.warning(文件 %s 读取失败, file.name) logging.info(合并完成共 %d 行, len(merged))日志的意义不只在于当下排查问题更在于你后续做自动化调度时可以通过日志回溯某天某个时段的处理情况。配合上文的失败清单Excel文件处理过程的透明度一下子就上来了。7. 常见问题速查表问题现象可能原因解决办法CSV读出来中文乱码编码用utf-8但没有BOM读取时加encodingutf-8-sigExcel打开CSV乱码写入时用纯utf-8没有BOM写入时加encodingutf-8-sig手机号/身份证显示科学计数法pandas默认推断为数值读取时dtype{手机号: str}前导零消失同上cast转str并补零或读取时指定dtype日期变成一串数字Excel日期序列值被直接读取读取时parse_dates或用pd.to_datetime转换合并后出现很多NaN某个文件多出额外列名对齐列名筛选标准列后再合并报错Excel file format cannot be determined文件其实是HTML/假扩展名先打开文件看真实格式或用openpyxl读真实内容报错PermissionError文件在Excel中被打开关闭Excel占用的文件再重试处理大文件时内存爆炸一次性读入数据量过大用chunksize分块读取或提取usecols减少列数写入excel后公式列丢失pandas不保留公式结果用openpyxl直接操作或先让Excel另存为值列顺序合并后变了不同文件列名的顺序不一致在合并前统一reindex(columnsstandard_columns)数字带了元字无法求和字符串混入了单位用str.replace(元, ).astype(float)8. 我的三点实操体会批量处理Excel和CSV这件事代码从来不复杂真正考验人的是细心程度和对文件本身的理解。第一次跑通合并脚本时我以为万事大吉结果检查数据发现有两千多行是全空的——原因很简单有个文件里插入了很多整行空行pandas读进来之后这些空行被保留了。从那次之后我所有的合并代码第一行必然是dropna(howall)先把全空行清掉再说。第二个体会是永远保留数据来源列。之前帮我一个同事做合并他死活找不到某条问题数据是从哪个文件来的。后来我养成了合并时自动加一列来源文件的习惯任何一条数据都能反查源头。这个习惯后来救了我不止一次。第三个体会是不要指望一条正则表达式解决所有日期格式。不同后台导出的日期风格差异太大最稳妥的方案是先pd.to_datetime(errorscoerce)转标准格式再把转失败的行打印出来人工处理。批量处理的核心价值是让机器处理90%的重复劳动剩下10%的异常情况老老实实手动校对也是成本的一部分。这是批量处理的常态别想着一个脚本解决所有问题也不要让一个脚本把问题藏起来。数据处理的本质是让异常浮出水面而不是让它们消失在沉默的合并结果里。