Python实战：如何用高德API批量获取POI数据（附完整代码与避坑指南）

Python实战：如何用高德API批量获取POI数据（附完整代码与避坑指南）

在地理信息数据驱动的今天，兴趣点（POI）数据已成为商业分析、城市规划、市场研究乃至个人项目开发不可或缺的燃料。无论是想分析一个城市的餐饮业态分布，还是为你的智能推荐系统寻找线下实体锚点，批量、精准地获取POI数据都是第一步。高德地图作为国内领先的地图服务商，其开放的Web服务API为我们提供了强大的数据接口。然而，从“知道有这个接口”到“稳定、高效地拿到想要的数据”，中间隔着一道由API限制、坐标系谜团和数据处理细节构成的鸿沟。这篇文章，我将以一个实战者的身份，带你手把手跨越这道鸿沟，不仅提供能直接运行的代码，更会深入剖析那些官方文档里不会明说，却能让你的项目中途“翻车”的关键细节。

1. 前期准备：理解游戏规则与备好入场券

在开始敲代码之前，我们必须先理解高德POI API的“游戏规则”。这绝非简单的技术调用，更像是一场与数据提供方约定俗成的对话。高德的“地点搜索”API（/v3/place/text）是其POI服务的核心，它允许我们通过关键词、类型和城市范围来检索数据。但这里有几个至关重要的前提，直接决定了你后续代码的成败。

首先，密钥（Key）是你的唯一身份凭证。没有它，一切免谈。申请过程在高德开放平台完成，相对简单，但有几个细节需要注意：

• 选择应用类型：对于个人学习或测试，选择“Web服务”即可。如果用于正式项目，请根据实际情况选择。
• 设置IP白名单：这是提升安全性的好习惯。如果你在固定IP的服务器上运行脚本，强烈建议设置。若在本地调试，可暂时设置为0.0.0.0/0（允许所有IP），但上线前务必修改。
• 配额限制：免费版有每日调用次数上限。批量获取时，务必规划好你的调用节奏，避免短时间内触发限流。可以考虑在代码中加入time.sleep()来控制请求频率。

其次，要深刻理解API的核心参数逻辑。keywords和types参数至少需要填写一个。types参数基于高德的一套分类编码，更为精确。例如，"050000"代表餐饮服务，"060000"代表购物服务。使用types能避免关键词歧义，是更推荐的方式。city参数限定了搜索的行政区域，这对于控制数据范围和规避“900条上限”问题至关重要。

提示：高德的分类编码表可以在其官方文档中下载。在代码中预定义一个分类字典，能让你的脚本更加灵活和可维护。

最后，关于输出格式，output参数默认为JSON，这也是我们处理起来最方便的形式。整个前期准备，可以总结为以下关键动作清单：

1. 访问高德开放平台，注册开发者账号。
2. 创建新应用，获取专属的API Key。
3. 仔细阅读“地点搜索”API文档，重点关注请求参数说明和返回参数示例。
4. 根据你的目标数据（如“上海市的咖啡馆”），确定使用keywords（“咖啡馆”）还是types（查找餐饮服务下的具体分类编码），并确定city（“上海”）。

2. 核心实战：构建稳健的数据抓取流水线

有了Key和理解，我们就可以开始构建自动化的数据抓取流水线了。这个流水线需要稳健地处理网络请求、分页逻辑、数据解析和错误应对。我将采用模块化的函数式设计，让代码清晰且易于复用。

2.1 构造基础请求函数

这个函数负责与高德API进行一次对话。我们需要处理中文字符的URL编码，并设置合理的请求头。

import requests import json from urllib.parse import quote import time def fetch_poi_page(api_key, city, types, page=1, extensions='base'): """ 获取单页POI数据。 Args: api_key (str): 高德API密钥。 city (str): 城市名称，如“北京市”。 types (str): POI类型编码，如“050000”（餐饮）。 page (int): 页码，从1开始。 extensions (str): 返回结果详略，'base'（基础）或'all'（全部）。 Returns: dict: 解析后的JSON数据字典。如果请求失败，返回None。 """ # 基础URL base_url = "https://restapi.amap.com/v3/place/text" # 参数编码与构造查询字符串 params = { 'key': api_key, 'city': quote(city), # 对中文城市名进行URL编码 'types': types, 'page': page, 'offset': 20, # 每页记录数，最大为25 'extensions': extensions, 'output': 'json' } # 设置请求头，模拟浏览器行为 headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' } try: response = requests.get(base_url, params=params, headers=headers, timeout=10) response.raise_for_status() # 检查HTTP请求是否成功 data = response.json() # 检查API返回状态 if data.get('status') == '1': return data else: print(f"请求失败: {data.get('info', '未知错误')}, 请求参数: {params}") return None except requests.exceptions.RequestException as e: print(f"网络请求异常: {e}") return None except json.JSONDecodeError as e: print(f"JSON解析异常: {e}") return None

这个函数加入了异常处理和状态检查，比简单的requests.get更加健壮。quote(city)确保了中文城市名在URL中正确传输。

2.2 实现分页抓取与总量控制

高德API默认每页最多返回25条记录（通过offset参数设置），并且对于关键词搜索，存在单次查询最多返回900条记录的限制。这是第一个大“坑”。我们的策略是：循环请求，直到当前页返回的POI数量为0。

def fetch_all_pois(api_key, city, types, max_pages=50): """ 抓取指定城市和类型的所有POI（受API900条限制）。 Args: api_key (str): 高德API密钥。 city (str): 城市名称。 types (str): POI类型编码。 max_pages (int): 最大抓取页数，防止意外无限循环。 Returns: list: 所有POI记录的列表。 """ all_pois = [] page = 1 while page <= max_pages: print(f"正在抓取 {city} - {types} 第 {page} 页...") data = fetch_poi_page(api_key, city, types, page) if data is None: print("获取数据失败，终止抓取。") break pois = data.get('pois', []) if not pois: # 当前页没有数据，说明已抓取完毕 print(f"第 {page} 页无数据，抓取结束。") break all_pois.extend(pois) print(f" 本页获取 {len(pois)} 条，累计 {len(all_pois)} 条。") # 检查是否可能达到900条上限（估算） total_count = int(data.get('count', '0')) if len(all_pois) >= total_count and total_count > 0: print(f"已获取所有 {total_count} 条数据。") break if len(all_pois) >= 900: print("警告：已触及或接近900条关键词搜索上限。建议使用矩形区域分割法。") # 这里可以加入逻辑，触发更细粒度的区域划分 break page += 1 time.sleep(0.2) # 礼貌性延时，避免请求过快 return all_pois

这里有几个关键点：

1. 循环终止条件：当API返回的pois列表为空时，说明没有更多数据。
2. 900条上限预警：当累计数据量达到900条时，发出警告。这是关键词搜索法的固有瓶颈。
3. 延时time.sleep：这是尊重服务方、避免IP被临时封锁的良好实践。0.2秒是一个比较保守的值。

2.3 破解900条限制：从“关键词搜索”到“区域网格扫描”

如果你需要获取一个区域内超过900个的同类POI（例如“上海市的全部便利店”），关键词搜索法就力不从心了。此时，必须切换思路，采用矩形区域搜索法。其核心思想是：将目标大区域（如整个上海）划分为多个不重叠的小矩形网格，然后对每个小网格分别调用周边搜索API(/v3/place/around) 或多边形搜索API(/v3/place/polygon)，从而绕过单个关键词查询的总数限制。

这涉及到地理区域划分的算法。一个简单有效的策略是使用经纬度网格划分。

def generate_grid_rectangles(bounds, grid_size_km=2): """ 根据地理矩形边界，生成指定边长的网格矩形列表。 Args: bounds (dict): 包含'southwest'和'northeast'键的字典，每个键是一个包含经度(lng)和纬度(lat)的列表。 例如: {'southwest': [121.0, 31.0], 'northeast': [122.0, 32.0]} grid_size_km (float): 网格边长，单位公里。 Returns: list: 每个元素是一个字典，表示一个网格的西南角(southwest)和东北角(northeast)。 """ from math import cos, radians sw_lng, sw_lat = bounds['southwest'] ne_lng, ne_lat = bounds['northeast'] # 将公里转换为经纬度（近似）。纬度1度约111公里，经度1度随纬度变化。 lat_step = grid_size_km / 111.0 # 在中间纬度处计算经度步长 mid_lat = (sw_lat + ne_lat) / 2 lng_step = grid_size_km / (111.0 * cos(radians(mid_lat))) rectangles = [] current_lat = sw_lat while current_lat < ne_lat: current_lng = sw_lng next_lat = min(current_lat + lat_step, ne_lat) while current_lng < ne_lng: next_lng = min(current_lng + lng_step, ne_lng) grid_sw = [current_lng, current_lat] grid_ne = [next_lng, next_lat] rectangles.append({'southwest': grid_sw, 'northeast': grid_ne}) current_lng = next_lng current_lat = next_lat print(f"生成了 {len(rectangles)} 个网格。") return rectangles

获取到网格列表后，你需要修改抓取函数，将针对每个网格的中心点，调用周边搜索API（/v3/place/around），并设置合适的半径（radius，单位米）。这样，每个网格独立查询，其返回数量限制只针对该网格内，从而实现了对大面积区域的完整覆盖。这是处理大规模POI采集的标准进阶方案。

3. 数据处理：坐标系转换与数据清洗

从高德API获取的POI数据并不能直接用于所有场景，最大的“坑”之一就是坐标系。

3.1 理解坐标系差异

高德地图、腾讯地图等国内服务使用的是一种被称为GCJ-02（俗称“火星坐标系”）的加密坐标系。这是一种在国家测绘局制定的WGS-84坐标系基础上，加入非线性偏移后的坐标系。而国际上通用、以及GPS设备直接采集的坐标，通常是WGS-84坐标系。如果你需要将高德的POI位置与GPS轨迹、或者某些国际标准的地理工具（如Google Earth）一起使用，就必须进行坐标转换。

3.2 实现GCJ-02向WGS-84的转换

转换算法涉及复杂的数学公式。以下是经过验证的GCJ-02转WGS-84的Python函数：

import math def gcj02_to_wgs84(lng, lat): """ 将GCJ-02坐标系（高德、腾讯）转换为WGS-84坐标系。 Args: lng (float): GCJ-02经度。 lat (float): GCJ-02纬度。 Returns: tuple: (WGS-84经度, WGS-84纬度)。 """ if not _is_in_china(lng, lat): return lng, lat PI = 3.1415926535897932384626 ee = 0.00669342162296594323 a = 6378245.0 dlat = _transform_lat(lng - 105.0, lat - 35.0) dlng = _transform_lng(lng - 105.0, lat - 35.0) radlat = lat / 180.0 * PI magic = math.sin(radlat) magic = 1 - ee * magic * magic sqrtmagic = math.sqrt(magic) dlat = (dlat * 180.0) / ((a * (1 - ee)) / (magic * sqrtmagic) * PI) dlng = (dlng * 180.0) / (a / sqrtmagic * math.cos(radlat) * PI) mglat = lat + dlat mglng = lng + dlng # 逆向计算 return lng * 2 - mglng, lat * 2 - mglat def _is_in_china(lng, lat): """简单判断坐标是否在中国大陆范围内（粗略）。""" return 72.004 <= lng <= 137.8347 and 0.8293 <= lat <= 55.8271 def _transform_lat(lng, lat): PI = 3.1415926535897932384626 ret = -100.0 + 2.0 * lng + 3.0 * lat + 0.2 * lat * lat + 0.1 * lng * lat + 0.2 * math.sqrt(abs(lng)) ret += (20.0 * math.sin(6.0 * lng * PI) + 20.0 * math.sin(2.0 * lng * PI)) * 2.0 / 3.0 ret += (20.0 * math.sin(lat * PI) + 40.0 * math.sin(lat / 3.0 * PI)) * 2.0 / 3.0 ret += (160.0 * math.sin(lat / 12.0 * PI) + 320 * math.sin(lat * PI / 30.0)) * 2.0 / 3.0 return ret def _transform_lng(lng, lat): PI = 3.1415926535897932384626 ret = 300.0 + lng + 2.0 * lat + 0.1 * lng * lng + 0.1 * lng * lat + 0.1 * math.sqrt(abs(lng)) ret += (20.0 * math.sin(6.0 * lng * PI) + 20.0 * math.sin(2.0 * lng * PI)) * 2.0 / 3.0 ret += (20.0 * math.sin(lng * PI) + 40.0 * math.sin(lng / 3.0 * PI)) * 2.0 / 3.0 ret += (150.0 * math.sin(lng / 12.0 * PI) + 300.0 * math.sin(lng / 30.0 * PI)) * 2.0 / 3.0 return ret

在数据保存前，对每个POI的location字段调用此函数进行转换。

3.3 数据清洗与结构化保存

从API获取的原始JSON数据包含很多字段，我们需要提取并清洗出有用的信息。使用pandas库会让这个过程异常简洁高效。

import pandas as pd def clean_and_save_poi_data(poi_list, output_filename='poi_data.csv', convert_coord=True): """ 清洗POI列表并保存为CSV文件。 Args: poi_list (list): 原始的POI字典列表。 output_filename (str): 输出文件名。 convert_coord (bool): 是否进行GCJ02到WGS84的坐标转换。 """ if not poi_list: print("POI列表为空，无需保存。") return cleaned_data = [] for poi in poi_list: # 提取核心字段 name = poi.get('name', '') address = poi.get('address', '') adname = poi.get('adname', '') # 区域名称，如“浦东新区” location_str = poi.get('location', '') # 格式："121.123456,31.123456" # 处理坐标 if location_str: lng_gcj02, lat_gcj02 = map(float, location_str.split(',')) if convert_coord: lng_wgs84, lat_wgs84 = gcj02_to_wgs84(lng_gcj02, lat_gcj02) else: lng_wgs84, lat_wgs84 = lng_gcj02, lat_gcj02 else: lng_wgs84 = lat_wgs84 = None # 构建清洗后的数据行 cleaned_row = { 'name': name, 'address': address, 'district': adname, 'longitude': lng_wgs84, 'latitude': lat_wgs84, 'type': poi.get('type', ''), 'tel': poi.get('tel', ''), 'id': poi.get('id', '') } cleaned_data.append(cleaned_row) # 转换为DataFrame并保存 df = pd.DataFrame(cleaned_data) # 去重（根据ID或名称+地址） df.drop_duplicates(subset=['id'], inplace=True, ignore_index=True) df.to_csv(output_filename, index=False, encoding='utf-8-sig') # utf-8-sig支持Excel直接打开 print(f"数据已清洗并保存至 {output_filename}， 共 {len(df)} 条记录。") # 可选：打印简要统计信息 print("\n数据预览：") print(df.head()) print(f"\n各区域POI数量统计：") print(df['district'].value_counts().head())

使用pandas不仅简化了数据操作，其DataFrame结构也便于后续进行更复杂的数据分析和可视化。

4. 工程化与最佳实践：让脚本健壮如牛

一个能在生产环境运行的数据抓取脚本，绝不能只是实验室里的玩具。我们需要考虑错误恢复、日志记录、配置管理和性能优化。

4.1 错误处理与重试机制

网络请求天生不稳定，API也可能返回临时错误。我们必须为关键操作添加重试机制。

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import requests.exceptions @retry( stop=stop_after_attempt(3), # 最多重试3次 wait=wait_exponential(multiplier=1, min=2, max=10), # 指数退避等待 retry=retry_if_exception_type((requests.exceptions.ConnectionError, requests.exceptions.Timeout)) ) def robust_fetch_page(api_key, city, types, page): """带有重试机制的请求函数""" return fetch_poi_page(api_key, city, types, page)

这里使用了tenacity这个优秀的重试库，它让代码非常清晰。我们只对网络连接超时类错误进行重试，对于API返回的业务错误（如status不为‘1’）则立即失败。

4.2 配置与密钥管理

永远不要将API密钥硬编码在脚本中！最佳实践是使用环境变量或配置文件。

config.yaml

amap: api_key: "你的高德API密钥" default_city: "上海市" default_types: "050000" request_timeout: 10 sleep_interval: 0.2

脚本中读取配置

import yaml import os def load_config(config_path='config.yaml'): with open(config_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) # 环境变量优先级最高 config['amap']['api_key'] = os.getenv('AMAP_API_KEY', config['amap']['api_key']) return config

4.3 完整脚本组装与运行示例

将以上所有模块组合起来，并添加主程序逻辑。

def main(): """主函数""" # 1. 加载配置 config = load_config() api_key = config['amap']['api_key'] # 2. 定义抓取任务 tasks = [ {'city': '杭州市', 'types': '060100'}, # 购物-商场 {'city': '深圳市南山区', 'types': '050000'}, # 餐饮服务 ] all_pois_for_export = [] # 3. 遍历任务执行抓取 for task in tasks: city = task['city'] types = task['types'] print(f"\n开始抓取任务: {city} - {types}") pois = fetch_all_pois(api_key, city, types) if pois: # 为每条数据添加任务标签 for poi in pois: poi['_task_city'] = city poi['_task_type'] = types all_pois_for_export.extend(pois) print(f"任务完成，获取 {len(pois)} 条POI。") else: print(f"任务失败，未获取到数据。") # 4. 统一清洗和保存所有数据 if all_pois_for_export: output_file = f"poi_export_{pd.Timestamp.now().strftime('%Y%m%d_%H%M%S')}.csv" clean_and_save_poi_data(all_pois_for_export, output_file, convert_coord=True) else: print("未获取到任何有效数据，程序结束。") if __name__ == '__main__': main()

运行这个脚本，你将得到一个带有时间戳的CSV文件，里面包含了清洗过、坐标系转换后的规整POI数据，可以直接导入数据库或用于分析。

在整个实践过程中，我最大的体会是，数据获取的稳定性往往比功能的复杂性更重要。那些关于配额、限流、坐标系和网络异常的“坑”，需要你在代码层面提前做好防御。与其追求一次性抓取百万数据，不如先构建一个能稳定运行、易于监控和重启的小而美的流水线。当你的脚本能在无人值守的情况下，安静地完成一夜的数据采集任务时，那种可靠性带来的安心感，远比炫技的代码更有价值。