Python大文件数据重排优化：900k条目处理耗时过长求高效方案

原方案低效的核心原因 #

• 一次性把900k条目的整个文件加载到内存列表，内存占用极高
• 嵌套循环遍历列表，时间复杂度飙升
• 使用list.remove()和list.insert()这类O(n)级别的操作，频繁修改列表会大幅拖慢处理速度
• 逐个修改列表元素的换行和空格，冗余操作多，完全没必要

优化思路 #

每个throw块都是独立结构，我们可以逐块读取、处理、输出，根本不需要把整个文件塞进内存：

1. 逐行读取源文件，用/begin throw和/end throw作为块的边界标记
2. 在块内分别提取STORE_ADDRESS、COLOR、DESCRIPTION的所有内容行、DIMENSION的所有内容行
3. 按目标格式拼接这些字段，直接写入输出文件，全程不堆积大量数据在内存里

实现代码 #

def process_throw_blocks(input_path, output_path):
    with open(input_path, 'r') as infile, open(output_path, 'w') as outfile:
        current_block = {}
        in_block = False
        # 标记是否在收集DESCRIPTION或DIMENSION的后续内容
        collecting_desc = False
        collecting_dim = False
        
        for line in infile:
            stripped_line = line.strip()
            if not stripped_line:
                continue  # 跳过空行
            
            if stripped_line == '/begin throw':
                in_block = True
                # 初始化当前块的存储结构，同时记录原begin行的缩进
                begin_indent = line[:line.find('/begin')]
                current_block = {
                    'begin_indent': begin_indent,
                    'color': '',
                    'store_address': '',
                    'description': [],
                    'dimension': []
                }
                continue
            
            if stripped_line == '/end throw':
                # 按目标格式拼接当前块内容并写入
                bi = current_block['begin_indent']
                out_content = (
                    f"{bi}/begin throw\n"
                    f"{bi} STORE_ADDRESS {current_block['store_address']}\n"
                    f"{bi}    COLOR {current_block['color']}\n"
                    f"{bi}     DESCRIPTION {' '.join(current_block['description'])}\n"
                    f"{bi}     DIMENSION {' '.join(current_block['dimension'])}\n"
                    f"{bi}/end throw\n"
                )
                outfile.write(out_content)
                # 重置状态
                in_block = False
                collecting_desc = False
                collecting_dim = False
                continue
            
            if not in_block:
                continue  # 不在块内的内容直接跳过
            
            # 处理块内的各个字段
            if stripped_line.startswith('COLOR'):
                current_block['color'] = stripped_line.split(' ', 1)[1]
            elif stripped_line.startswith('STORE_ADDRESS'):
                current_block['store_address'] = stripped_line.split(' ', 1)[1]
            elif stripped_line.startswith('DESCRIPTION'):
                collecting_desc = True
                collecting_dim = False
            elif stripped_line.startswith('DIMENSION'):
                collecting_dim = True
                collecting_desc = False
            else:
                # 收集DESCRIPTION或DIMENSION的后续内容行
                if collecting_desc:
                    current_block['description'].append(stripped_line)
                elif collecting_dim:
                    current_block['dimension'].append(stripped_line)

# 调用示例
process_throw_blocks('your_input_file.txt', 'your_output_file.txt')

代码优势 #

• 内存友好：每次只处理单个块，不会加载整个超大文件到内存
• 速度极快：避免了列表的低效修改操作，逐行处理+直接写入，时间复杂度接近O(n)
• 容错性强：通过状态标记收集内容，不依赖固定行号（原代码靠行号判断极易出错）
• 格式精准：保留原块的基础缩进，按目标要求生成对应层级的缩进

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