基于Python 3.11的动态CSV表头识别问题求助

一、XLSX合并单元格动态识别与处理 #

用openpyxl实现合并单元格的自动识别与值填充，确保转换为CSV后表头识别逻辑能正常工作：

from openpyxl import load_workbook

def handle_merged_cells(file_path):
    wb = load_workbook(file_path)
    ws = wb.active
    
    # 获取所有合并单元格范围，倒序处理避免覆盖
    merged_ranges = list(ws.merged_cells.ranges)[::-1]
    
    for merged_range in merged_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):
                ws.cell(row=row, column=col).value = top_left_val
        # 取消合并
        ws.unmerge_cells(str(merged_range))
    
    # 转换为CSV格式行数据
    csv_rows = []
    for row in ws.iter_rows(values_only=True):
        csv_rows.append(','.join([str(cell) if cell is not None else '' for cell in row]))
    
    return csv_rows

二、表头识别逻辑修复（解决数据行混入表头问题） #

核心思路是先定位含date的候选行，再通过字段数匹配和数据类型验证排除数据行，同时支持多行表头合并：

def identify_header(csv_rows):
    header_index = -1
    date_candidates = []
    
    # 遍历所有行，收集含date（不区分大小写）的行索引
    for idx, row in enumerate(csv_rows):
        if not row.strip():
            continue
        cells = [cell.strip().lower() for cell in row.split(',') if cell.strip()]
        if 'date' in cells:
            date_candidates.append(idx)
    
    if not date_candidates:
        return -1  # 无表头标记
    
    # 验证候选行是否为有效表头
    for candidate_idx in date_candidates:
        candidate_row = [cell.strip() for cell in csv_rows[candidate_idx].split(',')]
        # 检查后续3行的字段数是否与候选行一致
        match_count = 0
        for check_idx in range(candidate_idx + 1, min(candidate_idx + 4, len(csv_rows))):
            check_row = [cell.strip() for cell in csv_rows[check_idx].split(',') if cell.strip()]
            if len(check_row) == len(candidate_row):
                match_count += 1
        # 表头行多为纯字符串，数据行可能含数值/日期格式
        is_string_header = all(isinstance(cell, str) or (cell and cell.isalpha()) for cell in candidate_row)
        if match_count >= 2 and is_string_header:
            header_index = candidate_idx
            break
    
    # 处理多行表头：合并连续的表头行
    if header_index != -1 and header_index + 1 < len(csv_rows):
        next_row = [cell.strip() for cell in csv_rows[header_index + 1].split(',')]
        if len(next_row) == len(candidate_row) and all(isinstance(cell, str) or (cell and cell.isalpha()) for cell in next_row):
            merged_header = [f"{h1}_{h2}" if h2 else h1 for h1, h2 in zip(candidate_row, next_row)]
            csv_rows[header_index] = ','.join(merged_header)
            del csv_rows[header_index + 1]
    
    return header_index

三、全场景整合脚本框架 #

import os

def convert_to_csv(file_path):
    ext = os.path.splitext(file_path)[1].lower()
    csv_rows = []
    
    if ext in ['.csv', '.txt']:
        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            csv_rows = [line.strip() for line in f if line.strip()]
        # 处理|分隔的伪CSV
        if csv_rows and '|' in csv_rows[0] and ',' not in csv_rows[0]:
            csv_rows = [row.replace('|', ',') for row in csv_rows]
    
    elif ext == '.xlsx':
        csv_rows = handle_merged_cells(file_path)
    
    return csv_rows

def process_file(file_path):
    csv_rows = convert_to_csv(file_path)
    if not csv_rows:
        print("文件无有效内容")
        return
    
    header_idx = identify_header(csv_rows)
    
    # 无表头场景处理：自动生成含date的默认表头
    if header_idx == -1:
        data_cols = len(csv_rows[0].split(','))
        default_header = [f"col_{i}" for i in range(data_cols)]
        # 尝试识别日期列并替换为date
        from datetime import datetime
        for idx, cell in enumerate(csv_rows[0].split(',')):
            try:
                datetime.strptime(cell.strip(), '%Y-%m-%d')
                default_header[idx] = 'date'
                break
            except:
                continue
        csv_rows.insert(0, ','.join(default_header))
        header_idx = 0
    
    # 输出识别结果
    header = csv_rows[header_idx].split(',')
    data_rows = len(csv_rows) - header_idx - 1
    print(f"表头位于第{header_idx+1}行：{header}")
    print(f"有效数据行数：{data_rows}")

# 使用示例
if __name__ == "__main__":
    process_file("your_test_file.csv")
    process_file("your_test_file.xlsx")

四、场景适配说明 #

• 单行表头：通过字段数匹配+字符串验证，避免数据行误判为表头
• 多行表头：自动合并连续的纯字符串表头行
• 无表头：自动生成含date的默认表头（自动识别日期列）
• 表头前含元数据：自动跳过空行，仅识别含date的有效行
• 伪CSV：自动将|分隔转换为逗号分隔
• XLSX合并单元格：先填充合并值再转换为CSV，保证表头完整性

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