Python对比每日CSV差异并生成周变更统计新CSV的实现方法

核心实现思路 #

• 第一步：按日期命名规则加载所有历史CSV文件，先通过文件哈希做粗粒度校验，内容完全一致的无更新日期直接跳过，减少无效比对
• 第二步：逐行逐单元格比对相邻有效版本，记录每个单元格（行唯一标识+列名组合）发生值变更的具体日期
• 第三步：按自然周规则对所有变更日期做分组，统计每个单元格在对应自然周内发生值更新的累计天数
• 第四步：将统计结果导出为汇总CSV文件

前置依赖 #

安装数据处理库pandas，执行命令：
pip install pandas

完整实现代码 #

import os
import hashlib
import pandas as pd
from datetime import datetime

# -------------------------- 可配置参数 --------------------------
CSV_DIR = "./csv_versions"  # 所有日更CSV存放的文件夹路径
FILE_NAME_PATTERN = "data_%Y%m%d.csv"  # 文件名日期格式，例：data_20220714.csv对应2022年7月14日
UNIQUE_KEY = "id"  # CSV中唯一标识一行的列名（如用户ID、商品ID），无唯一列则填None
OUTPUT_PATH = "./weekly_update_stats.csv"  # 统计结果导出路径
WEEK_FIRST_DAY = "monday"  # 周起始日，可选值monday/sunday
# ----------------------------------------------------------------

def get_file_md5(file_path):
    """计算文件MD5值，快速判断两个文件内容是否完全一致"""
    with open(file_path, 'rb') as f:
        return hashlib.md5(f.read()).hexdigest()

def load_all_csv_versions():
    """加载所有有效版本CSV，跳过内容完全重复的无更新日期"""
    version_list = []
    last_file_md5 = None
    for file_name in sorted(os.listdir(CSV_DIR)):
        try:
            file_date = datetime.strptime(file_name, FILE_NAME_PATTERN).date()
        except ValueError:
            continue  # 跳过不符合命名规则的无关文件
        file_path = os.path.join(CSV_DIR, file_name)
        current_md5 = get_file_md5(file_path)
        if current_md5 == last_file_md5:
            continue  # 和上一版本内容完全一致，无更新直接跳过
        df = pd.read_csv(file_path, dtype=str).fillna("")
        version_list.append({"date": file_date, "df": df})
        last_file_md5 = current_md5
    return version_list

def track_all_changes(version_list):
    """追踪所有单元格的变更日期"""
    if not version_list:
        return pd.DataFrame()
    change_records = {}
    # 第一个版本作为基准，所有单元格记录首次上线日期
    first_df = version_list[0]["df"]
    first_date = version_list[0]["date"]
    for _, row in first_df.iterrows():
        row_key = row[UNIQUE_KEY] if UNIQUE_KEY else row.name
        for col in first_df.columns:
            change_records[(row_key, col)] = [first_date]
    # 逐版本比对差异
    for i in range(1, len(version_list)):
        current_ver = version_list[i]
        current_date = current_ver["date"]
        current_df = current_ver["df"]
        last_df = version_list[i-1]["df"]
        # 对齐两个版本的行和列，缺失值补空
        if UNIQUE_KEY:
            merged_index = pd.concat([last_df[UNIQUE_KEY], current_df[UNIQUE_KEY]]).drop_duplicates()
            last_aligned = last_df.set_index(UNIQUE_KEY).reindex(merged_index).fillna("")
            current_aligned = current_df.set_index(UNIQUE_KEY).reindex(merged_index).fillna("")
        else:
            max_row = max(len(last_df), len(current_df))
            last_aligned = last_df.reindex(range(max_row)).fillna("")
            current_aligned = current_df.reindex(range(max_row)).fillna("")
        # 逐单元格标记差异
        diff_mask = (last_aligned != current_aligned)
        for row_key in diff_mask.index:
            for col in diff_mask.columns:
                if diff_mask.loc[row_key, col]:
                    if (row_key, col) not in change_records:
                        change_records[(row_key, col)] = []
                    change_records[(row_key, col)].append(current_date)
    return change_records

def calc_weekly_stats(change_records):
    """按自然周统计每个单元格的更新天数"""
    all_rows = set()
    all_cols = set()
    week_range_map = {}
    # 收集所有行、列、周维度信息
    for (row_key, col), date_list in change_records.items():
        all_rows.add(row_key)
        all_cols.add(col)
        for d in date_list:
            if WEEK_FIRST_DAY == "monday":
                year, week_num = d.isocalendar()[0], d.isocalendar()[1]
                week_start = d - pd.Timedelta(days=d.weekday())
            else:
                year, week_num = d.year, (d.day + d.replace(day=1).weekday()) // 7 + 1
                week_start = d - pd.Timedelta(days=(d.weekday() + 1) % 7)
            week_end = week_start + pd.Timedelta(days=6)
            week_tag = f"{year}年第{week_num:02d}周"
            week_range_map[week_tag] = (week_start, week_end)
    # 生成统计结果
    result_rows = []
    for row_key in sorted(all_rows):
        row_data = {UNIQUE_KEY if UNIQUE_KEY else "row_id": row_key}
        for col in all_cols:
            date_list = change_records.get((row_key, col), [])
            for week_tag in sorted(week_range_map.keys()):
                col_name = f"{col}_{week_tag}"
                week_start, week_end = week_range_map[week_tag]
                update_count = sum([1 for d in date_list if week_start <= d <= week_end])
                row_data[col_name] = update_count
        result_rows.append(row_data)
    return pd.DataFrame(result_rows)

if __name__ == "__main__":
    versions = load_all_csv_versions()
    records = track_all_changes(versions)
    stats_df = calc_weekly_stats(records)
    stats_df.to_csv(OUTPUT_PATH, index=False, encoding="utf-8-sig")
    print(f"统计完成，结果已导出到{OUTPUT_PATH}")

关键逻辑说明 #

• 性能优化：优先通过文件MD5做粗粒度校验，内容完全相同的日期版本直接跳过逐格比对，大幅降低长周期文件的处理耗时
• 变更计数规则：同一单元格同一自然日内无论值修改几次，只要和上一版本存在差异就只计1天更新，不会重复计数
• 兼容结构变更：如果CSV中途新增列、新增行，代码会自动识别，首次出现的日期会被记为第一次更新
• 周度规则可配置：支持切换周一/周日作为周起始日，符合不同场景的周统计要求

注意：如果你的CSV没有唯一标识每行的主键列，请保证所有版本CSV的行顺序固定，不要随意调整行排序，否则会导致行匹配错位，统计结果出错。

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