
在日常工作里真正折磨人的往往不是那些高大上的算法而是每天都在跟你打交道的文件处理。二十几个CSV要合并、几千个TXT里要筛出重复内容、日志文件乱成一团需要按时间整理、从爬虫程序里落地的数据一堆乱码……这些活如果用鼠标点点点一晚上就交代了但如果用Python写个几十行的脚本可能连一杯咖啡都来不及凉就能跑完。Python文件处理这个方向学的就是这些“一劳永逸”的本事用代码代替手工把重复性的文件操作变成可复用的工具。这篇文章我会把文件处理从思路到实操完整讲一遍不绕弯子。文章里会重点拆解Python处理文件的几个核心知识点pathlib路径操作、open()读写模式、编码处理、大文件读取以及三个可以直接“抄作业”的实战脚本批量去重、日志合并、CSV清洗。新手可以照着代码一步步跑通全流程有经验的读者也能从常见问题排查和避坑心得里找到一些平时不太注意的细节。文章内容偏实用派聊的都是我自己在项目中踩过的坑、验证过的方式。1. 内容整体设计与思路拆解1.1 为什么文件处理要选Python而不是Shell或手工很多朋友一上来就问我文件处理不是用Shell脚本也能做吗比如Linux下用sed、awk、grepWindows下用批处理为什么非得用Python我的回答是看场景。如果你只需要简单地从一堆文件里grep一个关键词那Shell脚本确实更快。但只要需求复杂一丁点——比如要去重、要根据编码转换内容、要生成结构化数据、要跨平台运行、要把结果按字段排序——Shell脚本的维护成本立刻就上来了。Python的优势在于它把“文件操作”这个事抽象得很完整路径是一个对象Path文件是一个可迭代对象内容可以逐行处理编码可以显式控制数据结构可以用list、dict、Counter这些现成的容器随意组合。这意味着你能在同一个脚本里完成“读取-解析-计算-输出”一整条链路的逻辑而不是在Shell、sed、awk之间来回切换。Python的生态也帮了大忙。处理CSV有csv模块和pandas处理JSON有json模块处理文本模糊匹配有re模块随手就能把文件内容和数据分析、爬虫、自动化串起来。我的日常经验是凡是处理周期超过两次、处理文件超过十个的重复性任务就直接写Python脚本。哪怕脚本写得丑一点后续复用的时候就知道值了。还有一个很容易被忽略的点是跨平台。我手里的脚本经常在Windows开发机上写完再丢到Linux服务器上跑。如果用shell脚本写换平台就要改一堆语法但Python几乎不用改只要别用硬编码的路径分隔符就行。这种“一处编写处处运行”的体验在文件处理场景里尤其明显。1.2 文件处理的通用流程其实就五步无论你面对的是什么形式的文件处理需求拆解下来都逃不过这五个环节定位找到要处理的文件。路径是写在配置里、从命令行传入、还是按规则扫描目录读取用合适的模式打开文件。文本还是二进制UTF-8还是GBK全量读还是逐行读解析把文件内容变成程序能理解的数据。是逐行匹配正则还是用CSV/JSON模块解析处理对内存中的数据做筛选、合并、去重、排序、计算。写出把结果输出到目标文件。是覆盖还是追加要不要写临时文件要不要统一编码这五步看起来简单但每步都有坑。定位环节容易踩“相对路径与当前目录不一致”的坑读取环节最容易踩换行符和编码的坑解析环节经常栽在“以为文件格式规整但其实不规整”上写出环节很多人习惯直接用w模式覆盖原文件结果脚本跑到一半报错原文件就毁了。我在设计脚本的时候会先把这五步在脑子里过一遍想清楚每一步的边界条件然后再动手写代码。比起上来就敲两三行open()就开始处理提前把流程梳理清楚能省掉后面大量调试时间。尤其是当你面对的是几百个文件而不是单个文件时“全量跑完再回头查错”的成本是非常高的一开始就设计到位才不用返工。2. 核心细节解析与实操要点2.1 路径处理别再手拼字符串了pathlib才是正道我在早期写文件处理脚本时用的是os.path.join拼接路径写着写着就发现两个痛点一是Windows路径分隔符是反斜杠Linux是正斜杠硬编码字符串很容易出问题二是路径本质上是字符串但你要判断它是否存在、是不是文件、文件名叫什么都得调用一堆os.path函数代码看着又碎又长。后来我全面切换到pathlib体验提升非常明显。pathlib把路径封装成了Path对象你可以直接用“/”运算符来拼接路径代码读起来就像在描述路径本身。更重要的是Path对象自带exists()、is_file()、suffix、name、stem这些属性和方法文件处理里常见的需求基本上都有现成接口。from pathlib import Path # 拼接路径 base_dir Path(data) log_file base_dir / logs / app.log print(log_file) # data/logs/app.log # 判断路径状态 print(log_file.exists()) # 文件是否存在 print(log_file.is_file()) # 是不是文件 print(log_file.suffix) # 扩展名 .log print(log_file.name) # 文件名 app.log print(log_file.stem) # 不带扩展名的文件名 app # 获取绝对路径、父目录 print(log_file.resolve()) # /xxx/data/logs/app.log print(log_file.parent) # data/logs为什么我强调用pathlib而不是字符串拼路径因为文件处理脚本一旦跑在真实环境里路径问题是最容易暴雷的。比如你从配置文件里读到个根目录后面要拼几十个子路径如果用字符串拼接就必须在每次拼接时操心结尾的斜杠用Path对象就完全没有这种烦恼。它还能直接跟glob、rglob配合做批量匹配配合os.walk的场景也能用一个简单的Path.glob替代代码会更清晰。提示如果你的代码还要兼容Python 3.5甚至更早的版本可能需要用os.path但Python 3.6以上都建议直接用pathlib这是标准库不用额外安装。2.2 文件打开、读写模式与上下文管理器文件处理绕不开open()函数。很多教程会把这个知识点讲得很理论我这里就说几个我实测下来最重要的理解。第一个是文件模式。open()的第二个参数mode决定了你是读还是写、是文本还是二进制、是覆盖还是追加。我用得最多的几个模式如下表格模式说明常用场景r只读文件必须存在读取现有文件w写入覆盖已有内容文件不存在则创建输出处理结果a追加写入文件不存在则创建日志追加rb二进制只读图片、压缩包等非文本处理wb二进制写入生成二进制文件r读写文件必须在可读可写修改部分内容a追加读取边追加边读很多新手会犯的错是把w和r的模式搞混导致文件被意外清空。我的习惯是只读文件一定显式写r绝不省略输出文件尽量不用w这种读写混合模式除非有明确的需要。第二个是上下文管理器。with open(...) as f: 这个写法Python会在代码块结束后自动关闭文件句柄。很多人写文件处理代码时容易忽略关闭这一步在大量循环里反复open但不close最后报“Too many open files”。用with就可以根治这个问题。它还会在处理抛出异常时自动关闭文件这对稳定性非常重要。第三个是大文件的读取方式。如果你用read()把一个好几个GB的文件一次性读进内存很可能会直接MemoryError机器卡死只能重启。正确做法是逐行处理让文件对象成为一个迭代器每次只会有一行内容占用内存with open(big_file.log, r, encodingutf-8) as f: for line in f: # 每一行就在这个循环里处理 print(line.strip())这个写法在内存占用上是常数级别的不管文件多大都不会撑爆内存。我处理过几个GB的Nginx日志就是靠这种逐行读的方式顺利跑完的。2.3 编码问题必须一开始就想清楚文件处理里最让人头疼的问题十有七八跟编码有关。Python 3默认用UTF-8但我们在Windows环境下拿到的文件很多是GBK、GB2312或其它编码下载的网页可能是UTF-8带BOM从旧系统导出的数据可能是Latin-1。如果你直接按默认编码去读这些文件第一行代码可能就抛UnicodeDecodeError。我的处理原则是在读取之前先搞清楚源文件的编码不要赌。有几个办法可以做如果你知道文件来自Windows的记事本导出多半是GBK或带BOM的UTF-8如果文件来自爬虫、API导出通常是UTF-8如果实在不知道可以用chardet库去检测或者自己用一小段样本内容做启发式判断。在实际打开文件时我通常会显式传encoding参数而不是依赖默认值。比如# 读取GBK编码的旧数据 with open(legacy.txt, r, encodinggbk, errorsignore) as f: content f.read() # 输出时统一转成UTF-8带BOM防止Excel乱码 with open(result.csv, w, encodingutf-8-sig) as f: f.write(content)特别注意utf-8-sig这个编码。如果你的CSV文件生成出来要给业务方用Excel打开用普通的utf-8Excel默认会按本地编码中文系统常常是GBK解析结果表头全是乱码。用utf-8-sig写入会在文件开头带一个BOM标记Excel就能正确识别。这个小细节我在交付数据文件的时候可没少用到。另外errors参数也很有用。读旧系统文件时偶尔会遇到个别非法字符如果你直接报错整个任务就中断了。设置errorsignore可以跳过非法字符继续处理设置errorsreplace则会把识别不了的字符替换成。具体用哪个看你的需求——是保全字段还是限损处理。3. 实操过程与核心环节实现3.1 实战一批量找出重复文件按内容去重这个脚本是我帮朋友清理磁盘时写的。场景是这样的一个文件夹下面有好几层子目录里面散落了几百个TXT、Word导出文档和SQL备份因为多次copy、备份很多文件内容是重复的但文件名不一样想通过文件名去重根本做不到必须按内容来判断。思路其实很直接对每个文件计算一个哈希值我用MD5把哈希值当作“内容指纹”只要是内容一样的文件MD5一定相同。在遍历过程中遇到已经见过的指纹就认为是重复文件输出它的路径方便人工确认或直接删除。from pathlib import Path import hashlib def file_md5(file_path, chunk_size8192): 计算文件MD5值用分块方式避免大文件占满内存 h hashlib.md5() with open(file_path, rb) as f: while chunk : f.read(chunk_size): h.update(chunk) return h.hexdigest() def find_duplicates(target_dir): target Path(target_dir) seen {} # 指纹 - 第一个出现的文件路径 duplicates [] # 重复文件列表 for file_path in target.rglob(*): # rglob会递归所有子目录 if not file_path.is_file(): continue # 跳过目录 fp file_md5(file_path) if fp in seen: duplicates.append((file_path, seen[fp])) else: seen[fp] file_path return duplicates if __name__ __main__: dup_list find_duplicates(./data) for dup, original in dup_list: print(f重复文件: {dup} - 原始文件: {original}) print(f共发现 {len(dup_list)} 个重复文件)这里有两个细节经验想分享。第一我用了rglob(*)而不是手工写os.walk它会把当前目录及所有子目录的文件都递归取出来配合Path.is_file()过滤目录代码非常简洁。第二计算MD5用了分块读取每次8KB而不是一次性读整个文件。处理小文件时两者差别不大但一旦文件平均是几百MB甚至GB级别分块的优势立刻显现内存占用始终很低。脚本跑完后会输出一份重复文件清单。友情提醒自动删除文件这种操作最好只生成清单让用户确认后再删千万别在脚本里直接删。我就是吃了这个亏有一次写脚本太“自动化”去重时误删了一个文件后来发现内容略有不同后悔都来不及。凡是涉及删文件、覆盖文件的操作能交互就交互能备份就备份。3.2 实战二合并多个日志文件并排序、去重日志文件合并是运维和开发场景里特别常见的需求。比如你有多台服务器每天的访问日志、错误日志各自落在不同的文件里全量排查问题时希望把这些日志合并成一个文件按时间先后排列同时把完全相同的重复行比如重复打印的同一报错去掉。下面这个脚本是我的简化版本from pathlib import Path import glob def extract_time(line): 从日志行里提取时间字符串用于排序这里假设格式为 [2024-01-15 10:22:33] try: start line.index([) 1 end line.index(]) return line[start:end] except ValueError: return 0000-00-00 00:00:00 # 无法解析就排最前面 def merge_logs(source_pattern, output_file): merged_lines set() # 用set自动去重 for file_name in glob.glob(source_pattern): path Path(file_name) with open(path, r, encodingutf-8, errorsignore) as f: for line in f: merged_lines.add(line.strip()) # 按时间排序 sorted_lines sorted(merged_lines, keyextract_time) with open(output_file, w, encodingutf-8) as f: for line in sorted_lines: f.write(line \n) if __name__ __main__: merge_logs(./logs/*.log, ./logs/merged.log)这个脚本几行代码就完成了三个核心功能合并、去重、排序。先看合并。glob.glob(./logs/*.log)会匹配所有.log文件逐个打开、逐行读取。这里用set来存储所有行因为set天然去重——相同的行只保留一次。从集合的定义来说它要求元素是hashable的字符串正好满足。排序则依赖extract_time函数。这里的日志行格式是“[2024-01-15 10:22:33] ...”这种我用字符串的index方法把时间部分截取出来直接作为排序key。之所以不解析成datetime对象是因为排序只需要字符串比较就能工作——只要时间格式统一字符串的字典序和时间的先后序是一致的。省去了解析带来的性能损耗。如果日志格式再复杂一点比如时间字段不在行首、带毫秒或时区那再把这段解析逻辑换成datetime.strptime也不迟。合并完成后我还习惯加一个简单的统计处理后总行数、去重掉了多少条。这个统计能快速帮你判断结果是否合理比如某次统计发现去重掉了70%的行你就该想想是不是日志本身重复量就大或者是不是set的逻辑把不该去掉的内容也去掉了。3.3 实战三CSV结构化数据清洗CSV是数据导出、报表处理中最常见的文件格式但它也是最不讲武德的格式之一。字段之间用逗号分隔看似简单实际处理时可能出现某些字段本身带逗号却被引号包起来了、有空行、Excel导出时不时带BOM、字段顺序在不同批次导出中不一致、编码不统一……所以CSV清洗核心思路不是“读进来就行”而是解析成结构化数据再按规则清洗。Python标准库的csv模块是处理CSV的可靠选择。很多人图省事直接逐行split(,)这在简单场景下能用但遇到带引号字段就彻底崩了。csv.reader会正确解析带引号、转义符等特殊情况是更规范的做法。import csv from pathlib import Path def clean_csv(input_path, output_path, required_fieldsNone): required_fields: 必填字段列表如果某一行的这些字段为空就丢弃 fieldnames None cleaned_rows [] with open(input_path, r, encodingutf-8-sig, newline) as f: reader csv.DictReader(f) # 读表头确认字段结构 fieldnames reader.fieldnames for row in reader: # 跳过全空行 if not any(row.values()): continue # 跳过必填字段为空的脏行 if required_fields: if any(not row.get(field) for field in required_fields): continue # 这里还能加规则比如字段长度、数值范围、日期格式 cleaned_rows.append(row) with open(output_path, w, encodingutf-8-sig, newline) as f: writer csv.DictWriter(f, fieldnamesfieldnames) writer.writeheader() writer.writerows(cleaned_rows) if __name__ __main__: clean_csv(./raw/export_20240115.csv, ./clean/export_cleaned.csv, required_fields[订单号, 金额])这里面有几个要点值得展开。第一个是newline。如果不加这个参数在Windows环境下writer写出时会自动把行尾转成\r\n而csv模块本身又负责管理换行两者叠加可能产生空行。加newline之后让csv模块自己控制换行这是官方文档推荐的做法实测能避免导出文件每行之间多一个空行的尴尬。第二个是DictReader/DictWriter。相比reader以列表返回每一行DictReader直接用表头做key你可以像操作字典一样操作每行数据代码的可读性高很多。比如要判断“金额”字段是否为空直接写row.get(金额)而不是去猜它在第几列。这在字段顺序可能变化时非常有用——只要列名一致代码不用改。第三个是utf-8-sig。我在上一节讲过这里再强调一次生成给Excel打开的CSV带BOM的UTF-8是最稳的。很多业务同事拿到UTF-8无BOM的CSV双击打开中文全变乱码第一反应就是“你的程序有问题”。用utf-8-sig写就不会有这种误会。如果数据量非常大比如百万行级的CSV建议直接用pandas的read_csv和to_csv处理速度和内存表现更好csv模块是更轻量、依赖更少的选择。但是csv模块完全不依赖第三方库环境受限时它才是保底方案。4. 常见问题与排查技巧实录4.1 高频报错速查表文件处理脚本里几个报错出现的频率高到你闭着眼都能背出来。我整理了一个速查表方便你遇到问题第一时间定位报错信息原因解决办法FileNotFoundError路径拼错、文件不存在、相对路径参照的当前目录不是你想象的打印Path.cwd()确认当前目录用绝对路径或基于__file__计算路径PermissionError文件被其它程序占用、没有写入权限关闭Excel/编辑器再运行检查目录权限确认不是只读文件UnicodeDecodeError用错误编码读取文件或者文件里有非法字符显式指定正确编码设置errorsignore用chardet检测MemoryError一次性读取了超大文件到内存改成逐行读取用分块处理加大机器内存或改用pandas分块IsADirectoryError用open()打开了目录而不是文件读取前加Path.is_file()判断检查glob匹配到的具体对象输出文件每行多空行写入文件时没设newlineopen()加newline参数中文写入后变成乱码输出编码与读取工具不一致统一用UTF-8给Excel生成的CSV用utf-8-sig还有一类不那么显眼但非常常见的坑就是文件被另一个程序占用。Windows下特别明显你在Excel里打开了一个CSV然后脚本想往里面写会直接PermissionError。遇到这种情况先关掉编辑器再跑脚本不要怀疑自己的代码写错。4.2 相对路径绝对路径之争以及__file__的正确使用很多兄弟写脚本时习惯用相对路径比如open(data.csv)。这个写法在小脚本里没问题但一旦脚本被放在别的目录下、或者用cron定时任务、或者在当前工作目录不同的情况下运行它就可能会报找不到文件。关键要理解相对路径是相对于“当前工作目录”即命令行的执行目录的而不是脚本所在的目录。我遇到过太多次脚本放在A目录命令在B目录执行脚本里的相对路径就是从B目录出发找文件结果当然找不到。解决办法有两个。第一在脚本开头统一指定基准目录比如from pathlib import Path BASE_DIR Path(__file__).resolve().parent DATA_DIR BASE_DIR / data这里__file__是脚本自身的路径.resolve()会转换成绝对路径.parent拿到脚本所在目录。之后所有路径都从DATA_DIR去拼接这样无论你在哪个目录下执行脚本文件都能被正确地找到。第二如果脚本接收外部传入的路径参数就用绝对路径或者进入脚本后立刻做Path(input_path).resolve()避免在后续处理中因为当前目录变化而踩坑。我自己的习惯是文件处理脚本的路径从不在代码里硬编码全部做成参数或配置代码可复用性一下就上来了。4.3 写文件的原子性以及“备份优先”原则文件处理脚本里最危险的操作大概就是“覆盖写文件”。很多人拿到一批文件直接open(path, w)把原文件内容替换掉结果程序刚处理到一半因为某个数据异常抛了异常原文件已经被清空或写了一半——这个过程是不可逆的。我的做法是输出到一个临时文件等所有处理成功结束后再把临时文件改名为目标文件。这个操作在操作系统层面是接近原子的要么旧文件还在要么新文件完全替换旧文件不会出现写到一半的坏文件。import os from pathlib import Path target_path Path(output.csv) # 最终目标文件 tmp_path target_path.with_suffix(.tmp) # 临时文件 try: with open(tmp_path, w, encodingutf-8) as f: # ... 完成所有写入 pass # 全部安全写出后再替代原文件 os.replace(tmp_path, target_path) except Exception: if tmp_path.exists(): tmp_path.unlink() # 出错时清理临时文件 raise另外批量处理时备份优先是铁律。我在执行批量重命名、批量修改文件内容之前会先把原目录整个复制一份加个时间戳前缀或者直接把改动前的文件复制成.bak后缀。代价只是多占一点磁盘空间但万一处理逻辑有瑕疵你可以轻松回滚。我在生产环境跑批处理脚本备份这一步从来不会省。5. 文件处理还能怎么扩展5.1 从单文件到目录树用glob与Path.rglob脚本写得多了你会发现大量需求其实是“处理一个目录下的所有文件”而不是单个文件。这时用glob.glob或Path.glob可以做得非常优雅。from pathlib import Path # 找出当前目录所有txt文件 for file_path in Path(.).glob(*.txt): pass # 递归找出所有子目录下的csv文件 for file_path in Path(.).rglob(*.csv): pass # 按多后缀匹配 for file_path in Path(.).rglob(*.log): passPath.rglob()能列出目录所有内容配合条件判断筛选Path.rglob(.csv)能直接按后缀过滤。如果你需要非常复杂的匹配逻辑比如文件名包含特定关键词、排除某些目录、按修改日期范围过滤可以先用rglob拿到所有文件再用列表推导式做条件过滤。要注意rglob默认也会进入一些隐藏目录必要时可以用一个函数跳过不需要遍历的目录。5.2 用正则和argparse把脚本做成命令行工具文件处理脚本一旦要经常使用就应该把它做成一个正经的命令行工具输入路径、输出路径、过滤条件都作为参数传进来而不是每用一次就改代码。标准库argparse就可以完成这件事。一个典型的命令行接口长这样import argparse from pathlib import Path parser argparse.ArgumentParser(description批量清理文本文件脚本) parser.add_argument(input_dir, help输入目录) parser.add_argument(output_dir, help输出目录) parser.add_argument(--keyword, help只保留包含该关键词的行, defaultNone) parser.add_argument(--encoding, help源文件编码, defaultutf-8) args parser.parse_args()然后你在命令行里就可以这样用python clean_text.py ./data_in ./data_out --keyword ERROR --encoding gbk这种命令行方式特别适合配合定时任务、CI流程、其它脚本调用。而且只要参数设计得合理别人拿到你的脚本不需要读代码就能试跑。正则表达式则负责更复杂的文本匹配需求。文件内容筛选、日志关键字段提取、脱敏替换用re模块可以在一个脚本里完成非常复杂的处理逻辑。比如把文件里的手机号替换成脱敏格式、把包含特定报错码的日志行单独抽出来都是在文件处理中高频出现的需求。5.3 多文件并发处理什么时候该上什么时候别上文件很多、单个文件不大的场景用普通的循环就够了完全不需要想并发的复杂事。但如果文件数量上千而且单个文件解析耗时很长可以考虑用多线程或多进程加速。文件处理多为IO密集型任务比如读取磁盘、解析文本、写入结果这种情况下多线程是有明显效果的因为线程在等待IO时会把CPU让出来给其它线程。Python的多线程虽然有GIL限制但对IO密集场景影响不大。如果处理逻辑里CPU计算占比很高比如大量正则匹配、哈希计算那就要考虑用ProcessPoolExecutor来真正利用多核CPU。from concurrent.futures import ThreadPoolExecutor from pathlib import Path def process_one(file_path: Path): # 这里写单个文件的处理逻辑 ... return file_path with ThreadPoolExecutor(max_workers8) as executor: results list(executor.map(process_one, Path(.).rglob(*.txt)))使用并发时有几条经验分享第一批先用10~20个文件做测试确定逻辑正确后再全量跑并发数从4开始递增找到一个适合你机器磁盘IO的阈值进度日志记得打不然跑多久都没底。我自己遇到过一个情况并发从4改成16速度不但没提升磁盘IO反而成了瓶颈整体更慢了。并发不是万能的适度最好。最后再分享一个个人的习惯。我写文件处理脚本不管多简单都会顺手把关键操作封装成函数哪怕脚本的主体只有十几行。这么做的原因是这类脚本往往今天处理A场景明天换一个输入目录就能处理B场景。函数封装让复用变成复制粘贴就能完成的事也让后面的测试和维护省心很多。文件处理看着是个不起眼的领域但真的能把脚本写得稳定、可复用、可扩展之后你会在日常工作中节省出大把时间。