Python PermissionError Errno13咨询：CSV处理提速替代方案

运行以下CSV批量拆分代码时，随机出现PermissionError: Errno 13权限拒绝错误，推测是文件频繁打开关闭导致系统未及时释放文件句柄。目前通过添加time.sleep(.0000001)规避，但运行速度大幅下降，寻求更高效的解决办法。

原代码 #

import csv
import time

OriginalFilePath = 'c:/Users/jason/Desktop/FCC_Stuff/'
OriginalFile = 'bdc_29_Fiber-to-the-Premises_fixed_broadband_063022'
OriginalExt = '.csv'
CompleteOriginalFile = OriginalFilePath + OriginalFile + OriginalExt

with open(CompleteOriginalFile) as csv_file:
    csv_reader = csv.reader(csv_file, delimiter=',')
    line_count = 0
    TheList = []
    for row in csv_reader:
        if line_count == 0:
            print(f'Column names are {", ".join(row)}')
            HeaderRow = row
            line_count += 1
            print("")
            print(HeaderRow)
        else:
            Brand = (row[2])
            if Brand in TheList:
                TheOutputFile = OriginalFilePath + Brand + OriginalExt
                with open(TheOutputFile, 'a', newline='') as csvoutput:
                    csv_writer = csv.writer(csvoutput, delimiter=',')
                    csv_writer.writerow(row)
                csvoutput.close()

            if Brand not in TheList:
                print(Brand)
                TheList.append(Brand)
                TheOutputFile2 = OriginalFilePath + Brand + OriginalExt
                with open(TheOutputFile2, 'a', newline='') as csvoutput2:
                    csv_writer = csv.writer(csvoutput2, delimiter=',')
                    csv_writer.writerow(HeaderRow)
                    csv_writer.writerow(row)
                csvoutput2.close()
        line_count += 1
csv_file.close()

print(f'Processed {line_count} lines.')
print("")

print("done")

添加time.sleep后的代码 #

import csv
import time

OriginalFilePath = 'c:/Users/jason/Desktop/FCC_Stuff/'
OriginalFile = 'bdc_29_Fiber-to-the-Premises_fixed_broadband_063022'
OriginalExt = '.csv'
CompleteOriginalFile = OriginalFilePath + OriginalFile + OriginalExt

with open(CompleteOriginalFile) as csv_file:
    csv_reader = csv.reader(csv_file, delimiter=',')
    line_count = 0
    TheList = []
    for row in csv_reader:
        if line_count == 0:
            print(f'Column names are {", ".join(row)}')
            HeaderRow = row
            line_count += 1
            print("")
            print(HeaderRow)
        else:
            Brand = (row[2])
            if Brand in TheList:
                TheOutputFile = OriginalFilePath + Brand + OriginalExt
                with open(TheOutputFile, 'a', newline='') as csvoutput:
                    csv_writer = csv.writer(csvoutput, delimiter=',')
                    csv_writer.writerow(row)
                csvoutput.close()
                time.sleep(.0000001)

            if Brand not in TheList:
                print(Brand)
                TheList.append(Brand)
                TheOutputFile2 = OriginalFilePath + Brand + OriginalExt
                with open(TheOutputFile2, 'a', newline='') as csvoutput2:
                    csv_writer = csv.writer(csvoutput2, delimiter=',')
                    csv_writer.writerow(HeaderRow)
                    csv_writer.writerow(row)
                csvoutput2.close()
                time.sleep(.0000001)
        line_count += 1
csv_file.close()

print(f'Processed {line_count} lines.')
print("")

print("done")

高效解决方案 #

核心思路是减少文件打开/关闭的次数：不要每处理一行就打开一次文件，而是为每个品牌保持一个打开的文件句柄，直到所有数据处理完成后再统一关闭。

优化后的代码 #

import csv

OriginalFilePath = 'c:/Users/jason/Desktop/FCC_Stuff/'
OriginalFile = 'bdc_29_Fiber-to-the-Premises_fixed_broadband_063022'
OriginalExt = '.csv'
CompleteOriginalFile = OriginalFilePath + OriginalFile + OriginalExt

# 用字典存储每个品牌的文件对象和writer，避免重复打开
brand_files = {}

with open(CompleteOriginalFile) as csv_file:
    csv_reader = csv.reader(csv_file, delimiter=',')
    line_count = 0
    HeaderRow = []
    
    for row in csv_reader:
        if line_count == 0:
            print(f'Column names are {", ".join(row)}')
            HeaderRow = row
            line_count += 1
            print("\n", HeaderRow)
        else:
            brand = row[2]
            # 如果品牌不在字典中，创建文件并写入表头
            if brand not in brand_files:
                print(brand)
                output_path = f"{OriginalFilePath}{brand}{OriginalExt}"
                # 打开文件并保持句柄
                file_obj = open(output_path, 'a', newline='')
                writer = csv.writer(file_obj, delimiter=',')
                writer.writerow(HeaderRow)
                brand_files[brand] = (file_obj, writer)
            
            # 直接用已打开的writer写入行
            _, writer = brand_files[brand]
            writer.writerow(row)
            line_count += 1

# 处理完所有数据后，统一关闭所有文件
for file_obj, _ in brand_files.values():
    file_obj.close()

print(f'\nProcessed {line_count} lines.')
print("done")

关键优化点 #

1. 复用文件句柄：用字典brand_files存储每个品牌对应的文件对象和writer，每个品牌只打开一次文件，彻底避免频繁打开/关闭导致的系统句柄未释放问题。
2. 移除冗余操作：原代码中with语句已经自动关闭文件，手动调用csvoutput.close()是多余的，优化后完全移除。
3. 简化逻辑：将原代码中两个独立的if判断合并为一次逻辑判断，避免重复检查品牌是否存在。
4. 无延迟开销：不需要依赖time.sleep，运行速度比原代码快数倍。

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