1. 批量数据准备与API调用基础
在处理地理数据时，我们经常需要计算不同地点之间的距离。假设我们有四组列表数据，分别代表起始地点的纬度、经度以及目的地的纬度、经度，并希望通过一个外部api（例如osrm路由服务）来获取它们之间的驾驶距离。
首先，我们需要一个函数来封装对API的调用。这个函数接收两组坐标（起始点和目的点）并返回驾驶距离。
import requests
import json
def get_driving_distance(lat1, lon1, lat2, lon2):
"""
通过OSRM API计算两点之间的驾驶距离。
参数:
lat1 (float): 起始点纬度
lon1 (float): 起始点经度
lat2 (float): 目的点纬度
lon2 (float): 目的点经度
返回:
float: 驾驶距离（英里）
"""
# OSRM API的URL格式是 /lon1,lat1;lon2,lat2
url = f"http://router.project-osrm.org/route/v1/car/{lon1},{lat1};{lon2},{lat2}?overview=false"
r = requests.get(url)
# 解析API响应
routes = json.loads(r.content)
# 检查API响应是否包含有效的路线信息
if "routes" in routes and len(routes["routes"]) > 0:
route_info = routes["routes"][0]
driving_distance_meters = route_info['distance']
# 将米转换为英里 (1 英里 ≈ 1609.34 米)
return driving_distance_meters / 1609.34
else:
print(f"API响应未包含有效路线: {routes}")
return None # 或者抛出异常
当拥有多组坐标列表时，我们可以使用Python内置的zip函数将这些列表的对应元素打包成元组，然后进行迭代调用上述函数。
# 示例数据
location_latitudes = [51.5074, 51.5100]
location_longitudes = [-0.1278, -0.1300]
station_latitudes = [51.5200, 51.5300]
station_longitudes = [-0.1000, -0.1100]
# 迭代并计算距离
distances = []
for lat1, lon1, lat2, lon2 in zip(location_latitudes, location_longitudes, station_latitudes, station_longitudes):
distance = get_driving_distance(lat1, lon1, lat2, lon2)
distances.append(distance)
print(f"计算出的距离: {distances}")2. API请求的健壮性与错误处理
在实际的API调用中，网络问题、API服务中断或请求参数错误都可能导致失败。为了使代码更加健壮，我们应该检查API响应的状态码。requests库提供了Response.raise_for_status()方法，它会在收到错误状态码（如4xx或5xx）时抛出HTTPError异常。或者，我们也可以直接检查Response.status_code属性。
import requests
import json
from http import HTTPStatus # 导入HTTPStatus枚举，提高可读性
def get_driving_distance_robust(lat1, lon1, lat2, lon2):
"""
通过OSRM API计算两点之间的驾驶距离，包含错误处理。
"""
url = f"http://router.project-osrm.org/route/v1/car/{lon1},{lat1};{lon2},{lat2}?overview=false"
try:
r = requests.get(url)
r.raise_for_status() # 如果状态码不是200，则抛出HTTPError
routes = json.loads(r.content)
if "routes" in routes and len(routes["routes"]) > 0:
route_info = routes["routes"][0]
driving_distance_meters = route_info['distance']
return driving_distance_meters / 1609.34
else:
print(f"API响应未包含有效路线或数据: {routes}")
return None
except requests.exceptions.HTTPError as e:
print(f"HTTP错误发生: {e.response.status_code} - {e.response.text}")
return None
except requests.exceptions.ConnectionError as e:
print(f"网络连接错误: {e}")
return None
except json.JSONDecodeError as e:
print(f"JSON解析错误: {e.msg} - 响应内容: {r.text if 'r' in locals() else 'N/A'}")
return None
except Exception as e:
print(f"发生未知错误: {e}")
return None3. 实现API请求限流机制
频繁地调用API可能会触及服务提供商的请求限制，导致IP被封禁或服务暂时不可用。为了避免这种情况，我们可以实现一个简单的限流机制。Python的contextlib.contextmanager提供了一种优雅的方式来管理资源，非常适合实现这种“进入-执行-退出”的逻辑。
立即学习“Python免费学习笔记（深入）”；
我们将创建一个上下文管理器，它会跟踪API调用次数，并在达到预设限制时暂停执行一段时间。
from contextlib import contextmanager
from time import sleep
import requests
import json
# 全局变量，用于跟踪API调用次数
api_calls = 0
@contextmanager
def rate_limited(limit=500, delay=5):
"""
一个上下文管理器，用于实现API请求限流。
当API调用次数达到limit时，暂停delay秒。
参数:
limit (int): 允许在不暂停的情况下进行的API调用次数上限。
delay (int): 达到上限后暂停的秒数。
"""
global api_calls # 声明使用全局变量
# 在进入上下文之前检查是否需要暂停
if api_calls + 1 >= limit:
print(f"达到 {limit} 次API调用限制，暂停 {delay} 秒...")
sleep(delay)
api_calls = 0 # 暂停后重置计数器
api_calls += 1 # 每次进入上下文时增加计数
yield # 执行被包装的代码块
# 将限流机制集成到API调用函数中
def get_driving_distance_rate_limited(lat1, lon1, lat2, lon2):
"""
通过OSRM API计算两点之间的驾驶距离，集成限流和错误处理。
"""
with rate_limited(limit=500, delay=5): # 使用限流上下文管理器
url = f"http://router.project-osrm.org/route/v1/car/{lon1},{lat1};{lon2},{lat2}?overview=false"
try:
r = requests.get(url)
r.raise_for_status()
routes = json.loads(r.content)
if "routes" in routes and len(routes["routes"]) > 0:
route_info = routes["routes"][0]
driving_distance_meters = route_info['distance']
return driving_distance_meters / 1609.34
else:
print(f"API响应未包含有效路线或数据: {routes}")
return None
except requests.exceptions.HTTPError as e:
print(f"HTTP错误发生: {e.response.status_code} - {e.response.text}")
return None
except requests.exceptions.ConnectionError as e:
print(f"网络连接错误: {e}")
return None
except json.JSONDecodeError as e:
print(f"JSON解析错误: {e.msg} - 响应内容: {r.text if 'r' in locals() else 'N/A'}")
return None
except Exception as e:
print(f"发生未知错误: {e}")
return None
4. 整合与数据处理
现在，我们可以将所有组件整合起来，处理多组数据，并最终将结果整理成易于分析的格式，例如Pandas DataFrame。
import pandas as pd
# 假设 get_driving_distance_rate_limited 函数已定义
# 更多示例数据
location_latitudes = [51.5074, 51.5100, 51.5000, 51.5050, 51.5150]
location_longitudes = [-0.1278, -0.1300, -0.1100, -0.1200, -0.1400]
station_latitudes = [51.5200, 51.5300, 51.5100, 51.5250, 51.5350]
station_longitudes = [-0.1000, -0.1100, -0.0900, -0.1050, -0.1250]
# 准备一个列表来存储所有结果
results = []
# 批量调用API并收集结果
for i, (lat1, lon1, lat2, lon2) in enumerate(zip(location_latitudes, location_longitudes, station_latitudes, station_longitudes)):
print(f"处理第 {i+1} 组数据: ({lat1}, {lon1}) -> ({lat2}, {lon2})")
distance = get_driving_distance_rate_limited(lat1, lon1, lat2, lon2)
# 将原始数据和计算出的距离存储起来
results.append({
'Location_Lat': lat1,
'Location_Lon': lon1,
'Station_Lat': lat2,
'Station_Lon': lon2,
'Driving_Distance_Miles': distance
})
# 将结果转换为Pandas DataFrame
df = pd.DataFrame(results)
print("\n最终结果DataFrame:")
print(df)
# 可以将DataFrame保存到CSV文件
# df.to_csv("driving_distances.csv", index=False)5. 注意事项与最佳实践
- API服务条款： 在使用任何第三方API之前，务必仔细阅读其服务条款和使用限制。OSRM是一个开源路由引擎，通常用于自部署或低量请求。对于商业应用或高并发需求，可能需要考虑使用Google Maps API、Mapbox等提供商的付费服务，它们通常有更严格的限额和计费模式。
- 异常处理的粒度： 针对不同类型的API错误（例如，404 Not Found、429 Too Many Requests、5xx Server Error），可以实现更细致的错误处理逻辑，例如重试机制、错误日志记录或通知。
- 数据持久化： 如果需要处理大量数据，建议在处理过程中定期保存中间结果。这样即使程序中断，也无需从头开始，可以从上次保存的地方恢复。
- 并发/异步请求： 对于极其庞大的数据集，单线程顺序调用API效率可能较低。可以考虑使用concurrent.futures库（例如ThreadPoolExecutor）或asyncio库来实现并发或异步请求，进一步提高数据获取速度，但需要注意并发请求对API限流策略的影响，可能需要更复杂的限流逻辑。
- API密钥管理： 如果使用的API需要密钥，切勿将密钥硬编码在代码中。应使用环境变量、配置文件或专门的密钥管理服务来安全地存储和访问密钥。
总结
通过本文的学习，我们掌握了如何利用Python高效地处理来自多个列表的数据，并通过API进行批量操作。核心技术包括：
- 使用zip函数优雅地迭代多组相关数据。
- 利用requests库进行HTTP请求，并结合raise_for_status()进行健壮的错误处理。
- 通过contextlib.contextmanager创建自定义上下文管理器，实现智能的API请求限流机制，以避免触发API限制并确保程序稳定运行。
- 将处理结果整合到Pandas DataFrame中，便于后续的数据分析和存储。
这些技术不仅适用于地理数据处理，也适用于任何需要批量调用API的场景，是构建高效、可靠数据处理流程的关键。
