如何用Python批量处理Zip包内.psa文件并转换为CSV格式？

需求说明 #

• 批量将Zip压缩包内的.psa文件转为同名.csv文件，重新打包输出
• 可选将所有.psa文件处理后的内容合并为单个.csv文件输出

现有有效代码（单个.psa文件处理） #

import pandas as pd
import re

fname = '1 Area 2 - store 15 group.psa'
df = pd.read_csv(fname, usecols=[0,1,2], header=None, names=['type','upc', 'num'])
store = re.search(r'store\s+(\d+)', fname).group(1)
df = df[df['type'] == 'prod'].drop(columns=['type','upc']).assign(store=store)
df.to_csv("output.csv", index=False)

原Zip处理代码问题分析 #

1. 将Zip内文件读取后直接以字符串写入本地，可能破坏.psa文件的原始格式，导致后续处理异常
2. 处理完成后仍使用原.psa文件名写入输出Zip，未实现后缀转换
3. process_file函数内写死输出name.csv，逻辑冗余且未复用处理结果

方案1：批量转为同名.csv并打包输出 #

直接操作Zip文件流，无需本地临时文件，高效处理：

import pandas as pd
import re
import zipfile
from io import StringIO

input_zip_path = r'test.zip'
output_zip_path = 'results.zip'

def process_psa_content(content, filename):
    # 从字符串内容读取DataFrame
    df = pd.read_csv(StringIO(content), usecols=[0,1,2], header=None, names=['type','upc','num'])
    store = re.search(r'store\s+(\d+)', filename).group(1)
    df = df[df['type']=='prod'].drop(columns=['type','upc']).assign(store=store)
    # 将处理后的DataFrame转为CSV字符串
    return df.to_csv(index=False)

with zipfile.ZipFile(input_zip_path, 'r') as zin, zipfile.ZipFile(output_zip_path, 'w') as zout:
    for file_info in zin.infolist():
        # 只处理.psa文件
        if file_info.filename.lower().endswith('.psa'):
            # 读取Zip内文件的原始内容（编码根据实际情况调整，如gbk）
            content = zin.read(file_info).decode('utf-8')
            # 处理内容得到CSV字符串
            csv_content = process_psa_content(content, file_info.filename)
            # 生成新的文件名：替换.psa为.csv
            new_filename = file_info.filename.rsplit('.', 1)[0] + '.csv'
            # 将CSV内容写入输出Zip
            zout.writestr(new_filename, csv_content)
        else:
            # 非.psa文件直接复制到输出Zip（可选，根据需求决定是否保留）
            zout.writestr(file_info, zin.read(file_info))

关键改进点 #

• 使用StringIO直接读取Zip内文件的字符串内容，避免本地文件操作
• 动态生成.csv后缀的新文件名，确保输出格式正确
• 直接将处理后的CSV字符串写入输出Zip，无需本地存储

方案2：合并所有处理内容为单个.csv输出 #

将所有.psa文件处理后的数据合并到一个DataFrame，最终输出单个CSV文件（可选择放入Zip或单独保存）：

import pandas as pd
import re
import zipfile
from io import StringIO

input_zip_path = r'test.zip'
output_csv_path = 'merged_output.csv'
# 可选：如果要将合并后的CSV放入Zip，启用以下代码
# output_zip_path = 'merged_results.zip'

def process_psa_content(content, filename):
    df = pd.read_csv(StringIO(content), usecols=[0,1,2], header=None, names=['type','upc','num'])
    store = re.search(r'store\s+(\d+)', filename).group(1)
    return df[df['type']=='prod'].drop(columns=['type','upc']).assign(store=store)

# 初始化总DataFrame
merged_df = pd.DataFrame()

with zipfile.ZipFile(input_zip_path, 'r') as zin:
    for file_info in zin.infolist():
        if file_info.filename.lower().endswith('.psa'):
            content = zin.read(file_info).decode('utf-8')
            processed_df = process_psa_content(content, file_info.filename)
            merged_df = pd.concat([merged_df, processed_df], ignore_index=True)

# 保存合并后的CSV
merged_df.to_csv(output_csv_path, index=False)

# 可选：如果要将合并后的CSV放入Zip
# with zipfile.ZipFile(output_zip_path, 'w') as zout:
#     zout.writestr('merged_output.csv', merged_df.to_csv(index=False))

关键说明 #

• 使用pd.concat累加所有处理后的DataFrame，确保数据完整合并
• 最终可选择将合并后的CSV单独保存，或放入新的Zip压缩包

内容的提问来源于stack exchange，提问作者uranis