Google Maps API出现KeyError: 'duration_in_traffic'的排查与修复方案

我在Jupyter Notebook中编写了代码，用于从Google Maps采集指定路段的采集时间、起止地点及行程时间，期望代码不间断运行：

import datetime
import googlemaps
import csv
import pandas
import time
from datetime import datetime

file = open('key.txt', 'r')
key = file.read().strip()

gmaps = googlemaps.Client(key=key)

x=time.localtime()

while 1:
        ## consultation BR-040 urban (1) - BSB - BH
        query_040urb1AB = gmaps.distance_matrix('-19.741224, -44.148914', '-19.813856, -44.092239', mode='driving', departure_time='now')
        a = str(datetime.now())
        b = query_040urb1AB['rows'][0]['elements'][0]['duration_in_traffic']['value']

        consulta_final_040urb1AB = {'Segment': '1', 'Direction': 'BSB-BH', 'Data': str(datetime.now()),
                          'Origin': query_040urb1AB['origin_addresses'], 'Destination': query_040urb1AB['destination_addresses'], 'Travel Time': query_040urb1AB['rows'][0]['elements'][0]['duration_in_traffic'] ['value']}

        print(query_final_040urb1AB)
        with open('040urb1AB.csv', 'a', newline='', encoding='UTF-8') as csvfile:
            csv.writer(csvfile, delimiter=',').writerow([a,b])

        ## consultation BR-040 urban (1) - BH - BSB
        query_040urb1BA = gmaps.distance_matrix('-19.813770, -44.092143', '-19.741235, -44.148780', mode='driving', departure_time='now')
        a = str(datetime.now())
        b = query_040urb1BA['rows'][0]['elements'][0]['duration_in_traffic']['value']

        consulta_final_040urb1BA = {'Segment': '1', 'Direction': 'BH-BSB', 'Data': str(datetime.now()),
                          'Origin': query_040urb1BA['origin_addresses'], 'Destination': query_040urb1BA['destination_addresses'], 'Travel Time': query_040urb1BA['rows'][0]['elements'][0]['duration_in_traffic'] ['value']}

        print(query_final_040urb1BA)
        with open('040urb1BA.csv', 'a', newline='', encoding='UTF-8') as csvfile:
            csv.writer(csvfile, delimiter=',').writerow([a,b])

但运行一段时间后，代码总会终止并抛出如下错误：

KeyError Traceback (most recent call last)
~\AppData\Local\Temp/ipykernel_25760/3076371497.py in <module>
    113 query_262urb1AB = gmaps.distance_matrix('-19.977448, -44.255544', '-19.961123, -44.336237', mode='driving', departure_time='now')
    114 a = str(datetime.now())
--> 115 b = query_262urb1AB['rows'][0]['elements'][0]['duration_in_traffic']['value']
    116
    117 final_query_262urb1AB = {'Segment': '1', 'Direction': 'BH-UBE', 'Data': str(datetime.now()),

KeyError: 'duration_in_traffic'

错误原因分析 #

• Google Maps Distance Matrix API的响应并非每次都包含duration_in_traffic字段，常见触发场景：
  • 请求路段无实时交通数据支持
  • API调用频率超限，触发限流后返回异常响应
  • 网络波动导致响应不完整
  • 目标路段无法通行（如封闭），此时响应会返回status字段（例如"NOT_FOUND"或"ZERO_RESULTS"），而非duration_in_traffic
• 原代码直接硬编码提取字段，未做任何异常判断，一旦字段缺失就会触发KeyError导致程序崩溃

修复方案 #

要实现代码不间断运行，需要添加状态检查、异常捕获和调用间隔，同时优化代码结构：

修改后的完整代码 #

import datetime
import googlemaps
import csv
import time
from datetime import datetime

# 读取API密钥
with open('key.txt', 'r') as file:
    key = file.read().strip()

gmaps = googlemaps.Client(key=key)

def fetch_and_save_traffic_data(origin, dest, segment, direction, csv_filename):
    """封装路段数据采集与保存逻辑"""
    current_time = str(datetime.now())
    try:
        # 调用API
        response = gmaps.distance_matrix(origin, dest, mode='driving', departure_time='now')
        # 检查响应状态
        element = response['rows'][0]['elements'][0]
        if element['status'] != 'OK':
            print(f"{current_time} - {segment} {direction} 路段异常: {element['status']}")
            # 写入异常标记
            with open(csv_filename, 'a', newline='', encoding='UTF-8') as csvfile:
                csv.writer(csvfile).writerow([current_time, 'ERROR', element['status']])
            return
        
        # 提取有效数据
        travel_time = element['duration_in_traffic']['value']
        result = {
            'Segment': segment,
            'Direction': direction,
            'Data': current_time,
            'Origin': response['origin_addresses'][0],
            'Destination': response['destination_addresses'][0],
            'Travel Time': travel_time
        }
        print(result)
        
        # 写入CSV
        with open(csv_filename, 'a', newline='', encoding='UTF-8') as csvfile:
            csv.writer(csvfile).writerow([current_time, travel_time])
    
    except KeyError as e:
        print(f"{current_time} - 字段缺失错误: {str(e)}")
        with open(csv_filename, 'a', newline='', encoding='UTF-8') as csvfile:
            csv.writer(csvfile).writerow([current_time, 'KEY_ERROR', str(e)])
    except Exception as e:
        # 捕获其他所有异常，避免程序崩溃
        print(f"{current_time} - 未知错误: {str(e)}")
        with open(csv_filename, 'a', newline='', encoding='UTF-8') as csvfile:
            csv.writer(csvfile).writerow([current_time, 'UNKNOWN_ERROR', str(e)])

# 主循环
while True:
    # 采集BR-040 urban (1) BSB-BH
    fetch_and_save_traffic_data(
        '-19.741224, -44.148914', 
        '-19.813856, -44.092239', 
        '1', 
        'BSB-BH', 
        '040urb1AB.csv'
    )
    
    # 采集BR-040 urban (1) BH-BSB
    fetch_and_save_traffic_data(
        '-19.813770, -44.092143', 
        '-19.741235, -44.148780', 
        '1', 
        'BH-BSB', 
        '040urb1BA.csv'
    )
    
    # 添加调用间隔，避免触发API限流（可根据需求调整，比如60秒）
    time.sleep(60)

关键改进点 #

1. 封装逻辑：把采集和保存的代码封装成函数，减少重复代码，提升可维护性
2. 状态检查：先判断element['status']是否为OK，再提取数据，提前处理异常路段
3. 异常捕获：用try-except块捕获KeyError和其他所有异常，记录错误信息并继续运行
4. 调用间隔：添加time.sleep(60)控制调用频率，避免触发Google API的限流机制
5. 错误记录：把异常信息写入CSV，方便后续排查问题

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