Python处理多表头银行CSV文件的最优方案咨询

问题背景 #

正在构建个人财务规划器，核心流程为：

• 下载不同银行账户的CSV格式账单
• 用Python遍历文件夹读取所有账单数据
• 将标准化后的数据写入SQL数据库

当前核心痛点：各银行CSV表头结构不一致，目前为每家银行编写单独的转换函数，代码重复度极高，后续修改或新增银行时维护成本陡增。

表头示例 #

• 银行1：['account_num', 'text', 'date', 'valutadate', 'amount', 'currency']
• 银行2：['account_num', 'text', 'valutadate', 'currency', 'amount_withdrawal', 'amount_deposit']

目标统一结构 #

['account_num', 'text', 'valutadate', 'amount', 'currency', 'category']

优化方案：配置驱动+通用转换函数 #

核心思路是将银行特有的转换规则抽离为可配置的字典，用单一通用函数处理所有银行的CSV文件，彻底消除重复代码。

1. 定义银行转换配置 #

把每个银行的表头映射、金额计算逻辑、日期格式、分隔符等规则统一存入配置字典，后续新增银行仅需添加配置项，无需修改核心代码：

# 银行配置字典，key为银行标识（可通过文件名前缀/后缀区分）
BANK_CONFIGS = {
    "bank1": {
        "csv_header": ['account_num', 'text', 'date', 'valutadate', 'amount', 'currency'],
        "delimiter": ';',
        "skip_header": False,  # 是否跳过CSV自带的第一行表头
        "field_mapping": {
            "account_num": "account_num",
            "text": "text",
            "valutadate": "valutadate",
            "currency": "currency"
        },
        "amount_calculator": lambda row: float(row['amount'].replace('.','').replace(',','.')),
        "date_formats": {
            "date": "%d.%m.%Y",
            "valutadate": "%d.%m.%Y"
        }
    },
    "bank2": {
        "csv_header": ['account_num', 'text', 'valutadate', 'currency', 'amount_withdrawal', 'amount_deposit'],
        "delimiter": ';',
        "skip_header": True,
        "field_mapping": {
            "account_num": "account_num",
            "text": "text",
            "valutadate": "valutadate",
            "currency": "currency"
        },
        "amount_calculator": lambda row: float(row['amount_deposit'].replace('.','').replace(',','.')) - float(row['amount_withdrawal'].replace('.','').replace(',','.')),
        "date_formats": {
            "valutadate": "%d.%m.%Y"
        }
    }
}

2. 编写通用转换函数 #

基于配置字典实现通用处理逻辑，将文件读取、数据校验、字段转换、分类匹配等重复逻辑统一封装：

import csv
import datetime
import pandas as pd
import match_c  # 假设你的分类匹配模块已导入

def process_bank_csv(filename, bank_id):
    config = BANK_CONFIGS.get(bank_id)
    if not config:
        raise ValueError(f"未识别的银行类型：{bank_id}")
    
    # 目标标准化表头
    target_columns = ['account_num', 'text', 'valutadate', 'amount', 'currency', 'category']
    result_df = pd.DataFrame(columns=target_columns)
    
    with open(filename, 'r') as fhandle:
        csv_reader = csv.DictReader(fhandle, delimiter=config['delimiter'], fieldnames=config['csv_header'])
        
        # 跳过CSV自带表头（按配置要求）
        if config['skip_header']:
            next(csv_reader)
        
        for row_index, row in enumerate(csv_reader):
            # 数据长度校验，错误行跳过而非中断整个文件处理
            if len(row) != len(config['csv_header']):
                print(f"第{row_index}行数据异常，预期{len(config['csv_header'])}个字段，实际{len(row)}个")
                continue
            
            # 映射基础字段
            record = {}
            for target_field, source_field in config['field_mapping'].items():
                record[target_field] = row[source_field]
            
            # 格式化日期字段
            for date_field, fmt in config['date_formats'].items():
                if date_field in record:
                    record[date_field] = datetime.strptime(record[date_field], fmt)
            
            # 计算金额（调用配置中的自定义逻辑）
            record['amount'] = config['amount_calculator'](row)
            
            # 匹配交易分类
            record['category'] = match_c.match_category(record['text'])
            
            # 写入结果DataFrame
            result_df.loc[row_index] = [
                record['account_num'],
                record['text'],
                record['valutadate'],
                record['amount'],
                record['currency'],
                record['category']
            ]
    
    print(f"{bank_id}文件处理完成，共{len(result_df)}条有效记录")
    return result_df

3. 批量处理文件夹内文件 #

通过文件名规则自动识别所属银行，实现批量处理：

import os

def process_folder(folder_path):
    for filename in os.listdir(folder_path):
        if not filename.endswith('.csv'):
            continue
        
        file_path = os.path.join(folder_path, filename)
        
        # 根据文件名判断银行（可根据实际命名规则调整）
        if 'bank1' in filename.lower():
            df = process_bank_csv(file_path, 'bank1')
        elif 'bank2' in filename.lower():
            df = process_bank_csv(file_path, 'bank2')
        else:
            print(f"无法识别文件所属银行：{filename}")
            continue
        
        # 此处添加写入SQL数据库逻辑，例如：
        # df.to_sql('financial_records', your_db_connection, if_exists='append', index=False)

额外优化建议 #

• 用Pandas简化读取逻辑：直接使用pd.read_csv指定表头、分隔符，结合apply函数处理字段转换，进一步精简代码。
• 增强错误处理：添加try-except捕获日期解析、金额转换异常，记录错误日志而非仅打印提示。
• 配置文件分离：将BANK_CONFIGS写入JSON/YAML文件，无需修改代码即可新增或调整银行规则。
• 重复数据检测：利用你注释中的哈希值逻辑，在写入数据库前检测重复记录，避免重复插入。

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