本文介绍如何使用 python 自动遍历 zip 压缩包中的全部 txt 文件，统一执行列筛选、表头添加和字符串清洗操作，并将结果保存为带规范表头的 csv 文件（支持直接内存处理，无需临时磁盘文件）。
在实际数据处理中，我们常遇到一批结构一致但命名各异的纯文本文件（如 sales_01.txt、inventory_Q3.txt），它们被统一打包在 ZIP 文件中。手动逐个处理效率低下且易出错。幸运的是，Python 的 zipfile 和 pandas 模块可协同实现零临时文件、全内存流式处理，大幅提升自动化能力。
✅ 核心思路：用 ZipFile.open() 提供文件对象给 pd.read_csv()
pandas.read_csv() 不仅接受文件路径，也支持传入任意类文件对象（如 zipfile.ZipFile.open() 返回的 ZipExtFile）。这意味着我们可以跳过解压到磁盘的步骤，直接从 ZIP 流中读取内容，避免 I/O 开销与临时文件管理。
macOS UI 自动化技能包 —— 用 Python 控制你的 Mac
下载
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
? 完整可运行示例代码
import pandas as pd
import zipfile
import os
def process_txt_in_zip(zip_path: str, output_dir: str = "processed_csv"):
"""
批量处理 ZIP 内所有 .txt 文件，生成标准化 CSV。
Args:
zip_path: 输入 ZIP 文件路径
output_dir: 输出 CSV 目录（自动创建）
"""
# 创建输出目录
os.makedirs(output_dir, exist_ok=True)
with zipfile.ZipFile(zip_path, "r") as zippy:
for info in zippy.infolist():
# 仅处理 .txt 文件（忽略目录、隐藏文件等）
if not info.filename.lower().endswith(".txt") or info.is_dir():
continue
try:
# 1️⃣ 直接从 ZIP 中读取并解析：只取前两列，同时指定列名（替代 header=0 + names）
df = pd.read_csv(
zippy.open(info.filename),
usecols=[0, 1], # 保留第0、1列（原第1、2列，跳过无用列）
header=None, # 原文件无表头，设为 None 防误读
names=["store", "sku"] # 直接赋予标准列名
)
# 2️⃣ 清洗 store 列：提取 'R XXX -' 中的数字部分（如 'R 12345 -' → '12345'）
# 使用正则更鲁棒（推荐），兼容多种前缀/后缀格式
df["store"] = df["store"].str.extract(r'R\s+(\d+)', expand=False)
# 若需保留原始逻辑（split 方式），可改用以下两行（注意空值风险）：
# df["store"] = df["store"].str.split(r'R\s+', n=1).str[1]
# df["store"] = df["store"].str.split(r'\s*-\s*', n=1).str[0]
# 3️⃣ 生成输出 CSV 路径（同名替换扩展名）
csv_filename = os.path.splitext(info.filename)[0] + ".csv"
output_path = os.path.join(output_dir, csv_filename)
# 4️⃣ 保存结果（不保存索引，确保纯数据）
df.to_csv(output_path, index=False)
print(f"✅ 已处理: {info.filename} → {output_path}")
except Exception as e:
print(f"❌ 处理失败 {info.filename}: {e}")
continue
# ? 调用示例
if __name__ == "__main__":
process_txt_in_zip("data_archive.zip", output_dir="output_csv")⚠️ 关键注意事项
- 编码问题：若 TXT 文件含中文或特殊字符，请在 pd.read_csv() 中显式添加 encoding="utf-8"（或 "gbk" 等），否则可能报 UnicodeDecodeError。
- 空值与异常数据：.str.extract() 在匹配失败时返回 NaN；建议后续用 df.dropna(subset=["store"]) 过滤无效行。
- 内存安全：对超大 ZIP（GB 级别），上述方案仍高效——因每个文件独立加载、处理、释放，不会累积内存。
- 输出 ZIP 封装（进阶需求）：如需将所有 CSV 打包为新 ZIP，可在循环结束后用 zipfile.ZipFile(..., "w") 重新归档。
✅ 总结
通过 zipfile.ZipFile.open() 与 pandas.read_csv() 的无缝集成，我们实现了 ZIP 内 TXT 文件的批量化、内存化、标准化处理。相比原始单文件脚本，该方案消除了中间文件依赖，提升了健壮性与可维护性，是生产环境中处理批量结构化文本数据的理想实践。
