如何用Python实现POS数据CSV格式转换与自动化迁移？

核心思路拆解 #

• 先明确列映射规则：把旧CSV的每一列和新POS系统要求的列一一对应，标记出需要保留、重命名、新增（设默认值）或丢弃的列，比如旧列「交易ID」对应新列「TransactionID」，新增固定值列「门店ID」。
• 读取原始CSV数据：用Python的pandas（推荐，处理列更高效）或内置csv模块读取旧文件。
• 适配新结构：按照映射关系调整列顺序、重命名列、填充缺失列的默认值，同时清洗脏数据（比如金额字符串转数值、日期格式化）。
• 导出新CSV：将处理好的数据写入符合新系统要求的CSV文件。

代码示例（用pandas，高效简洁） #

先安装依赖：

pip install pandas

import pandas as pd

# 1. 定义新旧列映射 + 新增列的默认值
column_mapping = {
    "旧交易ID": "TransactionID",
    "旧交易时间": "TransactionTime",
    "旧商品名称": "ProductName",
    "旧交易金额": "TotalAmount"
}
extra_columns = {
    "StoreID": "POS001",  # 门店ID固定值
    "Status": "Completed"  # 默认交易状态
}

# 2. 读取旧CSV（注意编码，比如gbk或utf-8）
old_data = pd.read_csv("old_pos_data.csv", encoding="utf-8")

# 3. 重构数据结构
# 保留需要的旧列并重命名
new_data = old_data[column_mapping.keys()].rename(columns=column_mapping)
# 添加新增列
for col, val in extra_columns.items():
    new_data[col] = val

# 4. 按新系统要求调整列顺序
required_col_order = ["TransactionID", "StoreID", "TransactionTime", "ProductName", "TotalAmount", "Status"]
new_data = new_data[required_col_order]

# 5. 导出新CSV（index=False避免生成多余索引列）
new_data.to_csv("new_pos_data.csv", index=False, encoding="utf-8")

无依赖方案（用Python内置csv模块） #

如果不想安装第三方库，用内置模块也能实现：

import csv

# 列映射与新增列
column_mapping = {
    "旧交易ID": "TransactionID",
    "旧交易时间": "TransactionTime",
    "旧商品名称": "ProductName",
    "旧交易金额": "TotalAmount"
}
extra_columns = {"StoreID": "POS001", "Status": "Completed"}
new_col_names = list(column_mapping.values()) + list(extra_columns.keys())

with open("old_pos_data.csv", "r", encoding="utf-8") as old_file, \
     open("new_pos_data.csv", "w", newline="", encoding="utf-8") as new_file:
    
    reader = csv.DictReader(old_file)
    writer = csv.DictWriter(new_file, fieldnames=new_col_names)
    
    writer.writeheader()
    for row in reader:
        new_row = {}
        # 映射旧列到新列
        for old_col, new_col in column_mapping.items():
            new_row[new_col] = row[old_col]
        # 添加固定值列
        new_row.update(extra_columns)
        writer.writerow(new_row)

额外注意事项 #

• 如果旧CSV存在脏数据（比如金额带「¥」符号、日期格式不统一），需要先做清洗：比如new_data["TotalAmount"] = new_data["TotalAmount"].str.replace("¥", "").astype(float)
• 测试时可以先拿一小部分数据做验证，确认格式符合新系统要求后再批量处理。
• 找教程的话，直接搜「Python CSV数据格式转换实操」「POS数据迁移自动化案例」这类关键词，重点看列映射、数据清洗的实操部分即可。

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