malue-ai Excel 格式修复
自动检测并修复 Excel/CSV 常见格式问题。
使用场景
- 用户说「这个表格打开乱码了」「帮我修一下这个 Excel」
- Excel 分析前预处理（自动清洗）
- 从外部导入的数据格式不规范
常见问题与修复
1. 编码乱码
import pandas as pd
import chardet
def fix_encoding(file_path):
"""检测并修复 CSV 编码"""
with open(file_path, 'rb') as f:
raw = f.read(10000)
detected = chardet.detect(raw)
encoding = detected['encoding']
# 尝试用检测到的编码读取
df = pd.read_csv(file_path, encoding=encoding)
# 保存为 UTF-8
output = file_path.replace('.csv', '_fixed.csv')
df.to_csv(output, encoding='utf-8-sig', index=False)
return output, encoding
2. 合并单元格拆分
from openpyxl import load_workbook
def unmerge_cells(file_path):
"""拆分合并单元格，向下填充值"""
wb = load_workbook(file_path)
ws = wb.active
# 记录合并区域
merged_ranges = list(ws.merged_cells.ranges)
for merged in merged_ranges:
# 获取合并区域左上角的值
top_left_value = ws.cell(merged.min_row, merged.min_col).value
# 取消合并
ws.unmerge_cells(str(merged))
# 向下填充
for row in range(merged.min_row, merged.max_row + 1):
for col in range(merged.min_col, merged.max_col + 1):
ws.cell(row, col, top_left_value)
output = file_path.replace('.xlsx', '_unmerged.xlsx')
wb.save(output)
return output, len(merged_ranges)
3. 重复表头检测
def fix_duplicate_headers(df):
"""检测并修复重复表头行"""
# 检查前几行是否与列名重复
header_like_rows = []
for i, row in df.head(5).iterrows():
match_count = sum(1 for v in row.values if str(v) in df.columns.tolist())
if match_count > len(df.columns) * 0.5:
header_like_rows.append(i)
if header_like_rows:
df = df.drop(header_like_rows).reset_index(drop=True)
return df, len(header_like_rows)
4. 数据类型不一致
def fix_column_types(df):
"""检测并修复列内数据类型不一致"""
fixes = []
for col in df.columns:
# 尝试转为数字
numeric = pd.to_numeric(df[col], errors='coerce')
non_null_ratio = numeric.notna().sum() / len(df)
if non_null_ratio > 0.8 and df[col].dtype == object:
# 80% 以上是数字，可能是数字列混入了文本
bad_rows = df[numeric.isna() & df[col].notna()]
fixes.append(f"列 '{col}': {len(bad_rows)} 行非数字值")
return fixes
5. 一键修复流程
def auto_fix(file_path):
"""自动检测并修复所有常见问题"""
report = []
# 1. 读取文件
if file_path.endswith('.csv'):
# 编码修复
...
else:
# 合并单元格修复
...
# 2. 读取为 DataFrame
df = pd.read_excel(file_path) if file_path.endswith('.xlsx') else pd.read_csv(file_path)
# 3. 重复表头
df, dup_count = fix_duplicate_headers(df)
if dup_count:
report.append(f"移除 {dup_count} 行重复表头")
# 4. 数据类型
type_issues = fix_column_types(df)
report.extend(type_issues)
# 5. 空行空列
before = len(df)
df = df.dropna(how='all')
df = df.loc[:, ~df.columns.str.contains('^Unnamed')]
after = len(df)
if before != after:
report.append(f"移除 {before - after} 行空行")
return df, report
6. 分类列文本标准化（必须执行）
def standardize_categories(df, columns):
"""对分类列做文本标准化，合并近似重复值"""
for col in columns:
if col not in df.columns:
continue
# 去首尾空格
df[col] = df[col].astype(str).str.strip()
# 检测近似重复（如 "华东" vs "华东地区"）
unique_vals = df[col].unique()
for v1 in unique_vals:
for v2 in unique_vals:
if v1 != v2 and (v1 in v2 or v2 in v1):
# 合并为较短的值（更通用）
shorter = v1 if len(v1) <= len(v2) else v2
longer = v2 if shorter == v1 else v1
df[col] = df[col].replace(longer, shorter)
print(f" 合并: '{longer}' → '{shorter}'")
return df
输出规范
- 修复前先展示检测到的问题清单
- 重大修改（如删除行/列）需要 HITL 确认
- 修复后保存为新文件（不覆盖原文件）
- 展示修复报告：修了什么、修了多少
- 必须打印清洗后行数：让用户（和校验系统）确认没有误删数据
- 分类列必须标准化：合并近似重复值（如"华东"/"华东地区"）