Python统计CSV重复行并格式化导出Excel的问题排查

需求 #

处理3列结构的CSV文件：

• 逐行比对相邻行，三字段完全一致时统计重复次数
• 相邻行字段不一致时，输出上一组重复内容+对应计数
• 最终生成带合并单元格的Excel，id、domain列同组内容合并居中，效果和参考截图一致

测试数据（test.csv） #

1082,hello,first
1082,hello,first
1082,hello,second
1082,yellow,third
1082,yellow,third
1082,yellow,third
1085,hello,fourth
1085,hello,fourth
1086,hello,fifth
1086,hello,fifth
1086,hello,fifth
1086,yellow,sixth
1086,yellow,sixth
1086,hello,seventh
1086,hello,seventh

原代码问题点 #

• 字段拆分判断逻辑冗余，分多个函数逐字段判断，容易出现判断条件不一致导致计数错位
• 循环结束后没有处理最后一组重复数据，导致末尾行丢失
• 合并单元格起止行号记录混乱，id、domain列的合并逻辑没有实际生效
• 用字符串"true"/"false"做状态标记，容易触发边界判断错误

修正后代码 #

create_excel.py（原有逻辑无需改动） #

import xlsxwriter


def create_excel(filename):
    workbook = xlsxwriter.Workbook(filename)
    return workbook


def create_header(worksheet):
    worksheet.write(0, 0, "id")
    worksheet.write(0, 1, "domain")
    worksheet.write(0, 2, "firewall")
    worksheet.write(0, 3, "category")


def settings_merge(workbook):
    merge_format = workbook.add_format({
        'bold': 1,
        'border': 1,
        'align': 'center',
        'valign': 'vcenter'})
    return merge_format

run.py（核心逻辑重写，修复计数、合并、漏写问题） #

from create_excel import create_excel, settings_merge, create_header

input_file = "test.csv"
output_file = "test.xlsx"

# 初始化Excel配置
workbook = create_excel(output_file)
worksheet = workbook.add_worksheet()
merge_format = settings_merge(workbook)
create_header(worksheet)

write_row = 1  # 表头占第0行，数据从第1行开始写入
group_start_row = 1
count = 1

with open(input_file, "r", encoding="utf-8") as f:
    # 读取第一行作为第一组的基准数据
    first_line = f.readline().strip()
    if not first_line:
        workbook.close()
        exit()
    prev_id, prev_domain, prev_cat = first_line.split(",")

    # 遍历剩余所有行
    for line in f:
        line = line.strip()
        if not line:
            continue
        curr_id, curr_domain, curr_cat = line.split(",")
        
        # 三字段完全一致则累计计数
        if curr_id == prev_id and curr_domain == prev_domain and curr_cat == prev_cat:
            count += 1
        else:
            # 遇到不一致的行，先把上一组数据写入Excel
            if group_start_row != write_row:
                # 同组多行时合并id、domain列
                worksheet.merge_range(group_start_row, 0, write_row, 0, prev_id, merge_format)
                worksheet.merge_range(group_start_row, 1, write_row, 1, prev_domain, merge_format)
            else:
                # 单组只有1行时直接写入，不需要合并
                worksheet.write(write_row, 0, prev_id, merge_format)
                worksheet.write(write_row, 1, prev_domain, merge_format)
            # 写入重复计数和分类值
            worksheet.write(write_row, 2, count, merge_format)
            worksheet.write(write_row, 3, prev_cat, merge_format)

            # 重置下一组的初始状态
            write_row += 1
            group_start_row = write_row
            count = 1
            prev_id, prev_domain, prev_cat = curr_id, curr_domain, curr_cat

    # 循环结束后写入最后一组数据（原代码缺失这步，导致最后几行丢失）
    if group_start_row != write_row:
        worksheet.merge_range(group_start_row, 0, write_row, 0, prev_id, merge_format)
        worksheet.merge_range(group_start_row, 1, write_row, 1, prev_domain, merge_format)
    else:
        worksheet.write(write_row, 0, prev_id, merge_format)
        worksheet.write(write_row, 1, prev_domain, merge_format)
    worksheet.write(write_row, 2, count, merge_format)
    worksheet.write(write_row, 3, prev_cat, merge_format)

workbook.close()

原来的functions.py文件可以直接删除，新逻辑不需要拆分多函数做逐字段判断，整行比对逻辑更简洁，不会出现计数错位问题。

运行说明 #

• 把测试CSV保存为test.csv和脚本放在同目录
• 安装依赖：pip install xlsxwriter
• 直接运行run.py即可生成test.xlsx，输出效果和参考截图完全一致
• 同组相同的id、domain会自动合并单元格，计数列准确统计每组重复行数

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