使用get()函数读取GTFS tripupdates.pb实时数据报错求助

目标数据结构 #

执行dict_obj['entity'][0]得到的输出：

{'id': '8800314',
 'tripUpdate': {'trip': {'tripId': '8800314',
   'startTime': '11:30:00',
   'startDate': '20240313',
   'routeId': '20832'},
  'stopTimeUpdate': [{'stopSequence': 1,
    'arrival': {'time': '1710344086'},
    'departure': {'time': '1710344086'},
    'stopId': '86900',
    'stopTimeProperties': {}},
   {'stopSequence': 2,
    'arrival': {'time': '1710343956'},
    'departure': {'time': '1710343956'},
    'stopId': '86024',
    'stopTimeProperties': {}},
   {'stopSequence': 3,
    'arrival': {'time': '1710343995'},
    'departure': {'time': '1710343995'},
    'stopId': '86560',
    'stopTimeProperties': {}},
   {'stopSequence': 4,

编写的提取代码 #

#for trip updates

collector1 = []
counter1=0
for block1 in dict_obj1['entity']:
    counter1 += 1
    row = OrderedDict()
    row['stop_AT'] = block1.get('tripUpdate').get('stopTimeUpdate')[0].get('arrival').get('time')
    row['stop_DT'] = block1.get('tripUpdate').get('stopTimeUpdate')[0].get('departure').get('time')
    collector1.append(row)
df1 = pd.DataFrame(collector1)        

报错信息 #

AttributeError: 'list' object has no attribute 'get'

问题原因与解决方案 #

问题原因 #

报错核心是：某个环节调用.get()方法时，前置对象是列表而非字典。可能的触发场景包括：

• 部分entity元素不包含tripUpdate字段，或tripUpdate本身是列表类型（非预期的字典）
• stopTimeUpdate字段不存在/为空，或返回类型不符合预期，导致链式调用逻辑断裂

修复后的代码 #

添加类型判断与空值校验，避免直接链式调用.get()引发的异常：

from collections import OrderedDict
import pandas as pd

collector1 = []
for block1 in dict_obj1['entity']:
    # 校验当前entity的tripUpdate是否为有效字典
    trip_update = block1.get('tripUpdate')
    if not isinstance(trip_update, dict):
        continue
    
    # 校验stopTimeUpdate是否为非空列表
    stop_time_updates = trip_update.get('stopTimeUpdate')
    if not isinstance(stop_time_updates, list) or len(stop_time_updates) == 0:
        continue
    
    # 提取第一个站点的到达/出发时间
    first_stop = stop_time_updates[0]
    row = OrderedDict()
    row['stop_AT'] = first_stop.get('arrival', {}).get('time')
    row['stop_DT'] = first_stop.get('departure', {}).get('time')
    collector1.append(row)

df1 = pd.DataFrame(collector1)

扩展：提取所有站点的时间记录 #

如果需要获取每个车次下所有站点的到达/出发时间，可以嵌套遍历stopTimeUpdate列表：

from collections import OrderedDict
import pandas as pd

collector1 = []
for block1 in dict_obj1['entity']:
    trip_update = block1.get('tripUpdate')
    if not isinstance(trip_update, dict):
        continue
    
    stop_time_updates = trip_update.get('stopTimeUpdate')
    if not isinstance(stop_time_updates, list) or len(stop_time_updates) == 0:
        continue
    
    # 遍历每个站点的时间数据
    for stop in stop_time_updates:
        row = OrderedDict()
        row['trip_id'] = trip_update.get('trip', {}).get('tripId')
        row['stop_id'] = stop.get('stopId')
        row['stop_sequence'] = stop.get('stopSequence')
        row['stop_AT'] = stop.get('arrival', {}).get('time')
        row['stop_DT'] = stop.get('departure', {}).get('time')
        collector1.append(row)

df1 = pd.DataFrame(collector1)

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