
本文介绍如何使用pandas直接读取psa文本文件,精准筛选以“prod”开头的行、提取第3列(索引为2)数值,并自动解析文件名中的门店编号,最终生成结构清晰的csv表格,全程无需中间临时文件。
在处理批量PSA格式日志或报表文件时,常见需求是:仅提取含特定标识(如prod)的行中某固定位置的字段(如第3个逗号分隔值),同时将文件名中隐含的关键信息(如门店编号)作为辅助列一并写入结果。传统做法常依赖先转存为TXT/CSV再二次处理,不仅冗余,还易引入编码、换行或列对齐问题。
以下为高效、健壮的解决方案,完全基于Pandas原生IO能力,跳过临时文件环节:
✅ 核心步骤说明
- 直接读取原始PSA文件:用 pd.read_csv() 指定 usecols=[0, 2] 仅加载第1列(类型标识)和第3列(目标数值),避免读入全部30+列,提升性能;
- 赋予临时列名:通过 names=['type', 'num'] 明确语义,便于后续逻辑操作;
- 条件过滤与列清理:df[df['type'] == 'prod'] 筛选目标行,.drop(columns='type') 移除冗余标识列;
- 动态注入文件元数据:利用正则 re.search(r'store\s+(\d+)', fname) 从文件名中安全提取数字编号(支持 store 15、store_15 等常见变体);
- 合并结果并导出:.assign(store=store) 新增门店列,to_csv() 一键输出标准CSV。
? 完整可运行代码(支持单文件 & 批量处理)
import pandas as pd
import re
import os
from pathlib import Path
def process_psa_file(filepath: str) -> pd.DataFrame:
"""处理单个PSA文件:提取prod行第3列 + 解析门店号"""
# 仅读取第1列(索引0)和第3列(索引2)
df = pd.read_csv(
filepath,
usecols=[0, 2],
header=None,
names=['type', 'num'],
skip_blank_lines=True,
on_bad_lines='skip' # 自动跳过格式异常行(如空行、列数不足)
)
# 过滤以 'prod' 开头的行,并移除type列
df_filtered = df[df['type'] == 'prod'].drop(columns='type').copy()
# 从文件名提取store编号(兼容多种命名格式)
filename = Path(filepath).stem
store_match = re.search(r'(?:store|STORE)\s*(\d+)|(\d+)\s*(?:store|STORE)', filename)
store_num = store_match.group(1) if store_match and store_match.group(1) else \
store_match.group(2) if store_match else "UNKNOWN"
# 添加store列并返回
return df_filtered.assign(store=store_num)
# --- 批量处理示例(遍历ZIP解压后的文件夹)---
psa_folder = Path("path/to/your/psa_files") # 替换为实际路径
output_csv = "all_stores_output.csv"
all_dfs = []
for psa_file in psa_folder.glob("*.psa"):
try:
df = process_psa_file(psa_file)
all_dfs.append(df)
print(f"✓ 已处理: {psa_file.name} → 提取 {len(df)} 条 prod 数据")
except Exception as e:
print(f"✗ 处理失败 {psa_file.name}: {e}")
if all_dfs:
final_df = pd.concat(all_dfs, ignore_index=True)
final_df.to_csv(output_csv, index=False)
print(f"\n✅ 合并完成!总计 {len(final_df)} 行,已保存至 {output_csv}")
else:
print("⚠️ 未找到有效PSA文件或全部处理失败")⚠️ 关键注意事项
- 索引理解:usecols=[0, 2] 对应第1列(索引0)和第3列(索引2),因Python索引从0开始,切勿误用 [2](会报错);
- 文件名鲁棒性:正则表达式已覆盖 store 15、STORE_15、15_store 等常见格式,若命名规则特殊,可按需调整 re.search() 模式;
- 错误容错:on_bad_lines='skip' 和 skip_blank_lines=True 可避免因文件末尾空行或损坏行导致中断;
- 编码问题:若PSA文件非UTF-8(如GBK),在 read_csv() 中添加 encoding='gbk' 参数;
- 内存优化:对超大文件,可改用 chunksize 分块处理,但本场景通常无需。
该方案兼具简洁性、可维护性与生产就绪性,一次编写即可稳定支撑千级PSA文件的自动化清洗任务。










