1. 基础文件读写
首先，了解如何读取和写入文件是基础。
# 读取文件
with open('file1.txt', 'r') as file1:
data1 = file1.readlines()
# 写入文件
with open('merged.txt', 'w') as merged_file:
for line in data1:
merged_file.write(line)2. 文件内容对比
使用difflib库来对比两个文件的差异。
import difflib
with open('file1.txt', 'r') as file1, open('file2.txt', 'r') as file2:
diff = difflib.unified_diff(file1.readlines(), file2.readlines())
print('\n'.join(diff))3. 基于行的合并
当文件基于相同行结构合并时，可以直接遍历追加。
data = []
for filename in ['file1.txt', 'file2.txt']:
with open(filename, 'r') as file:
data.extend(file.readlines())
with open('merged.txt', 'w') as merged_file:
for line in data:
merged_file.write(line)4. 去重合并
利用集合去除重复行后合并。
unique_lines = set()
for filename in ['file1.txt', 'file2.txt']:
with open(filename, 'r') as file:
unique_lines.update(file.readlines())
with open('merged_unique.txt', 'w') as merged_file:
for line in sorted(unique_lines): # 排序确保一致的输出顺序
merged_file.write(line)5. CSV文件合并
对于CSV文件，可以使用pandas库。
import pandas as pd
df1 = pd.read_csv('file1.csv')
df2 = pd.read_csv('file2.csv')
# 假设合并依据为相同的列名
merged_df = pd.concat([df1, df2], ignore_index=True)
merged_df.to_csv('merged.csv', index=False)6. 按列合并CSV
特定列的合并，例如通过共同键连接。
merged_df = pd.merge(df1, df2, on='common_key', how='outer')
merged_df.to_csv('merged_by_key.csv', index=False)7. 大文件高效对比
对于大文件，逐行读取对比以节省内存。
with open('large_file1.txt', 'r') as f1, open('large_file2.txt', 'r') as f2:
for line1, line2 in zip(f1, f2):
if line1 != line2:
print("Difference found!")
break8. 文本文件的二进制对比
使用filecmp模块比较文件的二进制内容。
import filecmp
if filecmp.cmp('file1.txt', 'file2.txt'):
print("Files are identical.")
else:
print("Files differ.")9. 动态合并多个文件
使用循环动态合并多个文件路径列表中的文件。
file_paths = ['file{}.txt'.format(i) for i in range(1, 4)] # 假设有file1.txt到file3.txt
with open('merged_all.txt', 'w') as merged:
for path in file_paths:
with open(path, 'r') as file:
merged.write(file.read() + '\n') # 添加换行符区分不同文件的内容10. 高级合并策略：智能合并
如果合并依据更复杂，如按日期或ID排序合并，可以先对数据进行排序处理。
# 假设是CSV且按日期列排序合并
dfs = [pd.read_csv(f) for f in ['file1.csv', 'file2.csv']]
sorted_df = pd.concat(dfs).sort_values(by='date_column') # 假定'date_column'是日期列
sorted_df.to_csv('smart_merged.csv', index=False)进阶技巧和场景
11. 使用正则表达式进行复杂文本处理
在合并或对比前，可能需要对文件内容进行预处理，例如提取特定模式的数据。
import re
pattern = r'(\d{4}-\d{2}-\d{2})' # 假设提取日期模式
lines_with_dates = []
with open('source.txt', 'r') as file:
for line in file:
match = re.search(pattern, line)
if match:
lines_with_dates.append(match.group(0))
# 假设你想将提取的信息写入新文件
with open('dates_extracted.txt', 'w') as out_file:
for date in lines_with_dates:
out_file.write(date + '\n')12. 并行处理大文件对比
对于超大文件，可以利用多线程或多进程提高效率，但需注意文件访问冲突。
from multiprocessing import Pool
import os
def compare_lines(line1, line2):
return line1 == line2
if __name__ == "__main__":
with open('file1.txt', 'r') as f1, open('file2.txt', 'r') as f2:
lines_f1 = f1.readlines()
lines_f2 = f2.readlines()
with Pool(os.cpu_count()) as p: # 使用CPU核心数作为进程数
results = p.map(compare_lines, zip(lines_f1, lines_f2))
# results是一个布尔值列表，表示对应行是否相同13. 特殊格式文件的合并
例如XML文件，可以使用xml.etree.ElementTree进行解析合并。
import xml.etree.ElementTree as ET
root1 = ET.parse('file1.xml').getroot()
root2 = ET.parse('file2.xml').getroot()
for child in root2:
root1.append(child)
tree = ET.ElementTree(root1)
tree.write('merged.xml')14. 实时监控文件变化并合并
利用watchdog库监控文件变化，自动执行合并操作。
安装watchdog:
pip install watchdog示例脚本：
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time
class MyHandler(FileSystemEventHandler):
def on_modified(self, event):
if event.is_directory:
return
# 在这里实现你的文件合并逻辑
print(f'Event type: {event.event_type} path : {event.src_path}')
if __name__ == "__main__":
event_handler = MyHandler()
observer = Observer()
observer.schedule(event_handler, path='.', recursive=False)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()结语
通过这些高级策略和技巧，你可以更加灵活和高效地处理各种文件对比与合并的需求。
