批量处理文本文件应分步构建可复用流程:优先用pathlib或glob安全定位文件,用chardet或编码列表容错读取,处理后默认输出到output/目录,加tqdm进度条与try/except错误隔离,并拆分为小函数提升可维护性。

批量处理文本文件在日常数据清洗、日志分析、文档整理中非常常见。用 Python 实现,核心是结合 os、glob、pathlib 遍历文件,再用 open 或 codecs 读写,配合 re、csv、pandas 等做内容处理。关键不是“一次写完所有功能”,而是分步构建可复用、易调试的流程。
快速定位目标文件(支持通配与递归)
别硬写 for 循环遍历目录树。优先用 pathlib(Python 3.4+ 内置,面向对象更清晰)或 glob(语法简洁)。
-
用 pathlib 找当前目录下所有 .txt 文件:
from pathlib import Path
files = list(Path(".").glob("*.txt")) -
递归查找子目录中的 .log 和 .csv:
files = list(Path(".").rglob("*.log")) + list(Path(".").rglob("*.csv")) -
用 glob 需注意:返回的是字符串路径,且不自动递归,要加
**/并设recursive=True:import glob
files = glob.glob("**/*.md", recursive=True)
安全读取多种编码的文本文件
中文环境常遇到 UnicodeDecodeError。不要默认用 utf-8 硬开——先尝试 utf-8,失败后自动 fallback 到 gbk 或检测编码。
-
简单健壮方案(推荐):
用 chardet 库自动检测(需pip install chardet):import chardet
with open(file, "rb") as f:
raw = f.read()
encoding = chardet.detect(raw)["encoding"] or "utf-8"
text = raw.decode(encoding) -
轻量替代(无第三方依赖):逐个尝试常见编码:
encodings = ["utf-8", "gbk", "latin-1"]
for enc in encodings:
try:
with open(file, encoding=enc) as f:
text = f.read()
break
except UnicodeDecodeError:
continue
按规则修改内容并保存(原地 or 新目录)
处理逻辑写清楚,保存时注意区分“覆盖原文件”和“输出到新位置”。强烈建议默认输出到 output/ 子目录,避免误删原始数据。
本文档是Websphere教程;WebSphere 是因特网的基础架构软件,也就是我们所说的中间件。它使企业能够开发、部署和集成新一代电子商务应用(如 B2B 的电子交易),并且支持从简单的 Web 发布到企业级事务处理的商务应用。希望本文档会给有需要的朋友带来帮助;感兴趣的朋友可以过来看看
立即学习“Python免费学习笔记(深入)”;
-
示例:把所有文件里的多个空格替换成单个空格,并删除行首尾空白:
import re
for file in files:
# 读取
text = read_text_safely(file) # 上一步封装好的函数
# 处理
text = re.sub(r" +", " ", text)
text = "\n".join(line.strip() for line in text.splitlines())
# 保存到 output 目录(自动创建)
out_path = Path("output") / file.name
out_path.parent.mkdir(exist_ok=True)
out_path.write_text(text, encoding="utf-8") -
如需保留原文件名结构(含子目录),用
file.relative_to(root)构造输出路径:root = Path(".")
out_path = Path("output") / file.relative_to(root)
加进度提示与错误隔离,让脚本更可靠
几百个文件跑一半报错就中断?加 try/except 捕获单个文件异常,并记录日志;用 tqdm 显示进度条(pip install tqdm)提升体验。
-
带错误跳过的循环:
from tqdm import tqdm
success, failed = 0, []
for file in tqdm(files, desc="Processing"):
try:
process_and_save(file)
success += 1
except Exception as e:
failed.append((str(file), str(e)))
print(f"完成 {success}/{len(files)},失败 {len(failed)} 个") -
把失败列表写入 report.txt 方便复查:
if failed:
with open("report.txt", "w", encoding="utf-8") as f:
for path, err in failed:
f.write(f"{path}\t{err}\n")
基本上就这些。不复杂但容易忽略细节:编码容错、路径安全、错误隔离、输出分离。把每步拆成小函数(如 read_text_safely()、clean_text()、save_to_output()),后续改需求或加新格式(比如支持 Excel 或 JSONL)就非常顺手。









