You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何优化大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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.05 14:25:18