import osimport pandas as pdimport redef clean_data(df, remove_duplicates=True, fillna_value="", remove_empty_rows=True, strip_spaces=True, fix_numbers=True, remove_special_chars=False): """ 执行数据清洗 """ original_rows = len(df) report = [] # 1. 删除空行 if remove_empty_rows: before = len(df) df = df.dropna(how='all') after = len(df) if before - after > 0: report.append(f"删除空行: {before - after} 行") # 2. 删除重复行 if remove_duplicates: before = len(df) df = df.drop_duplicates() after = len(df) if before - after > 0: report.append(f"删除重复行: {before - after} 行") # 3. 填充空值 if fillna_value != "": df = df.fillna(fillna_value) report.append(f"空值填充为: {fillna_value}") # 4. 去除首尾空格 if strip_spaces: for col in df.select_dtypes(include=['object']).columns: before = df[col].astype(str).str.strip() df[col] = before report.append("已去除所有文本列的首尾空格") # 5. 修正数字格式 if fix_numbers: for col in df.select_dtypes(include=['object']).columns: try: df[col] = pd.to_numeric(df[col], errors='ignore') except: pass report.append("已尝试修正数字格式") # 6. 移除特殊字符 if remove_special_chars: for col in df.select_dtypes(include=['object']).columns: df[col] = df[col].astype(str).apply(lambda x: re.sub(r'[^\w\s\u4e00-\u9fa5]', '', x)) report.append("已移除特殊字符") report.append(f"清洗完成!{original_rows} 行 → {len(df)} 行") return df, reportdef batch_clean_excel(folder_path, output_folder="清洗后", remove_duplicates=True, fillna_value="", remove_empty_rows=True, strip_spaces=True, fix_numbers=True, remove_special_chars=False): """ 批量清洗文件夹中的Excel文件 """ excel_files = [f for f in os.listdir(folder_path) if f.lower().endswith(('.xlsx', '.xls'))] if not excel_files: print("❌ 未找到Excel文件") return print(f"📂 找到 {len(excel_files)} 个Excel文件") output_path = os.path.join(folder_path, output_folder) os.makedirs(output_path, exist_ok=True) print("\n📋 清洗规则:") if remove_duplicates: print(" ✅ 删除重复行") if remove_empty_rows: print(" ✅ 删除空行") if fillna_value: print(f" ✅ 空值填充为: {fillna_value}") if strip_spaces: print(" ✅ 去除首尾空格") if fix_numbers: print(" ✅ 修正数字格式") if remove_special_chars: print(" ✅ 移除特殊字符") print() all_reports = [] success_count = 0 for file in excel_files: file_path = os.path.join(folder_path, file) print(f" 📄 {file}") try: df = pd.read_excel(file_path) original_rows = len(df) cleaned_df, report = clean_data(df, remove_duplicates, fillna_value, remove_empty_rows, strip_spaces, fix_numbers, remove_special_chars) output_file = os.path.join(output_path, file) cleaned_df.to_excel(output_file, index=False) success_count += 1 for r in report: print(f" {r}") all_reports.append({'文件名': file, '原始行数': original_rows, '清洗后行数': len(cleaned_df)}) except Exception as e: print(f" ❌ 失败: {str(e)[:40]}") print(f"\n✅ 完成!成功清洗 {success_count} 个文件") print(f"📁 保存在: {output_path}") if all_reports: summary_df = pd.DataFrame(all_reports) summary_path = os.path.join(output_path, "清洗报告.xlsx") summary_df.to_excel(summary_path, index=False) print(f"📊 清洗报告已保存: {summary_path}")def quick_clean_current(folder_path): """快速清洗当前目录下的所有Excel""" print("⚡ 快速清洗模式") print("执行规则: 删除空行 + 删除重复 + 去除空格") print() excel_files = [f for f in os.listdir(folder_path) if f.lower().endswith(('.xlsx', '.xls'))] output_path = os.path.join(folder_path, "清洗后") os.makedirs(output_path, exist_ok=True) success_count = 0 for file in excel_files: file_path = os.path.join(folder_path, file) print(f" 📄 {file}") try: df = pd.read_excel(file_path) before = len(df) df = df.dropna(how='all') df = df.drop_duplicates() for col in df.select_dtypes(include=['object']).columns: df[col] = df[col].astype(str).str.strip() output_file = os.path.join(output_path, file) df.to_excel(output_file, index=False) success_count += 1 print(f" {before} 行 → {len(df)} 行") except Exception as e: print(f" ❌ 失败") print(f"\n✅ 完成!清洗 {success_count} 个文件")if __name__ == "__main__": print("=" * 50) print("🧹 批量数据清洗工具") print("=" * 50) print("自动清理:空行、重复值、空格、格式错误") print() path = input("请输入Excel文件夹路径: ").strip() if path.startswith('"') and path.endswith('"'): path = path[1:-1] path = path.replace('\\', '/') if not os.path.exists(path): print("❌ 路径不存在") input("按回车键退出...") exit() print("\n1. 完整清洗(自定义规则)") print("2. 快速清洗(一键清理)") choice = input("请选择(1或2,默认1): ").strip() or "1" if choice == "2": quick_clean_current(path) else: print("\n请选择清洗规则:") remove_empty = input("删除空行?(y/n,默认y): ").strip().lower() != 'n' remove_dupes = input("删除重复行?(y/n,默认y): ").strip().lower() != 'n' strip_spaces = input("去除首尾空格?(y/n,默认y): ").strip().lower() != 'n' fix_nums = input("修正数字格式?(y/n,默认y): ").strip().lower() != 'n' fillna = input("填充空值(直接回车跳过,输入文字则填充所有空值): ").strip() remove_special = input("移除特殊字符?(y/n,默认n): ").strip().lower() == 'y' batch_clean_excel(path, remove_duplicates=remove_dupes, fillna_value=fillna, remove_empty_rows=remove_empty, strip_spaces=strip_spaces, fix_numbers=fix_nums, remove_special_chars=remove_special) input("\n按回车键退出...")