使用Python将多个CSV文件的DataFrame表头统一重命名为标准表头

核心思路是提前维护异名同义列名到标准列名的映射表，读取CSV时自动匹配替换列名，配套简单的容错、校验逻辑，覆盖大小写、首尾空格等常见格式差异，也能及时发现未收录的列名别名。

实现步骤 #

1. 先定义匹配规则 #

首先明确目标标准表头，再把所有遇到过的非标准列名整理为映射字典。为了降低维护成本，匹配前先对原始列名做归一化处理（去首尾空格、统一转小写），不需要单独为大小写、前后空格的差异重复写映射规则。

import pandas as pd
import os

# 目标标准表头
STANDARD_COLUMNS = ["Txn Date", "Description", "Credit", "Debit", "Balance"]

# 列名映射：key为归一化后的非标准列名，value为对应标准列名
COLUMN_MAPPING = {
    # Txn Date 别名
    "date": "Txn Date",
    "txn date": "Txn Date",
    # Description 别名
    "description": "Description",
    # Credit（收入/贷方/存款）别名
    "credit": "Credit",
    "deposit": "Credit",
    # Debit（支出/借方/取款）别名
    "debit": "Debit",
    "withdrawl": "Debit",
    # Balance（余额）别名
    "balance": "Balance",
    "total balance": "Balance"
}

def normalize_col(col_name: str) -> str:
    """列名归一化工具函数"""
    return col_name.strip().lower()

以上映射已经覆盖问题中提到的两类CSV的所有列名，后续遇到新的别名直接往字典里追加键值对即可

2. 编写处理逻辑 #

先写单个CSV的处理函数，完成列名替换、合法性校验、列顺序调整，再写批量遍历逻辑处理整个文件夹下的所有CSV，支持单独存储处理后的文件，也支持直接合并为一个总表。

def load_and_standardize_csv(file_path: str) -> pd.DataFrame:
    """读取单个CSV并统一为标准表头"""
    df = pd.read_csv(file_path)
    # 构建列名重命名字典
    rename_rule = {}
    for raw_col in df.columns:
        norm_col = normalize_col(raw_col)
        if norm_col in COLUMN_MAPPING:
            rename_rule[raw_col] = COLUMN_MAPPING[norm_col]
        else:
            print(f"提示：文件 {os.path.basename(file_path)} 存在未收录列名：{raw_col}，请补充映射规则")
    # 执行重命名
    df = df.rename(columns=rename_rule)
    # 校验必要列是否齐全
    missing_cols = [col for col in STANDARD_COLUMNS if col not in df.columns]
    if missing_cols:
        raise ValueError(f"文件 {os.path.basename(file_path)} 缺失必要列：{missing_cols}")
    # 按标准顺序保留列，丢弃多余列
    return df[STANDARD_COLUMNS]

def batch_process(raw_dir: str, output_dir: str = "./standardized_csv", merge_all: bool = False):
    """
    批量处理目录下所有CSV文件
    :param raw_dir: 原始CSV存放目录
    :param output_dir: 处理后文件输出目录
    :param merge_all: 是否将所有文件合并为单个CSV
    """
    os.makedirs(output_dir, exist_ok=True)
    processed_dfs = []
    for filename in os.listdir(raw_dir):
        if filename.lower().endswith(".csv"):
            file_path = os.path.join(raw_dir, filename)
            std_df = load_and_standardize_csv(file_path)
            processed_dfs.append(std_df)
            if not merge_all:
                std_df.to_csv(os.path.join(output_dir, filename), index=False)
                print(f"已完成处理：{filename}")
    if merge_all:
        merged_df = pd.concat(processed_dfs, ignore_index=True)
        merged_df.to_csv(os.path.join(output_dir, "all_transactions.csv"), index=False)
        print(f"全部文件合并完成，总计{len(merged_df)}条记录")

3. 调用方式 #

把所有待处理的原始CSV放到同一个文件夹，传入路径即可运行：

# 替换为本地的实际路径
batch_process(
    raw_dir="./raw_bill_csvs",
    output_dir="./standard_bill_csvs",
    merge_all=False # 需要合并成总表就设为True
)

补充说明 #

• 代码自带的校验逻辑会在遇到未收录列名、缺失必要列时给出提示，避免静默处理导致的数据错配
• 如果遇到列名存在拼写差异、部分匹配的场景，可以扩展normalize_col函数，或者加模糊匹配规则适配
• 处理后的DataFrame会严格按照标准表头的顺序排列列，后续做统计、合并时不会出现列顺序错乱的问题

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