本文介绍如何使用 python 批量读取 zip 压缩包内的所有 `.txt` 文件，应用列筛选、表头添加和字符串清洗等数据处理逻辑，并将结果统一导出为 csv 文件（可选：打包为新 zip）。
在实际数据工程中，我们常遇到大量结构一致但命名各异的纯文本文件（如日志、报表导出），它们被压缩在一个 ZIP 包中。手动逐个处理效率低下且易出错。Python 提供了 zipfile 和 pandas 的无缝协作能力——无需解压到磁盘，即可直接从 ZIP 流中读取并解析文本内容。
核心思路是：利用 zipfile.ZipFile.open() 返回类文件对象（file-like object），将其直接传入 pd.read_csv()；同时通过 usecols、names 和 header 参数一步完成列选取与表头注入，避免中间文件（如原代码中的 test.csv）带来的 I/O 开销和潜在错误。
以下是完整、健壮的实现方案：
macOS UI 自动化技能包 —— 用 Python 控制你的 Mac
下载
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
import pandas as pd
import zipfile
import os
from pathlib import Path
def process_txt_in_zip(input_zip: str, output_dir: str = "processed_csv", output_zip: str = None):
"""
从 ZIP 中读取所有 .txt 文件，清洗并保存为 CSV。
Args:
input_zip: 输入 ZIP 文件路径
output_dir: CSV 输出目录（自动创建）
output_zip: 可选，将所有 CSV 打包为新 ZIP 的路径
"""
# 创建输出目录
Path(output_dir).mkdir(exist_ok=True)
# 存储生成的 CSV 路径，用于后续打包
csv_files = []
with zipfile.ZipFile(input_zip, 'r') as z:
for info in z.infolist():
# 过滤 .txt 文件（忽略目录、隐藏文件等）
if not info.filename.lower().endswith('.txt') or info.is_dir():
continue
try:
# 直接从 ZIP 流读取，指定列和表头
df = pd.read_csv(
z.open(info),
usecols=[0, 1], # 仅取第1、2列（0-indexed）
header=None, # 原文件无表头
names=['store', 'sku'], # 自定义列名
dtype=str # 统一为字符串，避免数值自动转换干扰清洗
)
# 清洗 store 列：提取 "R XXX -YYY" 中的数字部分（示例逻辑）
df['store'] = df['store'].str.extract(r'R\s*(\d+)', expand=False)
# 若需更严格匹配（如去除前后空格/连字符），可链式调用：
# df['store'] = df['store'].str.strip().str.replace(r'^R\s*|\s*-\w*$', '', regex=True)
# 构建输出 CSV 路径（保持原始文件名，替换扩展名为 .csv）
base_name = Path(info.filename).stem
csv_path = Path(output_dir) / f"{base_name}.csv"
df.to_csv(csv_path, index=False)
csv_files.append(csv_path)
print(f"✅ 已处理: {info.filename} → {csv_path.name}")
except Exception as e:
print(f"❌ 处理失败 {info.filename}: {e}")
continue
# 可选：将所有 CSV 打包为新 ZIP
if output_zip and csv_files:
with zipfile.ZipFile(output_zip, 'w', zipfile.ZIP_DEFLATED) as out_z:
for csv_path in csv_files:
out_z.write(csv_path, arcname=csv_path.name)
print(f"? 已打包 {len(csv_files)} 个 CSV 到: {output_zip}")
# 使用示例
if __name__ == "__main__":
process_txt_in_zip(
input_zip="data.zip",
output_dir="output_csv",
output_zip="cleaned_data.zip"
)关键注意事项：
- ✅ 内存友好：全程不落地解压，z.open() 返回的流对象由 pandas 内部处理，适合大 ZIP 包；
- ✅ 容错设计：try/except 捕获单个文件异常，不影响其余文件处理；
- ✅ 命名规范：自动将 report_01.txt → report_01.csv，便于追溯来源；
- ⚠️ 编码问题：若 TXT 文件非 UTF-8 编码（如 GBK），需在 pd.read_csv() 中显式添加 encoding='gbk'；
- ⚠️ 正则清洗：示例中 str.extract(r'R\s*(\d+)') 更安全可靠，比多次 str.split() 更鲁棒（避免空列表索引错误）；
- ? 可扩展性：清洗逻辑可封装为独立函数（如 clean_store(s)），便于单元测试与复用。
该方案将原始脚本的 3 步操作（读取→写中间 CSV→重读→清洗→再写）压缩为单次内存内处理，兼顾性能、可维护性与工程实践标准。
