通用Excel转SQL导入：合并单元格表头及无效数据处理

问题1：合并单元格表头的精准填充 #

DataFrame本身没有merged_cells属性，pandas读取时会丢失合并单元格的结构信息，直接用fillna会误填充数据行的有效NaN。解决方案是直接操作Excel工作簿获取合并范围，仅针对性填充表头行的NaN：

实现代码 #

from openpyxl import load_workbook
import pandas as pd

def fix_merged_header(file_path, sheet_name=0, header_candidate_rows=10):
    # 读取Excel工作簿获取合并单元格原始信息
    wb = load_workbook(file_path, read_only=True)
    ws = wb[sheet_name] if isinstance(sheet_name, str) else wb.worksheets[sheet_name]
    
    # 构建合并单元格坐标与值的映射表
    merged_map = {}
    for merged_range in ws.merged_cells.ranges:
        top_left_val = ws.cell(row=merged_range.min_row, column=merged_range.min_col).value
        # 遍历合并范围的所有单元格，记录对应值
        for row in range(merged_range.min_row, merged_range.max_row + 1):
            for col in range(merged_range.min_col, merged_range.max_col + 1):
                merged_map[(row, col)] = top_left_val
    
    # 读取前N行作为表头候选区域
    df_candidate = pd.read_excel(file_path, sheet_name=sheet_name, nrows=header_candidate_rows, header=None)
    
    # 仅填充候选表头行中的合并单元格NaN
    for row_idx in df_candidate.index:
        for col_idx in df_candidate.columns:
            # 转换为openpyxl的1-based坐标
            excel_row = row_idx + 1
            excel_col = col_idx + 1
            if (excel_row, excel_col) in merged_map:
                df_candidate.iloc[row_idx, col_idx] = merged_map[(excel_row, excel_col)]
    
    return df_candidate

问题2：自动识别表头行（跳过无效前置数据） #

通过非空值数量+字符串类型占比的综合得分，自动定位最可能的表头行，无需预先知晓位置：

实现代码 #

def detect_header_row(df_candidate):
    # 计算每行非空值数量（表头非空值通常最多）
    non_null_count = df_candidate.notna().sum(axis=1)
    # 计算每行字符串类型占比（表头多为文本，数据行可能含数字）
    str_ratio = df_candidate.apply(
        lambda row: sum(isinstance(val, str) for val in row if pd.notna(val)) / max(1, row.notna().sum()),
        axis=1
    )
    # 综合得分：非空数权重0.7，字符串占比权重0.3（可按需调整）
    score = non_null_count * 0.7 + str_ratio * 0.3
    # 取得分最高的行作为表头行（相同得分取最上方）
    return score.idxmax()

完整整合流程（导入SQL） #

将上述函数结合，完成从Excel到SQL的通用导入，无需硬编码任何列名或坐标：

from sqlalchemy import create_engine

def excel_to_sql(file_path, db_url, table_name, sheet_name=0):
    # 1. 修复合并表头候选行
    df_candidate = fix_merged_header(file_path, sheet_name)
    # 2. 自动检测表头行位置
    header_idx = detect_header_row(df_candidate)
    # 3. 读取完整数据，跳过头前无效行
    df_full = pd.read_excel(file_path, sheet_name=sheet_name, skiprows=header_idx + 1, header=None)
    # 4. 设置正确表头
    df_full.columns = df_candidate.iloc[header_idx].tolist()
    # 5. 清理无效行（全空行）
    df_full = df_full.dropna(how='all').reset_index(drop=True)
    # 6. 导入SQL（自动创建表，追加数据）
    engine = create_engine(db_url)
    df_full.to_sql(table_name, engine, if_exists='append', index=False)

使用示例 #

# MySQL示例，其他数据库替换对应的db_url即可
db_url = "mysql+pymysql://username:password@localhost:3306/your_database"
excel_to_sql("your_excel_file.xlsx", db_url, "target_table")

注意事项 #

• header_candidate_rows可调整为20或更大，确保覆盖所有可能的表头行位置
• 得分权重可根据实际文件调整：如果表头含数字（如年份），可降低字符串占比的权重
• 支持多层表头：可扩展detect_header_row函数，检测连续的高得分行作为多层表头
• 导入SQL时，pandas会自动匹配列类型，无需硬编码列信息

内容的提问来源于stack exchange，提问作者Cenderze