ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

Python批量处理Excel和CSV:表头归一化与自动化脚本实战

Python批量处理Excel和CSV:表头归一化与自动化脚本实战 做数据的人多少都有过这种经历月底要对账十几个Excel表格要从不同同事手里收过来表头不统一、格式五花八门你得像裁缝一样把每一块布对好纹路再缝在一起或者需要把几百个CSV文件转成Excel发给不懂技术的领导手动一个个打开另存为一下午就这么耗没了。直到我认真用Python把这一套流程脚本化之后才意识到过去浪费了多少时间。这篇内容我打算把用Python批量处理Excel和CSV文件的完整思路、工具选型、实操代码和排坑经验一次说透适合经常跟表格打交道的运营、财务、数据分析岗位的朋友也适合刚学Python、想拿真实场景练手的初学者。1. 环境准备与工具选型别一上来就装一堆库处理Excel和CSVPython里能用的库不少但真正干批量处理的活主力其实就三个pandas、openpyxl、csv。很多人一上来就装一整套Anaconda或者贪多把xlrd、xlwt、xlutils、pyexcel全装了真到用的时候反而不知道该用哪个。先搞清楚每个库的定位后面写代码才会有底气。1.1 库的分工逻辑谁负责读、谁负责写、谁只适合做数据运算先看三者的关系我用最直白的方式说清楚pandas负责“数据层”的活能做合并、筛选、分组、透视、清洗之类的高级操作。它的read_excel和read_csv几乎是批量处理的核心入口只要把数据读进DataFrame剩下的怎么折腾都行。openpyxl负责“格式层”的活能读写.xlsx文件的单元格样式、合并单元格、列宽行高、公式等。pandas内部写Excel的时候底层就是调它。csv模块Python自带的标准库不用安装。适合对付超大CSV文件——几个GB那种pandas读会直接内存爆掉csv模块配合迭代器可以一行一行流式读。我的建议是默认组合为pandas openpyxlCSV纯读取或超大文件另用csv模块处理。不要安装xlrd去读.xlsx它的设计只完整支持.xls老格式遇到新版xlsx会报错白白踩坑。1.2 安装与版本注意点用对镜像源能省一小时安装很简单一条命令搞定pip install pandas openpyxl如果你在国内网络环境下建议加上清华镜像否则下载pandas这种大体积包时很容易超时pip install pandas openpyxl -i https://pypi.tuna.tsinghua.edu.cn/simple装完之后验证一下版本注意pandas和openpyxl之间是有兼容关系的太老的pandas配太新的openpyxl偶尔会出一些莫名其妙的问题python -c import pandas; import openpyxl; print(pandas.__version__, openpyxl.__version__)实测比较稳妥的组合是pandas 2.0以上配openpyxl 3.1以上。我遇到过pandas 1.3搭配openpyxl 3.0时to_excel写入多sheet时偶发性地丢数据升级版本后问题消失。还要提醒一句如果电脑里同时装了Python 2和Python 3务必用python3 -m pip install这种方式把库装到正确的解释器里。排查“装了半天import还是报ModuleNotFoundError”的问题八成都是装串了环境。2. 批量处理的核心技术准备遍历、编码与表头归一化工具装好了先别急着写合并代码。批量处理真正的难点在于“批量”两个字带来的不确定因素几百个文件放在不同子目录里怎么找到CSV文件有的用UTF-8、有的用GBK怎么保证读出来不是乱码每个文件的列名大小写不同、空格不同怎么统一2.1 文件遍历方案glob和pathlib怎么选批量处理第一步就是把所有目标文件找出来。传统写法是os.listdir()加endswith()判断但这种写法有两个痛点第一是不方便遍历子目录第二是正则筛选能力弱。我推荐用pathlib模块Python 3.4之后就内置了代码可读性比os好太多。我用一个实际场景演示某个目录下有一堆Excel文件只要后半个月的名字形如data_20240715.xlsx到data_20240731.xlsx其中还混着几个.csv文件和一个.bak备份文件。from pathlib import Path data_dir Path(./data) excel_files list(data_dir.glob(data_202407*.xlsx)) csv_files list(data_dir.glob(data_202407*.csv))如果文件在多级子目录里用rgloball_excel_files list(data_dir.rglob(*.xlsx))glob和rglob返回的都是Path对象可以直接拿去做后续操作比os.path.join拼字符串省事得多。建议所有涉及文件路径的逻辑都用Path对象它天然帮你在Linux、macOS、Windows下处理分隔符差异把代码写到任何一台机器上都能跑。2.2 编码问题CSV乱码的根源与解决办法CSV文件的编码问题堪称批量处理的头号杀手。早期的Windows办公环境Excel默认用GBK简体中文版或GB18030保存CSV而Python默认认为字符串是UTF-8直接用pd.read_csv()读GBK文件一读就报UnicodeDecodeError读出来也是满屏乱码。我的处理方式是分两步走第一步读文件时带上encoding参数。但难点在于你事先不一定知道每个文件是啥编码所以在代码里做一个自动探测更保险from charset_normalizer import from_path def detect_encoding(file_path): result from_path(file_path).best() return result.encoding file_path ./data/raw_20240701.csv enc detect_encoding(file_path) df pd.read_csv(file_path, encodingenc)第二步写入CSV时如果想给Excel用户后续使用建议统一写成UTF-8-SIG编码而不是UTF-8。原因在于Excel打开无BOM头的UTF-8文件时会自动按ANSI解析中文直接乱码带BOM头UTF-8-SIG的则能正确识别。这个坑我踩过太多次现在所有导出给业务侧的CSV一律写encodingutf-8-sigdf.to_csv(./output/result.csv, indexFalse, encodingutf-8-sig)2.3 表头归一化合并多文件前必须做的“统一度量衡”批量合并不同人传来的表格最让人头疼的就是表头不统一。有人写“成交金额”有人写“金额”有人写“sales”还有大小写和空格的区别。直接pd.concat结果就是多出好几列重复数据。我的方案是做一个映射字典把别名全部归一化。实际操作中可以先扫描一遍所有文件的列名用代码把差异暴露出来for f in excel_files: df pd.read_excel(f) print(f.name, list(df.columns))然后把列名的清理规则写进函数统一去除首尾空格、统一转小写、把常见别名映射为标准名。COLUMN_ALIASES { 成交金额: amount, 金额: amount, sales: amount, sales_amount: amount, 日期: date, date_str: date, } def normalize_columns(df): df.columns df.columns.str.strip().str.lower() df.rename(columnsCOLUMN_ALIASES, inplaceTrue) return df表头归一化完成之后再做concat数据才会真正落到同一列里。这个步骤看起来不起眼却是整个批量合并流程里最容易出错、也最值得花时间打磨的一环。3. 实操案例三个可以直接抄作业的批量处理脚本理论铺垫完了进入真正的干货环节。我准备了三个实际工作中使用频率最高的批量处理场景批量合并多文件、批量清洗数据、批量拆分与格式转换。每个场景都给出完整代码你不需要理解每一行直接改路径就能跑跑通之后再去琢磨背后的细节。3.1 场景一批量合并多个Excel文件中的所有Sheet业务上最常见的需求一个目录下有12个月的销售报表每个文件里有12个Sheet需要全部合并成一个总表。这个需求如果用excel手工操作要分别打开144个Sheet复制粘贴能做到怀疑人生。用pandas合并代码不超过20行。import pandas as pd from pathlib import Path data_dir Path(./月度数据) output_path ./合并结果.xlsx all_data [] for file_path in data_dir.glob(*.xlsx): # 读入文件的所有Sheet sheets pd.read_excel(file_path, sheet_nameNone) for sheet_name, df in sheets.items(): if df.empty: continue # 打上来源标记方便日后追溯 df[来源文件] file_path.stem df[来源Sheet] sheet_name all_data.append(df) if all_data: merged_df pd.concat(all_data, ignore_indexTrue) merged_df.to_excel(output_path, indexFalse) print(f合并完成共 {len(merged_df)} 行{len(all_data)} 个Sheet) else: print(没有找到有效数据)这里有几个细节值得注意sheet_nameNone会返回一个字典键是Sheet名值是DataFrame这是读取全部Sheet最优雅的写法。ignore_indexTrue保证拼接后行号自动生成全新序列而不是每一块都从0开始否则后面做筛选排序时会出大问题。我在每个块上都加了“来源文件”和“来源Sheet”两列方便日后审计数据时追溯来源。别嫌这多余真到数据对不上账的时候这两列能帮你省掉三天扯皮时间。3.2 场景二批量清洗CSV文件并输出标准化结果再举一个更接近真实的数据清洗场景你收到一批从后台导出的CSV文件里面存在重复行、日期格式混乱、金额字段带货币符号、缺失值等各类问题。手工清洗一个文件要几分钟500个文件就是一整天的工时。脚本化之后让机器去处理重复劳动人来盯异常规则这才是批量处理的核心意义。import pandas as pd from pathlib import Path input_dir Path(./raw_csv) output_dir Path(./cleaned_csv) output_dir.mkdir(exist_okTrue) def clean_dataframe(df): # 1. 去除完全重复的行 df df.drop_duplicates() # 2. 删除所有列都为空的无效行 df df.dropna(howall) # 3. 统一日期格式把常见的 2024/7/1、2024-07-01、20240701 统一成标准格式 df[日期] pd.to_datetime(df[日期], errorscoerce, formatmixed) df df.dropna(subset[日期]) # 4. 清洗金额字段去掉货币符号和千分位逗号 df[金额] df[金额].astype(str).str.replace(r[$,], , regexTrue) df[金额] pd.to_numeric(df[金额], errorscoerce) df df.dropna(subset[金额]) return df for file_path in input_dir.glob(*.csv): try: df pd.read_csv(file_path, encodingutf-8) except UnicodeDecodeError: df pd.read_csv(file_path, encodinggbk) cleaned clean_dataframe(df) output_file output_dir / fcleaned_{file_path.name} cleaned.to_csv(output_file, indexFalse, encodingutf-8-sig) print(f{file_path.name}: {len(df)} 行 - {len(cleaned)} 行)关键点在于pd.to_datetime要加errorscoerce把无法解析的日期变成缺失值再删除金额清洗用regexTrue的正则替换比链式.replace().replace()要快得多也更不容易漏掉格式差异。关于性能这里分享一个实测经验pandas逐行操作非常慢上面的清洗逻辑全部使用向量化函数不写for循环去遍历每一行数据。处理几十万行的CSV整个脚本跑完可能只需要几秒钟如果你用了逐行apply可能就要几分钟。批量脚本的性能瓶颈多半不在文件数量而在你有没有用对批处理函数。3.3 场景三按类别拆分一个大数据文件为多个小文件合并之外拆分也是高频需求。比如你要把一个包含华东、华北、华南、西南四个大区数据的CSV拆成4个独立文件方便分发给对应区域的负责人。这类用pandas做就是三行代码的事import pandas as pd df pd.read_csv(./全国销售数据.csv, encodingutf-8-sig) groups df.groupby(大区) for name, group in groups: safe_name str(name).replace(/, _) # 防止非法文件名字符 group.to_excel(f./分区域/{safe_name}.xlsx, indexFalse) print(f{name}: {len(group)} 行)这里要特别提醒写法上的坑groupby(列名)返回的迭代对象里name是分组键的取值group是对应的子集DataFrame。很多人会把代码写成for name, group in df.groupby(...)然后忘记的组别排序问题这个没问题真正容易翻车的是文件名里不能包含/、\、:这些非法字符。我的建议是所有要作为文件名的内容都做一次安全替换。另外如果拆分后的目标是Excel文件每个文件十几万行那种写入会比较慢这是openpyxl写xlsx的正常性能不必焦虑。想更快的话可以继续输出CSV但考虑到Excel的普遍使用率这个速度也能接受。3.4 批量格式转换Excel与CSV互转的正确姿势最后补充一个几乎人人都用过的场景把CSV转成Excel方便给非技术同事查看或者把几十个Excel转成CSV喂给下游程序。我用一句话说清楚转换的要点别用什么excel另存为的宏直接上pandas且务必管好缺失值和数据类型。import pandas as pd from pathlib import Path def csv_to_excel(csv_path, excel_path): df pd.read_csv(csv_path, encodingutf-8-sig) df.to_excel(excel_path, indexFalse) def excel_to_csv(excel_path, csv_path): df pd.read_excel(excel_path) df.to_csv(csv_path, indexFalse, encodingutf-8-sig) # 批量互转 for csv_file in Path(./csv文件).glob(*.csv): csv_to_excel(csv_file, Path(./excel文件) / f{csv_file.stem}.xlsx) for excel_file in Path(./excel文件).glob(*.xlsx): excel_to_csv(excel_file, Path(./csv输出) / f{excel_file.stem}.csv)注意Excel转CSV时DataFrame里的日期会默认带上时间戳比如2024-07-01 00:00:00如果不想输出这个时分秒可以先做一次格式化df[日期] df[日期].astype(str).str.slice(0, 10)。这个小细节能让你的CSV干净很多尤其是对后续做SQL导入或其他系统对接的时候少踩不少坑。4. 常见问题与排查技巧实录写批量处理脚本跑一遍就成功的概率很低。我做过的数据处理脚本几乎每个都经历过一次或多次报错。下面把最经典、最高频的几个问题整理成速查表方便你遇到问题时按图索骥。4.1 速查表六大高频问题对照报错或异常现象根因解决方案ModuleNotFoundError: No module named pandas库装错环境或用错解释器用python -m pip list确认库在哪个环境切换到正确解释器后重装TypeError: expected str, bytes or os.PathLike传了None值或错误类型给路径参数检查文件路径变量是否被赋了空值用Path(str(path))包裹一次UnicodeDecodeError: utf-8 codec cant decode byteCSV编码不是UTF-8用charset_normalizer探测编码或直接改用encodinggbkKeyError: 列名表头归一化未处理列名不一致打印df.columns确认真实列名先跑归一化函数再操作ValueError: Excel file format cannot be determined文件扩展名是xlsx但实际是html、xls或损坏文件用file命令或读取前8字节判断真实文件类型对损坏文件单独隔离内存溢出MemoryError一次性读取超大CSV用pd.read_csv(..., chunksize10000)分块处理或改csv模块流式读取4.2 逐条拆解这些坑具体是怎么踩的又怎么填平第一类高频坑是引擎选择错误。用pd.read_excel读取.xlsx文件时pandas底层引擎默认在openpyxl和xlrd之间切换。如果只装了xlrd老版本读xlsx会报错让你安装openpyxl反过来如果目标是.xls老格式openpyxl是读不了的得用xlrd2.0配合enginexlrd。我的建议是文件后缀和实际格式必须逐一核验尤其是从业务系统导出的文件后缀经常不靠谱。检查真实格式用一段小代码from pathlib import Path def check_file_type(path): with open(path, rb) as f: head f.read(8) if head[:4] bPK\x03\x04: return zip-based (xlsx/docx) if head[:4] b\xd0\xcf\x11\xe0: return OLE2 (xls/doc) if head[:2] bID3: return mp3 return unknown第二类高频坑是to_excel之后数值变成文本。常见原因源数据里金额、序号这类字段是object类型里面混入了逗号或空格pandas写入Excel时按字符串写了。解决思路是写之前统一做一次pd.to_numeric(df[列名], errorscoerce)。另外如果你在Excel里看到1.00之类的多余小数原因也很简单pandas默认把整数读成了float64写入后自然带小数尾缀。这不是bug是数据类型选择问题先用dtypestr或转换整数类型或写Excel后手工设置单元格格式都能解决。第三类高频坑是批量处理过程中个别文件格式异常导致整个脚本中断。这种问题在几百个文件批量跑的时候实在太常见有一个文件其实是快速另存为的假xlsx就会拖垮整个流程。正确处理方式是写一个容错循环而不是让异常直接中断failed [] for file_path in data_dir.glob(*.xlsx): try: df pd.read_excel(file_path) # 正常业务处理... except Exception as e: failed.append((file_path.name, repr(e))) print(f[失败] {file_path.name}: {e}) print(f成功 {len(list(data_dir.glob(*.xlsx))) - len(failed)} 个失败 {len(failed)} 个) for name, err in failed: print(name, err)这样脚本会把失败文件记录下来跑完后单独处理而不是让一台机器在第一次报错时就罢工人还得守在屏幕前盯报错。一些实践后的额外建议把批量处理脚本写到第四五个版本之后我最大的体会就是批量处理的核心不是代码本身而是对数据质量的管理意识。脚本做得再健壮数据源表头不统一、异常值不标注、编码不明确终究会出问题。所以我在所有进入生产环境的批量脚本里都会加上一份简单的处理报告输入多少个文件、成功多少个、失败多少个、清洗前后行数变化、有哪些可疑数据被剔除。输出报告的方式很灵活写进文本日志就行再做定时任务跑批时这份日志就是事后审计的唯一依据。还有一个小技巧强烈建议跑任何批处理之前先找两个小文件做一轮试运行确认脚本逻辑没问题再跑全量。别拿几百个文件直接上生产万一逻辑有误数据一旦被覆盖写坏后悔都来不及。我就干过一次直接把原文件覆盖的事后来学乖了所有输出都写进独立的新目录原文件一律不动宁可让磁盘多占一点空间。这个习惯在数据工作中比任何代码技巧都值得养成。
返回列表