当前位置：首页 > news >正文

高德地图 Web 服务 API V3 公交线路爬取：Python 脚本 5 分钟获取 1000 条线路坐标

news 2026/7/13 15:20:37

高德地图公交数据采集实战：Python自动化获取千条线路坐标

1. 高德地图API与公交数据价值

公交线路数据作为城市交通网络的核心要素，在城市规划、交通分析和商业选址等领域具有重要价值。高德地图作为国内领先的数字地图服务提供商，其Web服务API V3版本为开发者提供了丰富的公交数据接口，包括线路查询、站点信息和路径坐标等关键数据。

通过高德API获取公交数据相比传统爬虫具有显著优势：

数据权威性：直接来自高德官方数据源，准确性有保障
接口稳定性：企业级API服务，可用性高达99.9%
格式标准化：返回统一的JSON数据结构，便于后续处理
更新及时性：与高德主站数据同步更新

典型应用场景包括：

城市公交网络密度分析
公共交通可达性研究
商业网点选址评估
智慧城市交通规划

2. 环境准备与API配置

2.1 开发环境搭建

推荐使用Python 3.8+版本进行开发，主要依赖库包括：

# 安装核心依赖库 pip install requests pandas numpy

2.2 API密钥申请流程

访问高德开放平台官网，注册开发者账号
进入控制台创建新应用
获取Web服务API的Key和安全密钥
设置IP白名单（建议配置服务器出口IP）

注意：免费版API有每日调用限额，商业项目建议购买企业级服务套餐

2.3 基础请求示例

import requests import json def get_bus_line(city, line_name, api_key): url = f"https://restapi.amap.com/v3/bus/linename?key={api_key}&city={city}&keywords={line_name}" response = requests.get(url) return json.loads(response.text) # 示例调用 api_key = "your_api_key_here" result = get_bus_line("北京", "1路", api_key) print(result)

3. 数据采集策略实现

3.1 循环遍历采集法

适用于获取城市主要公交线路的基础方案：

def batch_fetch_lines(city, start_line, end_line, api_key): lines_data = [] for line_num in range(start_line, end_line + 1): line_name = f"{line_num}路" try: data = get_bus_line(city, line_name, api_key) if data["buslines"]: lines_data.append(process_line_data(data)) except Exception as e: print(f"获取线路{line_name}失败: {str(e)}") time.sleep(0.1) # 控制请求频率 return pd.DataFrame(lines_data)

参数优化建议：

城市核心区建议采集1-999路
郊区线路可扩展至1000-1999范围
特殊线路（如夜班车）需单独处理

3.2 文本导入采集法

针对特定线路列表的高精度采集方案：

从公交信息网站（如8684）导出目标线路清单
保存为bus_lines.txt，每行一个线路名
使用以下代码精准采集：

def fetch_by_list(city, file_path, api_key): with open(file_path, "r", encoding="utf-8") as f: lines = [line.strip() for line in f.readlines()] results = [] for line_name in lines: try: data = get_bus_line(city, line_name, api_key) if data["buslines"]: results.append(process_line_data(data)) except Exception as e: print(f"获取线路{line_name}失败: {str(e)}") return pd.DataFrame(results)

4. 数据解析与格式化处理

4.1 坐标数据提取

高德API返回的polyline字段包含线路完整坐标串，需要特殊处理：

def parse_polyline(poly_str): """将高德polyline字符串转换为坐标列表""" points = [] segments = poly_str.split(";") for seg in segments: lon, lat = seg.split(",") points.append((float(lon), float(lat))) return points

4.2 数据结构化存储

建议将采集结果存储为CSV和GeoJSON两种格式：

def save_results(df, city): # CSV格式保存 csv_file = f"{city}_bus_lines.csv" df.to_csv(csv_file, index=False, encoding="utf-8-sig") # GeoJSON格式转换 geojson = { "type": "FeatureCollection", "features": [ { "type": "Feature", "geometry": { "type": "LineString", "coordinates": line["coordinates"] }, "properties": { "name": line["name"], "city": city } } for line in df.to_dict("records") ] } with open(f"{city}_bus_lines.geojson", "w") as f: json.dump(geojson, f, ensure_ascii=False)

4.3 数据质量校验

采集过程中应包含数据验证环节：

def validate_line_data(line_data): required_fields = ["name", "polyline", "start_stop", "end_stop"] if not all(field in line_data for field in required_fields): return False if len(line_data["polyline"].split(";")) < 2: return False return True

5. 性能优化与异常处理

5.1 请求频率控制

为避免触发API限流，建议实现智能请求控制：

class RateLimiter: def __init__(self, max_calls, period): self.max_calls = max_calls self.period = period self.timestamps = [] def __call__(self): now = time.time() self.timestamps = [t for t in self.timestamps if t > now - self.period] if len(self.timestamps) >= self.max_calls: sleep_time = self.period - (now - self.timestamps[0]) time.sleep(sleep_time) self.timestamps.append(time.time()) # 使用示例（每秒不超过5次请求） limiter = RateLimiter(5, 1) for line in lines: limiter() fetch_line_data(line)

5.2 异常重试机制

from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) def fetch_with_retry(city, line_name, api_key): return get_bus_line(city, line_name, api_key)

5.3 结果缓存策略

对于大规模采集任务，建议实现本地缓存：

import os from hashlib import md5 def get_cache_key(city, line_name): return md5(f"{city}_{line_name}".encode()).hexdigest() def cached_fetch(city, line_name, api_key, cache_dir="cache"): os.makedirs(cache_dir, exist_ok=True) cache_file = os.path.join(cache_dir, get_cache_key(city, line_name)) if os.path.exists(cache_file): with open(cache_file, "r") as f: return json.load(f) data = get_bus_line(city, line_name, api_key) with open(cache_file, "w") as f: json.dump(data, f) return data

6. 数据可视化应用

虽然高德Map Lab已下线，但可通过以下替代方案实现数据可视化：

6.1 使用GeoHUB可视化平台

将采集的GeoJSON数据上传至GeoHUB数据中心
创建可视化项目并配置图层样式
发布为可嵌入的Web组件

6.2 Python本地可视化方案

import folium def plot_lines(city, lines_df): # 创建基础地图 m = folium.Map(location=[39.9, 116.4], zoom_start=12) # 添加公交线路 for _, row in lines_df.iterrows(): folium.PolyLine( locations=row["coordinates"], color="blue", weight=2, popup=row["name"] ).add_to(m) # 保存为HTML m.save(f"{city}_bus_map.html")