如何编写Python代码按日期配对批量文件并逐一对比差异？

问题背景 #

本人Python水平有限，现有三段代码：代码1和代码2分别读取对应日期的Report1、Report2 CSV文件，去除无关列后转换为XLSX文件；代码3读取当日生成的两个XLSX文件，对比数据差异。现需实现读取文件夹中所有按日期配对的Report1File与Report2File（如Report1File 02.04.2024和Report2File 02.04.2024），逐一配对并对比差异，不知如何推进。

实现步骤与代码 #

核心思路 #

核心是从文件名中解析日期，将同日期的Report1和Report2文件配对，再循环执行“CSV转XLSX”和“文件对比”的逻辑。下面是完整可运行的代码：

1. 封装工具函数

把重复的转换、对比逻辑封装成函数，方便批量调用：

import pandas as pd
import os
from datetime import datetime

# CSV转XLSX函数（兼容Report1/Report2）
def csv_to_xlsx(csv_path, report_type):
    try:
        # 读取CSV文件（无表头）
        df = pd.read_csv(csv_path, header=None)
        # 按逗号拆分第一列为多列
        max_commas = df[0].str.split(',').transform(len).max()
        df[[f'name_{x}' for x in range(max_commas)]] = df[0].str.split(',', expand=True)
        # 删除原始第一列和name_0列（时间列）
        df.drop(0, axis=1, inplace=True)
        df = df.drop(columns=[col for col in df.columns if col == 'name_0'])
        # 从文件名提取日期（适配原CSV命名格式：Report1 -YYYY-MM-DD.csv）
        file_name = os.path.basename(csv_path)
        date_str = file_name.split('-')[-1].replace('.csv', '').strip()
        # 生成输出XLSX路径
        output_path = os.path.join(os.path.dirname(csv_path), f"{report_type}File {date_str}.xlsx")
        # 保存文件
        df.to_excel(output_path, index=None)
        print(f"转换完成: {output_path}")
        return output_path
    except Exception as e:
        print(f"{csv_path} 转换失败: {str(e)}")
        return None

# XLSX文件对比函数
def compare_xlsx(file1_path, file2_path):
    try:
        # 读取XLSX（注意skiprows需根据实际文件调整）
        df1 = pd.read_excel(file1_path, skiprows=[2])
        df2 = pd.read_excel(file2_path, skiprows=[3])
        # 对比数据差异
        diff = df1.compare(df2)
        # 提取日期用于结果输出
        date_str = os.path.basename(file1_path).split(' ')[-1].replace('.xlsx', '')
        if diff.empty:
            print(f"日期 {date_str}: 无数据差异")
        else:
            print(f"日期 {date_str}: 发现差异如下")
            print(diff)
            # 可选：将差异保存为单独文件
            # diff.to_excel(os.path.join(os.path.dirname(file1_path), f"差异报告_{date_str}.xlsx"))
    except Exception as e:
        print(f"{file1_path} 与 {file2_path} 对比失败: {str(e)}")

2. 主程序：批量处理与配对对比

def main(target_folder):
    # 1. 遍历文件夹，分类收集CSV和XLSX文件
    file_collection = {
        'csv': {'Report1': [], 'Report2': []},
        'xlsx': {'Report1': [], 'Report2': []}
    }

    for filename in os.listdir(target_folder):
        full_path = os.path.join(target_folder, filename)
        if not os.path.isfile(full_path):
            continue
        
        # 分类收集文件
        if filename.startswith('Report1 -') and filename.endswith('.csv'):
            file_collection['csv']['Report1'].append(full_path)
        elif filename.startswith('Report2 -') and filename.endswith('.csv'):
            file_collection['csv']['Report2'].append(full_path)
        elif filename.startswith('Report1File ') and filename.endswith('.xlsx'):
            file_collection['xlsx']['Report1'].append(full_path)
        elif filename.startswith('Report2File ') and filename.endswith('.xlsx'):
            file_collection['xlsx']['Report2'].append(full_path)
    
    # 2. 批量转换CSV为XLSX（如果存在未转换的CSV）
    if file_collection['csv']['Report1'] or file_collection['csv']['Report2']:
        print("\n=== 开始转换CSV文件 ===")
        for csv_file in file_collection['csv']['Report1']:
            csv_to_xlsx(csv_file, 'Report1')
        for csv_file in file_collection['csv']['Report2']:
            csv_to_xlsx(csv_file, 'Report2')
        
        # 转换后重新收集XLSX文件，确保包含刚生成的文件
        file_collection['xlsx'] = {'Report1': [], 'Report2': []}
        for filename in os.listdir(target_folder):
            full_path = os.path.join(target_folder, filename)
            if filename.startswith('Report1File ') and filename.endswith('.xlsx'):
                file_collection['xlsx']['Report1'].append(full_path)
            elif filename.startswith('Report2File ') and filename.endswith('.xlsx'):
                file_collection['xlsx']['Report2'].append(full_path)
    
    # 3. 按日期配对XLSX文件
    date_pairing = {}
    # 处理Report1的XLSX文件，提取日期
    for xlsx_file in file_collection['xlsx']['Report1']:
        date_str = os.path.basename(xlsx_file).split(' ')[-1].replace('.xlsx', '')
        if date_str not in date_pairing:
            date_pairing[date_str] = {'Report1': None, 'Report2': None}
        date_pairing[date_str]['Report1'] = xlsx_file
    # 处理Report2的XLSX文件，补全配对
    for xlsx_file in file_collection['xlsx']['Report2']:
        date_str = os.path.basename(xlsx_file).split(' ')[-1].replace('.xlsx', '')
        if date_str not in date_pairing:
            date_pairing[date_str] = {'Report1': None, 'Report2': None}
        date_pairing[date_str]['Report2'] = xlsx_file
    
    # 4. 逐一对比配对文件
    print("\n=== 开始对比文件 ===")
    for date_str, files in date_pairing.items():
        if files['Report1'] and files['Report2']:
            compare_xlsx(files['Report1'], files['Report2'])
        else:
            missing = []
            if not files['Report1']:
                missing.append('Report1File')
            if not files['Report2']:
                missing.append('Report2File')
            print(f"日期 {date_str}: 缺少 {'、'.join(missing)}")

# 执行主程序，替换为你的目标文件夹路径
if __name__ == "__main__":
    target_folder = "C:/Users/User/Python/Project"
    main(target_folder)

关键调整点 #

• 日期格式适配：如果你的文件名日期格式是DD.MM.YYYY（比如02.04.2024），需要修改提取日期的代码，将split('-')改为split('.')，并调整日期字符串的处理逻辑。
• skiprows参数：原代码中Report1跳过2行、Report2跳过3行，需根据实际文件的表头行数调整，否则会读错数据。
• 差异保存：如果需要将对比结果保存为文件，取消对比函数中diff.to_excel的注释即可。

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