如何优化大CSV多条件拆分代码,解决嵌套块过多语法错误?
解决Python处理多输出CSV文件时的嵌套块过多问题
问题背景
有一个大型CSV文件,格式示例如下:
Column1;Column2 01;BE 02;ED 12;FD 14;DS 03;ED 04;DF
需要读取该文件并按特定条件将数据导出至多个CSV文件。最初使用Python的csv模块实现,但因输出文件数量超过20个,触发SyntaxError: too many statically nested blocks错误。原实现代码如下:
import csv import os output_path=r'C:\myfolder\large_file.csv' with open(os.path.join(os.path.dirname(output_path),"first_subset_total.csv"), "w", encoding="utf-8", newline='') as \ out_01, open(os.path.join(os.path.dirname(output_path),"excluded_first.csv"), "w", encoding="utf-8", newline='') as \ out_02, open(os.path.join(os.path.dirname(output_path),"pure_subset.csv"), "w", encoding="utf-8", newline='') as \ out_03_a, open(os.path.join(os.path.dirname(output_path),"final_subset.csv"), "w", encoding="utf-8", newline='') as \ out_04_b: cw01 = csv.writer(out_01, delimiter=";", quoting=csv.QUOTE_MINIMAL) cw02 = csv.writer(out_02, delimiter=";", quoting=csv.QUOTE_MINIMAL) cw03_a = csv.writer(out_03_a, delimiter=";", quoting=csv.QUOTE_MINIMAL) cw04_b = csv.writer(out_04_b, delimiter=";", quoting=csv.QUOTE_MINIMAL) with open(output_path, encoding="utf-8") as in_f: cr = csv.reader(in_f, delimiter=";") header = next(cr) cw01.writerow(header) cw02.writerow(header) cw03_a.writerow(header) cw04_b.writerow(header) for line in cr: if (line[0][:2]=="01" and ...): cw01.writerow(line) if (line[0][:2]=="02"): cw02.writerow(line) if (line[0][:2]=="03" and ...): cw03_a.writerow(line) if (line[0][:2]=="04" and ...): cw04_b.writerow(line)
临时采用嵌套循环的解决方案,但存在效率低、可读性差的问题,需要更高效的处理方式。
优化方案
核心思路是用字典管理输出文件对象与对应的writer,同时将条件判断逻辑与文件写入操作解耦,彻底避免大量嵌套的with块和重复代码。
方案1:动态管理文件与writer(基础版)
将输出规则集中配置,动态创建文件和writer,最后统一关闭资源:
import csv import os output_path = r'C:\myfolder\large_file.csv' output_dir = os.path.dirname(output_path) # 集中定义输出规则:(文件名, 行匹配函数) output_rules = [ ("first_subset_total.csv", lambda line: line[0][:2] == "01" and ...), # 替换...为实际条件 ("excluded_first.csv", lambda line: line[0][:2] == "02"), ("pure_subset.csv", lambda line: line[0][:2] == "03" and ...), ("final_subset.csv", lambda line: line[0][:2] == "04" and ...), # 可继续添加任意数量的输出规则 ] # 初始化writer字典与文件句柄列表 writers = {} file_handles = [] # 创建所有输出文件并写入表头 for filename, _ in output_rules: file_path = os.path.join(output_dir, filename) file_handle = open(file_path, "w", encoding="utf-8", newline='') file_handles.append(file_handle) writers[filename] = csv.writer(file_handle, delimiter=";", quoting=csv.QUOTE_MINIMAL) # 读取输入文件并处理数据 with open(output_path, encoding="utf-8") as in_f: reader = csv.reader(in_f, delimiter=";") header = next(reader) # 给所有输出文件写入表头 for writer in writers.values(): writer.writerow(header) # 逐行匹配规则并写入对应文件 for line in reader: for filename, condition in output_rules: if condition(line): writers[filename].writerow(line) # 统一关闭所有输出文件 for handle in file_handles: handle.close()
方案2:用上下文管理器自动管理文件(进阶版)
使用contextlib.ExitStack自动管理多个文件上下文,无需手动关闭文件,更安全可靠:
import csv import os from contextlib import ExitStack output_path = r'C:\myfolder\large_file.csv' output_dir = os.path.dirname(output_path) output_rules = [ ("first_subset_total.csv", lambda line: line[0][:2] == "01" and ...), ("excluded_first.csv", lambda line: line[0][:2] == "02"), # 更多输出规则... ] with ExitStack() as stack: writers = {} # 打开所有输出文件并加入上下文栈(自动管理关闭) for filename, _ in output_rules: file_path = os.path.join(output_dir, filename) file_handle = stack.enter_context(open(file_path, "w", encoding="utf-8", newline='')) writers[filename] = csv.writer(file_handle, delimiter=";", quoting=csv.QUOTE_MINIMAL) # 读取输入文件处理数据 with open(output_path, encoding="utf-8") as in_f: reader = csv.reader(in_f, delimiter=";") header = next(reader) # 写入所有输出文件的表头 for writer in writers.values(): writer.writerow(header) # 逐行匹配规则写入 for line in reader: for filename, condition in output_rules: if condition(line): writers[filename].writerow(line)
优化优势
- 彻底解决
too many statically nested blocks错误,不再受输出文件数量限制 - 规则集中管理,新增/修改输出文件仅需调整
output_rules,维护成本极低 - 逻辑解耦:条件判断与文件操作分离,代码结构清晰,可读性大幅提升
- 保持逐行处理的特性,适合大型CSV文件,不会占用过多内存
内容的提问来源于stack exchange,提问作者PSt
相关产品推荐
相关产品推荐

