基于Python Pandas实现CSV与XML文件数据差异校验及提示输出

完整代码实现 #

import pandas as pd

def read_csv(file_path):
    # 读取CSV文件，兼容UTF-8/GBK编码解决中文乱码问题
    try:
        return pd.read_csv(file_path, encoding='utf-8')
    except UnicodeDecodeError:
        return pd.read_csv(file_path, encoding='gbk')

def read_xml(file_path, xpath='.//record'):
    # 读取XML文件，xpath参数指定需解析的目标节点（根据XML结构调整）
    return pd.read_xml(file_path, xpath=xpath)

def compare_files(df_csv, df_xml, primary_key):
    # 校验主键字段是否存在于两个数据集
    if primary_key not in df_csv.columns or primary_key not in df_xml.columns:
        print(f"错误：主键字段「{primary_key}」在其中一个文件中不存在")
        return
    
    # 提取公共字段作为对比子集，排除无关列干扰
    common_columns = list(set(df_csv.columns) & set(df_xml.columns))
    df_csv_subset = df_csv[common_columns].drop_duplicates().reset_index(drop=True)
    df_xml_subset = df_xml[common_columns].drop_duplicates().reset_index(drop=True)
    
    # 1. 检测单向缺失数据
    missing_in_csv = df_xml_subset.merge(
        df_csv_subset, on=primary_key, how='left', indicator=True
    ).query('_merge == "left_only"').drop('_merge', axis=1)
    
    missing_in_xml = df_csv_subset.merge(
        df_xml_subset, on=primary_key, how='left', indicator=True
    ).query('_merge == "left_only"').drop('_merge', axis=1)
    
    # 2. 检测内容不一致数据（主键存在但其他字段值不同）
    merged_df = pd.merge(
        df_csv_subset, df_xml_subset, on=primary_key, suffixes=('_csv', '_xml'), how='inner'
    )
    mismatch_mask = False
    for col in common_columns:
        if col == primary_key:
            continue
        mismatch_mask |= merged_df[f"{col}_csv"] != merged_df[f"{col}_xml"]
    mismatches = merged_df[mismatch_mask].reset_index(drop=True)
    
    # 输出校验结果
    print("=== 数据同步校验结果 ===")
    if not missing_in_csv.empty:
        print(f"\n1. CSV中缺失但XML存在的数据（共{len(missing_in_csv)}条）：")
        print(missing_in_csv.to_string(index=False))
    if not missing_in_xml.empty:
        print(f"\n2. XML中缺失但CSV存在的数据（共{len(missing_in_xml)}条）：")
        print(missing_in_xml.to_string(index=False))
    if not mismatches.empty:
        print(f"\n3. 双向存在但内容不一致的数据（共{len(mismatches)}条）：")
        print(mismatches.to_string(index=False))
    if missing_in_csv.empty and missing_in_xml.empty and mismatches.empty:
        print("\n所有数据同步一致，未发现异常")

# 调用示例
if __name__ == "__main__":
    CSV_FILE = "your_data.csv"
    XML_FILE = "your_data.xml"
    PRIMARY_KEY = "id"  # 替换为数据的唯一标识字段（如ID、编码）
    
    csv_data = read_csv(CSV_FILE)
    xml_data = read_xml(XML_FILE)
    compare_files(csv_data, xml_data, PRIMARY_KEY)

关键逻辑说明 #

• 文件读取：
  • CSV读取自动适配常见编码，避免中文乱码
  • XML读取支持自定义XPath，适配不同结构的XML文件（需根据实际节点调整xpath参数）
• 数据对比：
  • 先提取两个文件的公共字段作为对比子集，排除无关列干扰
  • 用merge+indicator精准检测单向缺失的数据
  • 内连接合并后逐字段对比，筛选出内容不一致的行
• 结果输出：分三类输出异常情况，无异常时直接提示同步一致

内容的提问来源于stack exchange，提问作者My Vö