本文介绍如何使用pandas直接读取逗号分隔的psa文本文件，精准筛选以“prod”开头的行、提取第3列（索引为2）数值，并自动解析文件名中的门店编号作为第二列，最终导出结构清晰的csv，全程无需中间临时文件。
在处理批量PSA类日志或报告文件时，常需跳过头部注释、忽略无关记录（如pla行），仅提取关键字段（如prod行的第3个值），同时将文件元信息（如门店编号）结构化写入结果。传统做法（先复制到TXT再读取）不仅冗余，还易引入编码、换行或列对齐问题。更优解是：用pandas原生支持的灵活读取 + 正则解析 + 链式数据操作，一步到位。
以下为完整、可扩展的实现方案：
Knowledge Graph - Csv Graph Loader Generator
下载
从 CSV 数据集生成图数据库加载器和三元组映射。将表格 CSV 数据转换为图数据库可直接使用的节点、边和三元组。
✅ 核心步骤说明
- 直接读取原始PSA文件：使用 pd.read_csv(..., header=None) 避免首行被误判为列名；usecols=[0, 2] 精确指定只需第1列（类型标识）和第3列（目标数值），跳过全部中间列，提升效率与健壮性；
- 动态命名列：通过 names=['type', 'num'] 明确语义，便于后续逻辑过滤；
- 条件筛选与清理：df[df['type'] == 'prod'] 保留仅含 prod 的行，.drop(columns='type') 移除冗余标识列；
- 文件名智能解析：借助 re.search(r'store\s+(\d+)', filename) 提取连续数字形式的门店号（如 "store 15" → "15"），确保兼容空格/大小写变体；
- 注入元数据列：.assign(store=store) 原地添加 store 列，避免 df['store'] = store 可能引发的SettingWithCopyWarning；
- 批量处理就绪：该逻辑可无缝嵌入 pathlib.Path 或 os.listdir() 循环，遍历ZIP解压后的所有PSA文件。
? 完整可运行代码（含批量支持）
import pandas as pd
import re
from pathlib import Path
def process_psa_file(filepath: Path) -> pd.DataFrame:
# 读取：仅取第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 列
result = df[df['type'] == 'prod'].drop(columns='type').copy()
# 从文件名提取 store 编号（支持 "store 15", "Store:15", "store_15" 等常见变体）
store_match = re.search(r'(?:store|STORE|Store)[\s_:.-]+(\d+)', filepath.name)
store_num = store_match.group(1) if store_match else "unknown"
# 添加 store 列并重排列顺序
result = result.assign(store=store_num)[['num', 'store']]
return result
# 批量处理示例（假设所有 PSA 文件位于 ./psa_files/ 目录下）
input_dir = Path("./psa_files")
output_file = "all_stores_output.csv"
all_dfs = []
for psa_path in input_dir.glob("*.psa"):
try:
df = process_psa_file(psa_path)
print(f"✅ 处理完成: {psa_path.name} → {len(df)} 条 prod 记录")
all_dfs.append(df)
except Exception as e:
print(f"❌ 处理失败 {psa_path.name}: {e}")
# 合并所有结果并导出
if all_dfs:
final_df = pd.concat(all_dfs, ignore_index=True)
final_df.to_csv(output_file, index=False)
print(f"\n? 合并完成！总计 {len(final_df)} 行，已保存至 {output_file}")
else:
print("⚠️ 未找到任何有效PSA文件")⚠️ 注意事项与最佳实践
- 编码兼容性：若PSA文件含中文或特殊字符，建议在 pd.read_csv() 中显式指定 encoding='utf-8' 或 'gbk'；
- 列索引校验：PSA文件若存在空行或不规则分隔符，启用 on_bad_lines='skip' + skip_blank_lines=True 可增强鲁棒性；
- 正则健壮性：示例正则 r'store\s+(\d+)' 假设门店号紧邻单词 store；如实际格式多变（如 "15_store.psa"），可升级为 r'(\d+)(?=_store|\s+store|\.psa)'；
- 内存优化：处理超大文件时，可用 chunksize 参数分块读取，逐块过滤后追加写入CSV，避免全量加载。
此方案彻底规避了临时文件IO开销，逻辑内聚、易于维护，并天然支持生产环境中的批量自动化场景。
