脚本怎样比对两端数据内容

本文目录导读：

1. Python - 基础文件比对
2. Python - 使用difflib库（更专业）
3. Python - 大数据集高效比对
4. Shell脚本 - 文件比较
5. JSON数据比对
6. 实时监控文件变化
7. 性能优化建议
8. 使用建议

我来介绍几种常用的数据比对脚本方法,涵盖不同场景：

Python - 基础文件比对

# 读取两个文件并比对
with open('file1.txt', 'r') as f1, open('file2.txt', 'r') as f2:
    lines1 = f1.readlines()
    lines2 = f2.readlines()
# 逐行比对
for i, (line1, line2) in enumerate(zip(lines1, lines2), 1):
    if line1 != line2:
        print(f"差异在第{i}行:")
        print(f"文件1: {line1.rstrip()}")
        print(f"文件2: {line2.rstrip()}")
        print()
# 行数不一致时
if len(lines1) != len(lines2):
    print(f"行数不同: 文件1={len(lines1)}行, 文件2={len(lines2)}行")

Python - 使用difflib库（更专业）

import difflib
def compare_files(file1_path, file2_path):
    with open(file1_path, 'r', encoding='utf-8') as f1:
        lines1 = f1.readlines()
    with open(file2_path, 'r', encoding='utf-8') as f2:
        lines2 = f2.readlines()
    # 生成差异对比
    differ = difflib.Differ()
    diff = list(differ.compare(lines1, lines2))
    # 打印差异
    for line in diff:
        if line.startswith('+') or line.startswith('-') or line.startswith('?'):
            print(line.rstrip())
# 生成HTML格式的对比报告
def create_html_diff(file1_path, file2_path, output_html):
    with open(file1_path, 'r', encoding='utf-8') as f1:
        lines1 = f1.readlines()
    with open(file2_path, 'r', encoding='utf-8') as f2:
        lines2 = f2.readlines()
    html_diff = difflib.HtmlDiff().make_file(lines1, lines2)
    with open(output_html, 'w', encoding='utf-8') as f:
        f.write(html_diff)
# 使用示例
compare_files('data1.csv', 'data2.csv')
create_html_diff('data1.csv', 'data2.csv', '对比报告.html')

Python - 大数据集高效比对

import pandas as pd
# 对比两个CSV文件
def compare_csv(file1, file2, key_column='id'):
    df1 = pd.read_csv(file1)
    df2 = pd.read_csv(file2)
    # 找出不同
    merged = df1.merge(df2, on=key_column, suffixes=('_1', '_2'), how='outer', indicator=True)
    # 只在一个文件中存在的记录
    only_in_1 = merged[merged['_merge'] == 'left_only']
    only_in_2 = merged[merged['_merge'] == 'right_only']
    # 共同记录中数据不同的部分
    common = merged[merged['_merge'] == 'both']
    # 比较各列差异
    cols = [c for c in df1.columns if c != key_column]
    differences = []
    for col in cols:
        diff_mask = common[f'{col}_1'] != common[f'{col}_2']
        if diff_mask.any():
            diff_records = common[diff_mask][[key_column, f'{col}_1', f'{col}_2']]
            differences.append({f'{col}列差异': diff_records})
    return {
        '仅在文件1中': only_in_1,
        '仅在文件2中': only_in_2,
        '数据差异': differences
    }
# 使用示例
result = compare_csv('old_data.csv', 'new_data.csv')
print(result)

Shell脚本 - 文件比较

#!/bin/bash
# 文件比较脚本
# 使用diff命令
diff file1.txt file2.txt
# 只显示不同的行
diff --suppress-common-lines file1.txt file2.txt
# 将差异输出到文件
diff -u file1.txt file2.txt > diff_output.patch
# 比较并忽略空白字符
diff -w file1.txt file2.txt
# 递归比较目录
diff -r directory1/ directory2/
# 使用comm命令（需要排序）
sort file1.txt > sorted1.txt
sort file2.txt > sorted2.txt
comm -12 sorted1.txt sorted2.txt  # 共同的行
# 检查文件是否相同
if cmp -s file1.txt file2.txt; then
    echo "文件完全相同"
else
    echo "文件不同"
fi

JSON数据比对

import json
def compare_json(json1_path, json2_path):
    with open(json1_path, 'r') as f1, open(json2_path, 'r') as f2:
        data1 = json.load(f1)
        data2 = json.load(f2)
    def deep_compare(obj1, obj2, path=""):
        differences = []
        if type(obj1) != type(obj2):
            return [f"{path}: 类型不同 ({type(obj1).__name__} vs {type(obj2).__name__})"]
        if isinstance(obj1, dict):
            all_keys = set(list(obj1.keys()) + list(obj2.keys()))
            for key in all_keys:
                if key not in obj1:
                    differences.append(f"{path}.{key}: 仅在文件2中存在")
                elif key not in obj2:
                    differences.append(f"{path}.{key}: 仅在文件1中存在")
                else:
                    differences.extend(deep_compare(obj1[key], obj2[key], f"{path}.{key}"))
        elif isinstance(obj1, list):
            if len(obj1) != len(obj2):
                differences.append(f"{path}: 列表长度不同 ({len(obj1)} vs {len(obj2)})")
            else:
                for i, (item1, item2) in enumerate(zip(obj1, obj2)):
                    differences.extend(deep_compare(item1, item2, f"{path}[{i}]"))
        elif obj1 != obj2:
            differences.append(f"{path}: {obj1} != {obj2}")
        return differences
    return deep_compare(data1, data2)
# 使用示例
diffs = compare_json('config1.json', 'config2.json')
for diff in diffs:
    print(diff)

实时监控文件变化

import hashlib
import os
import time
def monitor_file_changes(file_path, interval=1):
    """监控文件变化"""
    last_hash = None
    while True:
        with open(file_path, 'rb') as f:
            current_hash = hashlib.md5(f.read()).hexdigest()
        if last_hash and current_hash != last_hash:
            print(f"文件已修改: {time.ctime()}")
            # 可以在这里触发比对操作
        last_hash = current_hash
        time.sleep(interval)
# 使用示例
# monitor_file_changes('data.txt')

性能优化建议

# 大文件分块读取比对
def compare_large_files(file1, file2, chunk_size=8192):
    with open(file1, 'rb') as f1, open(file2, 'rb') as f2:
        while True:
            chunk1 = f1.read(chunk_size)
            chunk2 = f2.read(chunk_size)
            if chunk1 != chunk2:
                return False
            if not chunk1:  # 文件结束
                break
    return True
# 使用哈希快速比对
def quick_compare(file1, file2):
    import hashlib
    def get_md5(filename):
        hash_md5 = hashlib.md5()
        with open(filename, "rb") as f:
            for chunk in iter(lambda: f.read(4096), b""):
                hash_md5.update(chunk)
        return hash_md5.hexdigest()
    return get_md5(file1) == get_md5(file2)

使用建议

1. 小文件：直接读取所有内容比对
2. 大文件：分块读取或用哈希快速比对
3. 结构化数据：使用pandas或json库
4. 需要详细报告：使用difflib生成HTML报告
5. 版本控制：使用Git diff

选择合适的比对方法取决于你的具体需求和数据规模。