通过Python实现API请求限速与批量地理距离计算-Python教程-PHP中文网

通过Python实现API请求限速与批量地理距离计算

碧海醫心

发布： 2025-08-11 19:38:16

原创

1060人浏览过

通过Python实现API请求限速与批量地理距离计算

本教程详细介绍了如何使用Python高效且负责任地通过API计算两点间的驾驶距离。文章从基础的API调用函数出发，深入探讨了利用contextlib模块实现API请求限速的策略，以避免因请求频率过高而被服务器拒绝。此外，教程还强调了API响应错误处理的重要性，并提供了将计算结果整合到Pandas DataFrame中的实用方法，旨在帮助开发者构建稳定、高效的地理数据处理应用。

1. 理解地理距离计算与API调用基础

在地理信息处理中，计算两点间的驾驶距离是一个常见需求。通常，我们会利用第三方地图服务的api来获取这些数据。本教程以开源路由服务osrm为例，展示如何通过http请求获取驾驶距离。

一个基本的距离计算函数可能如下所示：

import requests
import json

def get_driving_distance(lat1, lon1, lat2, lon2):
    """
    通过OSRM API计算两点间的驾驶距离。
    参数:
        lat1 (float): 起点纬度
        lon1 (float): 起点经度
        lat2 (float): 终点纬度
        lon2 (float): 终点经度
    返回:
        float: 驾驶距离（英里），如果请求失败则返回None
    """
    try:
        # OSRM API的URL格式为 /route/v1/car/{lon1},{lat1};{lon2},{lat2}
        url = f"http://router.project-osrm.org/route/v1/car/{lon1},{lat1};{lon2},{lat2}?overview=false"
        r = requests.get(url)
        r.raise_for_status() # 检查HTTP状态码，如果不是200则抛出异常
        routes = json.loads(r.content)

        # 确保API返回了有效的路由数据
        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}")
                return None
        else:
            print(f"警告: API响应中未找到有效路由: {url}")
            return None
    except requests.exceptions.RequestException as e:
        print(f"请求API时发生错误: {e}")
        return None
    except json.JSONDecodeError as e:
        print(f"解析API响应时发生错误: {e}")
        return None
    except Exception as e:
        print(f"发生未知错误: {e}")
        return None

# 示例调用
# distance = get_driving_distance(51.5074, 0.1278, 51.5000, 0.1000)
# if distance is not None:
#     print(f"驾驶距离: {distance:.2f} 英里")

登录后复制

在上述代码中，我们引入了requests.raise_for_status()来自动检查HTTP响应状态码，并在出现错误时抛出异常。这是一种处理网络请求错误的重要实践。

2. 实现API请求限速机制

频繁地调用API可能会导致服务器过载，或触发API提供商的速率限制，从而导致请求失败或IP被封禁。为了解决这个问题，我们需要实现一个请求限速机制。Python的contextlib模块提供了一种优雅的方式来创建上下文管理器，非常适合实现这类功能。

我们可以创建一个rate_limited上下文管理器，它会在达到一定请求次数后暂停执行一段时间。

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

from contextlib import contextmanager
from time import sleep
import requests
import json

# 全局变量，用于跟踪API调用次数
api_calls_counter = 0

@contextmanager
def rate_limited(limit=500, delay=5):
    """
    一个上下文管理器，用于限制API调用频率。
    当API调用次数达到指定限制时，暂停指定延迟时间。
    参数:
        limit (int): 在暂停前允许的最大API调用次数。
        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 += 1

    try:
        yield # 执行上下文内的代码
    finally:
        # 可以在这里进行一些清理工作，例如记录日志
        pass

# 重新定义 get_driving_distance 函数，集成限速机制
def get_driving_distance_with_rate_limit(lat1, lon1, lat2, lon2):
    """
    通过OSRM API计算两点间的驾驶距离，并集成请求限速。
    """
    with rate_limited(limit=500, delay=5): # 每500次请求暂停5秒
        try:
            url = f"http://router.project-osrm.org/route/v1/car/{lon1},{lat1};{lon2},{lat2}?overview=false"
            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}")
                    return None
            else:
                print(f"警告: API响应中未找到有效路由: {url}")
                return None
        except requests.exceptions.RequestException as e:
            print(f"请求API时发生错误: {e} (URL: {url})")
            return None
        except json.JSONDecodeError as e:
            print(f"解析API响应时发生错误: {e} (URL: {url}, Response: {r.text[:200]}...)")
            return None
        except Exception as e:
            print(f"发生未知错误: {e} (URL: {url})")
            return None

登录后复制

注意事项：

全局计数器： 示例中使用了一个简单的全局计数器api_calls_counter。在更复杂的应用中，可以考虑使用线程安全的计数器或将计数器封装在类中。
重置逻辑： api_calls_counter = 0 是一个简单的重置方式。如果API有滑动窗口限制（例如，每分钟X次请求），则需要更复杂的逻辑来跟踪请求时间。
错误处理： 即使使用了限速，API请求仍然可能失败（例如，网络问题、服务器内部错误）。因此，保留try-except块进行错误处理至关重要。

3. 批量处理数据并整合到DataFrame

通常，我们会有多组起点和终点坐标需要处理。Pandas库是处理表格数据的强大工具，非常适合将输入数据和计算结果整合到DataFrame中。

算家云

高效、便捷的人工智能算力服务平台

查看详情

假设我们有四个列表，分别存储起点纬度、起点经度、终点纬度、终点经度：

import pandas as pd

# 示例数据
location_latitudes = [51.5074, 51.4998, 51.5123]
location_longitudes = [0.1278, 0.1749, 0.0884]
station_latitudes = [51.5000, 51.4800, 51.5200]
station_longitudes = [0.1000, 0.2000, 0.0500]

# 创建一个DataFrame来存储原始数据和结果
data = {
    'location_lat': location_latitudes,
    'location_lon': location_longitudes,
    'station_lat': station_latitudes,
    'station_lon': station_longitudes
}
df = pd.DataFrame(data)

# 初始化一个空列表来存储计算出的距离
driving_distances = []

# 遍历DataFrame的每一行，调用API计算距离
# 使用zip函数迭代多个列表，或使用df.itertuples()迭代DataFrame行
for index, row in df.iterrows():
    lat1 = row['location_lat']
    lon1 = row['location_lon']
    lat2 = row['station_lat']
    lon2 = row['station_lon']

    distance = get_driving_distance_with_rate_limit(lat1, lon1, lat2, lon2)
    driving_distances.append(distance)

# 将计算出的距离添加到DataFrame的新列中
df['driving_distance_miles'] = driving_distances

print(df)

登录后复制

输出示例 (实际数值会根据API返回而定):

   location_lat  location_lon  station_lat  station_lon  driving_distance_miles
0       51.5074        0.1278       51.500       0.1000                1.503456
1       51.4998        0.1749       51.480       0.2000                1.876543
2       51.5123        0.0884       51.520       0.0500                2.123456

登录后复制