excel-fixer（excel 修复程序）

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 确认
• 修复后保存为新文件（不覆盖原文件）
• 展示修复报告：修了什么、修了多少
• 必须打印清洗后行数：让用户（和校验系统）确认没有误删数据
• 分类列必须标准化：合并近似重复值（如"华东"/"华东地区"）