使用Python通过API计算地理距离：数据整合与API速率限制实践

花韻仙語

发布时间：2025-08-11 19:48:28

622人浏览过

来源于php中文网

原创

使用Python通过API计算地理距离：数据整合与API速率限制实践

本文旨在指导读者如何使用Python高效地通过外部API计算地理位置间的驾驶距离。内容涵盖了从多源列表数据中提取信息、构建API请求、集成OSRM路由服务进行距离计算的核心方法，并重点介绍了如何利用Python的contextlib.contextmanager实现健壮的API请求速率限制机制，以避免超出API调用频率限制。最终，文章将展示如何将原始数据与计算结果整合至Pandas DataFrame，形成结构化的输出。

1. 引言：地理距离计算与API交互挑战

在地理信息处理和位置服务领域，计算两点间的驾驶距离是常见的需求。通常，我们会拥有多组起点和终点的经纬度数据，需要通过地图服务api进行批量查询。然而，直接进行大量api调用可能会遇到服务提供商的速率限制，导致请求失败甚至ip被封禁。因此，在进行批量api请求时，实现有效的速率限制和错误处理机制至关重要。

2. 核心功能：基于OSRM API的距离计算

本教程将以开源路由服务OSRM (Open Source Routing Machine) 为例，展示如何通过其API获取驾驶距离。OSRM提供了一个简洁的HTTP接口，允许用户查询两点间的路线信息，包括距离和持续时间。

首先，定义一个基础函数来执行API请求并解析结果：

import requests
import json
import pandas as pd
from time import sleep
from contextlib import contextmanager

def get_driving_distance(lat1, lon1, lat2, lon2):
    """
    通过OSRM API计算两点间的驾驶距离。

    Args:
        lat1 (float): 起点纬度
        lon1 (float): 起点经度
        lat2 (float): 终点纬度
        lon2 (float): 终点经度

    Returns:
        float: 驾驶距离（英里），如果API请求失败或无有效路线则返回None。
    """
    # OSRM API的URL格式为 /route/v1/car/{lon1},{lat1};{lon2},{lat2}
    # 注意：OSRM API的坐标顺序是经度在前，纬度在后。
    url = f"http://router.project-osrm.org/route/v1/car/{lon1},{lat1};{lon2},{lat2}?overview=false"

    try:
        r = requests.get(url)
        # 检查HTTP状态码，如果不是2xx，则抛出HTTPError异常
        r.raise_for_status() 

        routes = json.loads(r.content)

        # 检查API响应中是否存在'routes'键以及其是否为空
        if routes and "routes" in routes and len(routes["routes"]) > 0:
            route_info = routes["routes"][0]
            driving_distance_meters = route_info.get('distance')

            if driving_distance_meters is not None:
                # 将米转换为英里 (1 英里 ≈ 1609.34 米)
                return driving_distance_meters / 1609.34
        else:
            print(f"API响应未包含有效路线信息: {url}, 响应: {routes}")
            return None

    except requests.exceptions.HTTPError as e:
        print(f"HTTP错误: {e.response.status_code} - {e.response.text} for URL: {url}")
        return None
    except requests.exceptions.ConnectionError as e:
        print(f"连接错误: {e} for URL: {url}")
        return None
    except requests.exceptions.Timeout as e:
        print(f"请求超时: {e} for URL: {url}")
        return None
    except json.JSONDecodeError as e:
        print(f"JSON解析错误: {e} for URL: {url}, 响应内容: {r.content}")
        return None
    except Exception as e:
        print(f"发生未知错误: {e} for URL: {url}")
        return None

在上述代码中，我们增加了try-except块来捕获可能发生的网络请求异常（如HTTPError、ConnectionError、Timeout）以及JSON解析错误，这使得函数更加健壮。r.raise_for_status()是一个非常实用的方法，它会在HTTP请求返回错误状态码（如4xx或5xx）时自动抛出异常。

3. 高效处理多源地理数据

通常，我们的地理位置数据会以列表的形式存储，例如：

立即学习“Python免费学习笔记（深入）”；

location_latitudes = [34.0522, 34.0522]
location_longitudes = [-118.2437, -118.2437]
station_latitudes = [34.0689, 34.0753]
station_longitudes = [-118.2942, -118.2575]

为了将这些列表中的对应元素传递给get_driving_distance函数，可以使用Python内置的zip函数来同时迭代多个列表：

# 示例：迭代并调用函数（暂不考虑速率限制）
# for lat1, lon1, lat2, lon2 in zip(location_latitudes, location_longitudes, station_latitudes, station_longitudes):
#     distance = get_driving_distance(lat1, lon1, lat2, lon2)
#     print(f"距离: {distance} 英里")

4. API速率限制与鲁棒性实践

当需要进行大量API调用时，必须考虑API提供商的速率限制。不加限制的频繁请求可能导致服务拒绝或IP被暂时封禁。Python的contextlib.contextmanager提供了一种优雅的方式来封装资源管理逻辑，非常适合实现API速率限制。

我们将创建一个上下文管理器，它会在每次API调用前检查计数器，如果达到预设限制，则暂停一段时间。

NeuralText

Neural Text是一个使用机器学习自动生成文本的平台

下载

api_calls_counter = 0 # 全局计数器，用于跟踪API调用次数

@contextmanager
def rate_limited(limit=500, delay=5):
    """
    一个上下文管理器，用于实现API调用速率限制。
    当API调用次数达到指定限制时，暂停指定时间。

    Args:
        limit (int): 达到此次数后暂停。
        delay (int): 暂停的秒数。
    """
    global api_calls_counter

    # 在进入上下文之前检查并处理速率限制
    if api_calls_counter + 1 >= limit:
        print(f"达到 {limit} 次API调用限制，暂停 {delay} 秒...")
        sleep(delay)
        # 暂停后，将计数器重置，以允许新的批次调用
        api_calls_counter = 0 # 或者 api_calls_counter -= limit，取决于具体策略
        print("暂停结束，继续调用。")

    # 增加API调用计数
    api_calls_counter += 1

    try:
        yield # 执行被包装的代码块
    finally:
        # 可以在这里执行一些清理工作，但对于速率限制不是必需的
        pass

rate_limited 上下文管理器详解：

api_calls_counter: 一个全局变量，用于记录从上次重置以来已经进行了多少次API调用。
limit 和 delay: 上下文管理器的参数，分别定义了触发暂停的调用次数阈值和暂停的秒数。
if api_calls_counter + 1 >= limit:: 在每次进入上下文（即每次API调用前）检查，如果当前调用将使总数达到或超过限制，则触发暂停。
sleep(delay): 执行实际的暂停操作。
api_calls_counter = 0: 暂停结束后，将计数器重置为0。这意味着每500次调用后，会暂停5秒，然后重新开始计数。这是一种简单的固定窗口计数器实现。
api_calls_counter += 1: 无论是否暂停，每次进入上下文，调用计数器都会增加。
yield: 这是上下文管理器的核心。它将控制权交给with语句块内的代码。当with块执行完毕或抛出异常时，控制权会回到finally块。

接下来，我们将这个速率限制逻辑集成到get_driving_distance函数中：

# 修改后的 get_driving_distance 函数，集成速率限制
def get_driving_distance_with_rate_limit(lat1, lon1, lat2, lon2):
    """
    通过OSRM API计算两点间的驾驶距离，并集成速率限制。
    """
    with rate_limited(limit=5, delay=1): # 示例：为演示目的将限制设低，实际使用请根据API规则调整
        return get_driving_distance(lat1, lon1, lat2, lon2)

注意事项： 在实际应用中，limit 和 delay 参数应根据所使用的API服务提供商的具体速率限制策略进行调整。OSRM的公共实例通常有非常宽松的限制，但商业API服务通常会更严格。

5. 数据整合与结果输出：构建Pandas DataFrame

最终目标是将原始的经纬度数据与计算出的驾驶距离整合到一个Pandas DataFrame中，便于后续分析。

# 示例数据
location_latitudes = [34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522,
                      34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522,
                      34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522,
                      34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522,
                      34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522,
                      34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522] # 60个点
location_longitudes = [-118.2437] * 60
station_latitudes = [34.0689 + i*0.001 for i in range(60)]
station_longitudes = [-118.2942 + i*0.001 for i in range(60)]

# 用于存储结果的列表
distances = []

# 迭代所有点对并计算距离
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_with_rate_limit(lat1, lon1, lat2, lon2)
    distances.append(distance)

# 创建DataFrame
data = {
    'Location_Latitude': location_latitudes,
    'Location_Longitude': location_longitudes,
    'Station_Latitude': station_latitudes,
    'Station_Longitude': station_longitudes,
    'Driving_Distance_Miles': distances
}
df = pd.DataFrame(data)

print("\n生成的DataFrame:")
print(df.head())
print(f"\nDataFrame包含 {len(df)} 条记录。")

6. 完整示例与注意事项

以下是包含所有组件的完整代码示例，便于读者直接运行和理解：

import requests
import json
import pandas as pd
from time import sleep
from contextlib import contextmanager

# 1. 全局API调用计数器
api_calls_counter = 0

# 2. 速率限制上下文管理器
@contextmanager
def rate_limited(limit=500, delay=5):
    global api_calls_counter
    if api_calls_counter + 1 >= limit:
        print(f"达到 {limit} 次API调用限制，暂停 {delay} 秒...")
        sleep(delay)
        api_calls_counter = 0 # 重置计数器
        print("暂停结束，继续调用。")

    api_calls_counter += 1
    yield

# 3. 核心API请求函数
def get_driving_distance(lat1, lon1, lat2, lon2):
    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() # 检查HTTP状态码

        routes = json.loads(r.content)

        if routes and "routes" in routes and len(routes["routes"]) > 0:
            route_info = routes["routes"][0]
            driving_distance_meters = route_info.get('distance')
            if driving_distance_meters is not None:
                return driving_distance_meters / 1609.34
        else:
            print(f"API响应未包含有效路线信息: {url}, 响应: {routes}")
            return None

    except requests.exceptions.HTTPError as e:
        print(f"HTTP错误: {e.response.status_code} - {e.response.text} for URL: {url}")
        return None
    except requests.exceptions.ConnectionError as e:
        print(f"连接错误: {e} for URL: {url}")
        return None
    except requests.exceptions.Timeout as e:
        print(f"请求超时: {e} for URL: {url}")
        return None
    except json.JSONDecodeError as e:
        print(f"JSON解析错误: {e} for URL: {url}, 响应内容: {r.content}")
        return None
    except Exception as e:
        print(f"发生未知错误: {e} for URL: {url}")
        return None

# 4. 集成速率限制的API请求函数
def get_driving_distance_with_rate_limit(lat1, lon1, lat2, lon2):
    # 注意：这里的limit和delay可以根据实际API限制和测试需求调整
    # 为演示目的，我们设置一个较低的限制（例如，每5次调用暂停1秒）
    with rate_limited(limit=5, delay=1): 
        return get_driving_distance(lat1, lon1, lat2, lon2)

# 5. 示例数据（增加数据量以触发速率限制）
location_latitudes = [34.0522] * 20
location_longitudes = [-118.2437] * 20
station_latitudes = [34.0689 + i * 0.001 for i in range(20)]
station_longitudes = [-118.2942 + i * 0.001 for i in range(20)]

# 6. 计算并整合结果
distances = []
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_with_rate_limit(lat1, lon1, lat2, lon2)
    distances.append(distance)

# 7. 构建DataFrame
data = {
    'Location_Latitude': location_latitudes,
    'Location_Longitude': location_longitudes,
    'Station_Latitude': station_latitudes,
    'Station_Longitude': station_longitudes,
    'Driving_Distance_Miles': distances
}
df = pd.DataFrame(data)

print("\n--- 结果DataFrame ---")
print(df.head(10)) # 打印前10行
print(f"\nDataFrame包含 {len(df)} 条记录。")

重要注意事项：

API密钥： 某些商业API服务（如Google Maps API）需要API密钥进行认证。如果使用此类服务，请确保将密钥安全地包含在请求中（通常通过URL参数或HTTP头）。OSRM公共实例通常不需要密钥。
错误处理： 尽管我们已经添加了基本的错误处理，但生产级的应用程序可能需要更复杂的重试逻辑（例如，指数退避）来处理瞬时网络问题或API服务暂时性错误。
数据验证： 在将经纬度传递给API之前，最好进行数据验证，确保它们是有效的浮点数且在合理范围内。
并发： 对于非常大的数据集，可以考虑使用多线程或异步IO（如asyncio配合aiohttp）来并行化API请求，但这会使速率限制逻辑变得更复杂，需要更精细的控制。
OSRM服务： 本教程使用的是公共OSRM路由服务。对于生产环境或高并发需求，建议搭建自己的OSRM服务器或使用商业地理服务提供商。

7. 总结

本教程详细介绍了如何使用Python处理多源地理位置数据，通过OSRM API计算驾驶距离，并重点演示了如何利用contextlib.contextmanager实现优雅且健壮的API请求速率限制。通过将原始数据、计算逻辑和结果整合到Pandas DataFrame中，我们能够高效地管理和分析地理空间数据。掌握这些技术将有助于您在处理大规模地理数据时，构建更稳定、

Python 3 中实现字节串格式化：替代 %b 的标准方法

Python 3 中如何为内置类型生成字节串格式化结果

Python 3 中字节串格式化：如何正确将内置类型转换为 bytes

如何在 Windows 11 上成功安装 TA-Lib（Python 版）

如何在Python中正确获取类实例变量的最新值