使用Python提取交通工程TXT数据至CSV文件的技术问询

针对你提供的从交通工程TXT报告提取队列长度等数据到CSV的代码，下面是修正后的版本及问题说明：

原代码核心问题 #

• 变量定义语法错误：多行变量连写未分隔（如TOD_List = ['AM','PM']Years = 2025），直接触发语法报错
• 数据存储与清空时机错误：每读取一行就清空Col/TM/QL/VC列表，导致未收集完完整数据就丢失
• 写入逻辑异常：写入触发条件被注释，且未等待所有相关数据项收集完成就尝试写入
• 字符串匹配容错性差：直接匹配带固定空格的字符串，若原文件空格数量变化则匹配失败
• 变量初始化位置不合理：部分变量未在每个文件处理前重置，可能残留上一个文件的无效数据

修正后的代码 #

import csv

# 配置参数
TOD_List = ['AM', 'PM']
Years = [2025]  # 后续可扩展为多年份列表
BEBR_List = ['Design']
Folder = r"I:\TPA\PRJ\000011957\1.0_Task_Work_Orders_(TWOs)\1.5_TWO_5_081419\I75atSR52_Study\Traffic\B - Operational Analysis\test"
Output = r"I:\TPA\PRJ\000011957\1.0_Task_Work_Orders_(TWOs)\1.5_TWO_5_081419\I75atSR52_Study\Traffic\B - Operational Analysis\test\test.csv"

with open(Output, 'w', newline='') as OutCSV:
    fieldnames = ['TOD', 'Year', 'BEBR', 'Int', 'Mvmt', 'QL95', 'Stor', 'QL50']
    writer = csv.DictWriter(OutCSV, fieldnames=fieldnames)
    writer.writeheader()

    for TOD in TOD_List:
        for Year in Years:
            for BEBR in BEBR_List:
                Input_Report = fr"{Folder}\Volume_{TOD}_{Year}_{BEBR}_Report.txt"
                
                # 每个文件处理前重置变量，避免残留数据
                Count = 0
                Intersection = ""
                Col = []
                TM = []
                QL = []
                VC = []
                Num_Move = 0

                with open(Input_Report, 'r') as infile:
                    reader = csv.reader(infile, delimiter='\t')
                    
                    for row in reader:
                        if len(row) == 0:
                            continue  # 跳过空行
                        
                        # 触发Queues章节时重置计数
                        if row[0].strip() == 'Queues':
                            Count = 0
                            Sig = 1
                            Unsig = 0
                        Count += 1

                        # 提取交叉口ID
                        if Count == 2:
                            Intersection = row[0].split(':')[0].strip()

                        # 提取转向组名称
                        if row[0].strip() == 'Lane Group':
                            Num_Move = len(row)
                            # 从第3列开始提取转向组（索引2）
                            for x in range(2, Num_Move):
                                if row[x].strip():  # 跳过空列
                                    Col.append(row[x].strip())

                        # 提取95th队列长度
                        if row[0].strip() == 'Queue Length 95th (ft)':
                            for x in range(2, Num_Move):
                                if x-2 < len(Col):  # 对齐转向组数量
                                    TM.append(row[x].strip())

                        # 提取50th队列长度
                        if row[0].strip() == 'Queue Length 50th (ft)':
                            for x in range(2, Num_Move):
                                if x-2 < len(Col):
                                    QL.append(row[x].strip())

                        # 提取转弯 bay长度
                        if row[0].strip() == 'Turn Bay Length (ft)':
                            for x in range(2, Num_Move):
                                if x-2 < len(Col):
                                    VC.append(row[x].strip())

                        # 当收集完所有必要数据后，写入CSV
                        # 假设"Intersection Summary"是当前交叉口数据结束的标记
                        if row[0].strip() == 'Intersection Summary':
                            # 确保所有列表长度一致再写入
                            if len(Col) == len(TM) == len(QL) == len(VC):
                                for x in range(len(Col)):
                                    writer.writerow({
                                        'TOD': TOD,
                                        'Year': Year,
                                        'BEBR': BEBR,
                                        'Int': Intersection,
                                        'Mvmt': Col[x],
                                        'QL95': TM[x],
                                        'QL50': QL[x],
                                        'Stor': VC[x]
                                    })
                            # 重置当前交叉口的变量，准备下一个交叉口
                            Col = []
                            TM = []
                            QL = []
                            VC = []
                            Intersection = ""

print("数据提取完成")

关键修正说明 #

1. 语法错误修复：拆分连写的变量定义，将Years改为列表形式方便后续扩展多年份
2. 数据生命周期管理：在每个文件处理前初始化变量，仅当检测到Intersection Summary（交叉口数据结束标记）时才写入并重置数据列表
3. 字符串匹配优化：使用strip()去除字符串前后空格，避免因原文件空格数量不一致导致匹配失败
4. 数据一致性检查：写入前校验所有数据列表长度是否一致，避免索引越界错误
5. 空行处理：跳过空行，减少无效循环

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