本文目录导读：
实现本地数据去重有多种方式，我给你一个通用的Python脚本方案,支持常见的数据格式：
文本文件去重（最常用）
def deduplicate_text(input_file, output_file):
"""文本文件按行去重，保留首次出现的顺序"""
seen = set()
with open(input_file, 'r', encoding='utf-8') as f_in, \
open(output_file, 'w', encoding='utf-8') as f_out:
for line in f_in:
line_stripped = line.strip()
if line_stripped and line_stripped not in seen:
seen.add(line_stripped)
f_out.write(line)
print(f"去重完成！原始行数: {len(seen)}, 保留行数: {len(seen)}")
# 使用示例
deduplicate_text('input.txt', 'output.txt')
CSV/Excel文件去重
import pandas as pd
def deduplicate_csv(input_file, output_file, key_columns=None):
"""
CSV/Excel去重
key_columns: 用于判断重复的列名列表，None表示整行去重
"""
# 读取文件（支持CSV和Excel）
if input_file.endswith('.csv'):
df = pd.read_csv(input_file)
else:
df = pd.read_excel(input_file)
# 去重
if key_columns:
df_dedup = df.drop_duplicates(subset=key_columns, keep='first')
else:
df_dedup = df.drop_duplicates(keep='first')
# 保存结果
if output_file.endswith('.csv'):
df_dedup.to_csv(output_file, index=False)
else:
df_dedup.to_excel(output_file, index=False)
print(f"原始数据: {len(df)} 行")
print(f"去重后: {len(df_dedup)} 行")
print(f"删除重复: {len(df) - len(df_dedup)} 行")
# 使用示例：按"email"列去重
deduplicate_csv('data.csv', 'dedup_data.csv', key_columns=['email'])
多字段组合去重
import pandas as pd
def deduplicate_excel_advanced(input_file, output_file,
primary_key=None,
fuzzy_columns=None):
"""
高级去重：支持精确去重和模糊去重
"""
df = pd.read_excel(input_file)
# 1. 精确去重
if primary_key:
df = df.drop_duplicates(subset=primary_key, keep='first')
# 2. 模糊去重（忽略大小写、空格）
if fuzzy_columns:
for col in fuzzy_columns:
df[col + '_normalized'] = df[col].str.lower().str.strip()
df = df.drop_duplicates(
subset=[f'{col}_normalized' for col in fuzzy_columns],
keep='first'
)
# 删除辅助列
df = df.drop([f'{col}_normalized' for col in fuzzy_columns], axis=1)
# 保存
df.to_excel(output_file, index=False)
print(f"去重完成，最终数据: {len(df)} 行")
# 使用示例
deduplicate_excel_advanced(
'contacts.xlsx',
'unique_contacts.xlsx',
primary_key='email',
fuzzy_columns=['name', 'phone']
)
JSON数据去重
import json
def deduplicate_json(input_file, output_file, unique_fields=None):
"""
JSON数组去重
unique_fields: 用于判断重复的字段列表，None表示整个对象相同才去重
"""
with open(input_file, 'r', encoding='utf-8') as f:
data = json.load(f)
if isinstance(data, list):
unique_data = []
seen = set()
for item in data:
if unique_fields:
# 根据指定字段生成唯一标识
key = tuple(str(item.get(field, '')) for field in unique_fields)
else:
# 整个对象序列化作为唯一标识
key = json.dumps(item, sort_keys=True)
if key not in seen:
seen.add(key)
unique_data.append(item)
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(unique_data, f, ensure_ascii=False, indent=2)
print(f"原始数据: {len(data)} 条，去重后: {len(unique_data)} 条")
else:
print("输入文件格式错误：应为JSON数组")
# 使用示例
deduplicate_json('data.json', 'unique_data.json', unique_fields=['id', 'email'])
命令行去重工具
创建一个通用的命令行工具：
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import argparse
import pandas as pd
from pathlib import Path
def main():
parser = argparse.ArgumentParser(description='本地数据去重工具')
parser.add_argument('input', help='输入文件路径')
parser.add_argument('-o', '--output', default='dedup_output.csv', help='输出文件路径')
parser.add_argument('-k', '--key', nargs='+', help='用于去重的列名（多个用空格分隔）')
parser.add_argument('-t', '--type', choices=['auto', 'csv', 'excel'], default='auto',
help='文件类型，默认自动识别')
parser.add_argument('-f', '--first', action='store_true', help='保留第一条（默认）')
parser.add_argument('-l', '--last', action='store_true', help='保留最后一条')
args = parser.parse_args()
# 自动识别文件类型
file_type = args.type
if file_type == 'auto':
suffix = Path(args.input).suffix.lower()
if suffix in ['.xlsx', '.xls']:
file_type = 'excel'
else:
file_type = 'csv'
# 读取数据
if file_type == 'excel':
df = pd.read_excel(args.input)
else:
df = pd.read_csv(args.input)
# 去重
keep = 'last' if args.last else 'first'
if args.key:
df_dedup = df.drop_duplicates(subset=args.key, keep=keep)
else:
df_dedup = df.drop_duplicates(keep=keep)
# 保存结果
if args.output.endswith(('.xlsx', '.xls')):
df_dedup.to_excel(args.output, index=False)
else:
df_dedup.to_csv(args.output, index=False)
print(f"✓ 去重完成！")
print(f" 原始数据: {len(df)} 行")
print(f" 去重后: {len(df_dedup)} 行")
print(f" 删除重复: {len(df) - len(df_dedup)} 行")
if __name__ == '__main__':
main()
使用示例
# 1. 基本去重（整行去重） python dedup.py input.csv -o output.csv # 2. 按指定列去重 python dedup.py input.csv -k email -o output.csv # 3. 多列组合去重 python dedup.py input.xlsx -k name phone -o output.xlsx # 4. 保留最后一条 python dedup.py input.csv -k id -l -o output.csv
安装依赖
pip install pandas openpyxl
注意事项
- 内存管理：对于超大文件，使用Pandas可能会占用大量内存，可以考虑分块读取
- 数据清洗：去重前最好先进行数据清洗，如去除空白、统一大小写等
- 保留策略：考虑是保留第一次出现的记录还是最后一次
- 备份原文件：操作前建议先备份原始数据
需要根据你的具体数据格式和需求选择合适的方案，有具体的场景可以告诉我,我来帮你定制！
