本文目录导读：
我来介绍几种常用的数据比对脚本方法,涵盖不同场景：
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)
使用建议
- 小文件：直接读取所有内容比对
- 大文件：分块读取或用哈希快速比对
- 结构化数据：使用pandas或json库
- 需要详细报告：使用difflib生成HTML报告
- 版本控制：使用Git diff
选择合适的比对方法取决于你的具体需求和数据规模。
