怎么用脚本生成数据比对报告

本文目录导读：

1. Python 脚本方案（最常用）
2. Excel 报告生成
3. Shell 脚本方案（Linux/Unix）
4. 高级功能脚本
5. 实用示例：数据库表对比
6. 使用建议

我来详细介绍几种生成数据比对报告的方法,根据不同的使用场景和技术栈：

Python 脚本方案（最常用）

基础对比脚本

import pandas as pd
import numpy as np
from datetime import datetime
def compare_dataframes(df1, df2, key_columns, report_name="comparison_report"):
    """
    比对两个DataFrame并生成报告
    """
    report = []
    report.append(f"=== 数据比对报告 - {datetime.now()} ===\n")
    # 1. 基本信息
    report.append(f"源数据行数: {len(df1)}")
    report.append(f"目标数据行数: {len(df2)}")
    # 2. 行数差异
    only_in_df1 = df1[~df1[key_columns].isin(df2[key_columns])]
    only_in_df2 = df2[~df2[key_columns].isin(df1[key_columns])]
    report.append(f"\n仅在源数据中的记录: {len(only_in_df1)}")
    report.append(f"仅在目标数据中的记录: {len(only_in_df2)}")
    # 3. 字段值对比
    common = pd.merge(df1, df2, on=key_columns, suffixes=('_source', '_target'))
    diff_columns = [col for col in df1.columns if col not in key_columns]
    for col in diff_columns:
        source_col = f"{col}_source"
        target_col = f"{col}_target"
        if source_col in common.columns and target_col in common.columns:
            differences = common[common[source_col] != common[target_col]]
            if len(differences) > 0:
                report.append(f"\n字段 '{col}' 差异数: {len(differences)}")
                # 可以输出具体差异
    return "\n".join(report)
# 使用示例
df1 = pd.read_csv('source.csv')
df2 = pd.read_csv('target.csv')
report_text = compare_dataframes(df1, df2, ['id'])
print(report_text)

生成 HTML 报告

def generate_html_report(df1, df2, key_columns):
    """生成可视化的HTML比对报告"""
    html_parts = ["""
    <html>
    <head>
        <style>
            body { font-family: Arial; padding: 20px; }
            .summary { background: #f0f0f0; padding: 10px; margin: 10px 0; }
            table { border-collapse: collapse; width: 100%; }
            th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
            .match { background-color: #d4edda; }
            .mismatch { background-color: #f8d7da; }
        </style>
    </head>
    <body>
        <h1>数据比对报告</h1>
    """]
    # 添加统计信息
    html_parts.append(f"""
    <div class="summary">
        <h3>统计摘要</h3>
        <p>源数据: {len(df1)} 行</p>
        <p>目标数据: {len(df2)} 行</p>
        <p>匹配: {len(pd.merge(df1, df2, on=key_columns))} 行</p>
    </div>
    """)
    html_parts.append("</body></html>")
    return "\n".join(html_parts)

Excel 报告生成

import openpyxl
from openpyxl.styles import PatternFill, Font
from openpyxl.utils.dataframe import dataframe_to_rows
def create_excel_comparison_report(df1, df2, key_columns, output_file):
    """生成Excel格式的比对报告"""
    wb = openpyxl.Workbook()
    # Sheet 1: 汇总信息
    ws_summary = wb.active
    ws_summary.title = "汇总"
    ws_summary['A1'] = "数据比对报告"
    # Sheet 2: 差异详情
    ws_diff = wb.create_sheet("差异详情")
    # 添加数据
    diff_df = find_differences(df1, df2, key_columns)
    for r in dataframe_to_rows(diff_df, index=False, header=True):
        ws_diff.append(r)
    # 标记差异单元格
    red_fill = PatternFill(start_color="FF0000", end_color="FF0000", fill_type="solid")
    green_fill = PatternFill(start_color="00FF00", end_color="00FF00", fill_type="solid")
    for row in ws_diff.iter_rows(min_row=2):
        for cell in row:
            if isinstance(cell.value, str) and "DIFF" in cell.value:
                cell.fill = red_fill
    wb.save(output_file)

Shell 脚本方案（Linux/Unix）

#!/bin/bash
# 数据比对脚本
generate_comparison_report() {
    local file1=$1
    local file2=$2
    local report_file="comparison_report_$(date +%Y%m%d).txt"
    echo "=== 数据比对报告 - $(date) ===" > $report_file
    echo "" >> $report_file
    # 行数对比
    echo "文件1行数: $(wc -l < $file1)" >> $report_file
    echo "文件2行数: $(wc -l < $file2)" >> $report_file
    # 差异对比
    echo "" >> $report_file
    echo "=== 差异分析 ===" >> $report_file
    diff $file1 $file2 > /tmp/diff_out.txt 2>&1
    cat /tmp/diff_out.txt >> $report_file
    echo "报告已生成: $report_file"
}
# 使用
generate_comparison_report "data_source.csv" "data_target.csv"

高级功能脚本

包含智能匹配和统计

import json
from typing import Dict, List, Any
class DataComparisonReport:
    def __init__(self):
        self.report = {
            "summary": {},
            "details": [],
            "statistics": {}
        }
    def compare_json_objects(self, obj1: Dict, obj2: Dict, path: str = ""):
        """深度对比JSON对象"""
        differences = []
        all_keys = set(list(obj1.keys()) + list(obj2.keys()))
        for key in all_keys:
            current_path = f"{path}.{key}" if path else key
            if key not in obj1:
                differences.append({
                    "path": current_path,
                    "status": "missing_in_source",
                    "value": obj2[key]
                })
            elif key not in obj2:
                differences.append({
                    "path": current_path,
                    "status": "missing_in_target",
                    "value": obj1[key]
                })
            elif obj1[key] != obj2[key]:
                differences.append({
                    "path": current_path,
                    "status": "mismatch",
                    "source_value": obj1[key],
                    "target_value": obj2[key]
                })
        return differences
    def export_report(self, format: str = "json"):
        """导出报告"""
        if format == "json":
            return json.dumps(self.report, indent=2, ensure_ascii=False)
        elif format == "text":
            return self._format_text_report()
    def _format_text_report(self):
        """格式化为文本报告"""
        lines = ["="*50, "数据比对报告", "="*50]
        # 格式化逻辑
        return "\n".join(lines)

实用示例：数据库表对比

import sqlite3
import mysql.connector
def compare_database_tables(db_config1, db_config2, table_name):
    """对比两个数据库中的表"""
    # 连接数据库
    conn1 = mysql.connector.connect(**db_config1)
    conn2 = mysql.connector.connect(**db_config2)
    # 读取数据
    df1 = pd.read_sql(f"SELECT * FROM {table_name}", conn1)
    df2 = pd.read_sql(f"SELECT * FROM {table_name}", conn2)
    # 生成报告
    report = compare_dataframes(df1, df2, ['id'])
    # 保存报告
    with open(f"db_comparison_{table_name}.txt", "w", encoding="utf-8") as f:
        f.write(report)
    conn1.close()
    conn2.close()

使用建议

1. 选择合适的工具：

   • 简单对比 → Shell 脚本
   • 数据分析 → Python pandas
   • Web 展示 → HTML/JS 方案
   • 企业级 → 商业工具如 Tableau

2. 建议：

   • 总览统计
   • 差异明细
   • 异常标记
   • 时间戳和版本信息

3. 自动化部署：

   # crontab 定时任务
   0 8 * * * /path/to/compare_script.py

需要我针对具体的使用场景,提供更详细的脚本实现吗？