openpyxl WriteOnlyWorksheet模式下按字母顺序排序工作表方法
问题根因
WriteOnlyWorksheet是openpyxl提供的低内存写入模式,设计上仅支持顺序追加创建工作表、追加写入行内容,不支持创建完成后调整工作表顺序。直接操作私有属性_sheets属于未公开的内部实现逻辑,在只写模式下不会触发工作表元数据的同步更新,因此排序无法生效。
无需重新打开生成后的xlsx文件调整顺序,也不用放弃只写模式的内存优势,两种可行方案如下:
方案1:两次遍历源文件(内存占用最低,推荐大文件场景使用)
核心逻辑是第一次遍历仅收集客户端元信息,不存储具体行内容,完成排序后第二次遍历按顺序创建工作表写入数据,全程内存占用和原只写逻辑几乎一致:
- 第一次遍历源文件:跳过无效行,识别所有客户端分组,记录每个客户端对应的清洗后工作表名、数据起始行号,收集完成后按工作表名做字母排序
- 初始化只写工作簿,按排序后的客户端顺序依次创建对应工作表,先写入公共表头
- 第二次遍历源文件:跳过无效行,判断当前行所属客户端,直接写入对应已创建好的工作表
该方案仅额外存储客户端名称、行号类的轻量元数据,即使是十万行以上的大文件也不会出现明显内存上涨。第一次遍历打开源文件时可加上read_only=True参数进一步降低内存占用。
方案2:单遍历+按客户端缓存数据(代码更简洁,适合中小文件场景)
如果单文件数据量没有大到内存吃紧,可以在遍历时用字典临时缓存每个客户端的行数据,遍历完成后排序再统一写入:
- 遍历源文件时跳过无效行,识别到新客户端就初始化字典键,后续同属该客户端的行直接追加到对应键的列表中
- 遍历完成后,把字典的键(即清洗后的工作表名)按字母顺序排序
- 初始化只写工作簿,按排序后的顺序依次创建工作表,把缓存的对应客户端行数据逐行写入即可
该方案会把全量有效数据临时存在内存中,客户端数量多、单客户端数据量大时内存占用会明显升高,适合数据量在几万行以内的场景。
核心代码修改示例(以方案1为例)
替换原split_workbook函数即可实现按字母序输出工作表:
def split_workbook(input_file, output_file): """ Split workbook each client into its own sheet, sorted alphabetically. """ workbook = None output_workbook = None try: logger.info(f"Loading workbook {input_file} for metadata collection") # 第一次遍历用只读模式打开,降低内存占用 workbook = load_workbook(input_file, read_only=True) data_sheet = workbook.active client_meta = [] # 存储格式:(清洗后sheet名, 客户端起始行号) rows = data_sheet.rows header = next(rows) # 第一次遍历:仅收集客户端元数据 for index, row in enumerate(rows, start=2): row_dimension = data_sheet.row_dimensions[index] # 跳过无效行逻辑与原逻辑保持一致 if skip_row(row, row_dimension): continue if is_client_row(row, row_dimension): sheet_title = clean_sheet_title(row[0].value) client_meta.append((sheet_title, index)) # 按工作表名称字母序排序 client_meta.sort(key=lambda x: x[0]) workbook.close() # 第二次遍历:按排序后顺序创建sheet写入数据 logger.info(f"Loading workbook {input_file} for data writing") workbook = load_workbook(input_file) data_sheet = workbook.active output_workbook = Workbook(write_only=True) # 删除只写模式默认创建的空工作表 if "Sheet" in output_workbook.sheetnames: del output_workbook["Sheet"] # 预创建所有排序后的工作表,建立表名到对象的映射 sheet_map = {} for title, _ in client_meta: new_sheet = output_workbook.create_sheet(title) # 复制列宽配置 for key, column_dimension in data_sheet.column_dimensions.items(): new_sheet.column_dimensions[key] = copy(column_dimension) new_sheet.column_dimensions[key].worksheet = new_sheet new_sheet.append(create_write_only_row(header, new_sheet)) sheet_map[title] = new_sheet # 遍历行写入对应工作表 current_sheet = None current_client_start_row = None skipped_rows_per_client = 0 skip_child = False skipped_parent_outline_level = 0 client_title_map = {start_row: title for title, start_row in client_meta} rows = data_sheet.rows next(rows) # 跳过已处理的表头行 for index, row in enumerate(rows, start=2): row_dimension = data_sheet.row_dimensions[index] # 原有跳过子行、无效行逻辑保持不变 if skip_child and skipped_parent_outline_level < row_dimension.outlineLevel: skipped_rows_per_client += 1 continue if skip_child and skipped_parent_outline_level >= row_dimension.outlineLevel: skip_child = False if skip_row(row, row_dimension): skipped_rows_per_client += 1 skip_child = True skipped_parent_outline_level = row_dimension.outlineLevel continue # 识别到新客户端行,切换当前写入目标工作表 if index in client_title_map: current_sheet = sheet_map[client_title_map[index]] current_client_start_row = index skipped_rows_per_client = 0 # 复制行维度配置、写入当前行 new_row_index = index - skipped_rows_per_client - current_client_start_row + 2 current_sheet.row_dimensions[new_row_index] = copy(row_dimension) current_sheet.row_dimensions[new_row_index].worksheet = current_sheet current_sheet.append(create_write_only_row(row, current_sheet)) if index % 10000 == 0: logger.info(f"{index} rows processed") logger.info(f"Writing workbook {output_file}") output_workbook.save(output_file) finally: if workbook: workbook.close() if output_workbook: output_workbook.close()
注意:如果存在不同客户端名称清洗后得到相同工作表名的情况,需要在第一次收集元数据时增加去重重名逻辑,避免创建工作表时报错。
内容的提问来源于stack exchange,提问作者William
相关产品推荐
相关产品推荐

