Python CSV数据清洗与合并脚本
批量清洗CSV文件，去除空行、重复数据、统一格式，合并多个CSV，输出干净的数据集。
详细内容
#!/usr/bin/env python3
# Python CSV数据清洗与合并脚本
# 依赖: pip install pandas
# 使用方法: python3 csv_cleaner.py
import os
import sys
import pandas as pd
def clean_csv(filepath, output_path=None):
"""清洗单个CSV文件"""
print(f'\n处理文件: {filepath}')
# 读取CSV(自动尝试编码)
encodings = ['utf-8', 'gbk', 'gb2312', 'latin-1']
df = None
for enc in encodings:
try:
df = pd.read_csv(filepath, encoding=enc)
print(f' 编码: {enc}')
break
except:
continue
if df is None:
print(' ❌ 无法读取文件')
return None
original_count = len(df)
print(f' 原始行数: {original_count}')
print(f' 列名: {list(df.columns)}')
# 1. 去除完全空行
df = df.dropna(how='all')
print(f' 去除空行后: {len(df)}')
# 2. 去除重复行
df = df.drop_duplicates()
print(f' 去重后: {len(df)}')
# 3. 去除列名前后空格
df.columns = df.columns.str.strip()
# 4. 字符串列去除前后空格
for col in df.select_dtypes(include=['object']).columns:
df[col] = df[col].astype(str).str.strip()
# 5. 重置索引
df = df.reset_index(drop=True)
# 保存
if output_path is None:
base, ext = os.path.splitext(filepath)
output_path = f'{base}_cleaned{ext}'
df.to_csv(output_path, index=False, encoding='utf-8-sig')
print(f' ✅ 已保存: {output_path}')
print(f' 清洗后行数: {len(df)} (减少 {original_count - len(df)} 行)')
return df
def merge_csvs(csv_files, output_path='merged.csv'):
"""合并多个CSV文件"""
print(f'\n合并 {len(csv_files)} 个CSV文件...')
dfs = []
for f in csv_files:
df = clean_csv(f)
if df is not None:
dfs.append(df)
if not dfs:
print('没有可合并的文件')
return
merged = pd.concat(dfs, ignore_index=True)
merged = merged.drop_duplicates()
merged.to_csv(output_path, index=False, encoding='utf-8-sig')
print(f'\n合并完成！')
print(f' 总行数: {len(merged)}')
print(f' 总列数: {len(merged.columns)}')
print(f' 输出文件: {output_path}')
if __name__ == '__main__':
if len(sys.argv) < 2:
print('使用方法:')
print(' 清洗单个文件: python3 csv_cleaner.py data.csv')
print(' 合并目录所有CSV: python3 csv_cleaner.py ./csv_dir --merge')
sys.exit(1)
path = sys.argv[1]
merge = '--merge' in sys.argv
if os.path.isdir(path):
csv_files = [os.path.join(path, f) for f in os.listdir(path) if f.endswith('.csv')]
if merge:
merge_csvs(csv_files)
else:
for f in csv_files:
clean_csv(f)
else:
clean_csv(path)