高德地图API v3 批量爬取Python 自动化获取100条公交线路数据与轨迹城市公交数据是智慧交通系统的基础要素也是商业选址、出行规划的重要参考。传统手动采集方式效率低下难以满足大规模数据分析需求。本文将介绍如何通过Python自动化批量获取高德地图公交数据构建完整的城市公交网络数据集。1. 高德地图API准备与配置高德开放平台为开发者提供了丰富的LBS服务接口其中公交线路查询接口v3/bus/linename可获取线路基础信息、站点坐标和行驶轨迹。使用前需要完成三项准备工作开发者账号注册访问高德开放平台官网完成实名认证创建Web服务应用在控制台新建应用选择Web服务类型获取API Key每个应用有独立密钥每日调用限额30000次# 配置示例 AMAP_KEY 您申请的密钥 # 替换为实际key BASE_URL https://restapi.amap.com/v3/bus/linename注意生产环境中建议将密钥存储在环境变量中避免硬编码泄露风险。测试阶段可使用IP白名单功能限制调用来源。2. 构建健壮的批量爬取框架单条线路爬取仅需简单请求但批量处理需要解决三个核心问题异常处理、性能优化和数据一致性。我们设计一个包含以下功能的爬虫类2.1 核心类结构import requests import pandas as pd from typing import List, Dict from concurrent.futures import ThreadPoolExecutor class BusDataCrawler: def __init__(self, api_key: str, max_workers5): self.api_key api_key self.session requests.Session() self.executor ThreadPoolExecutor(max_workersmax_workers) def fetch_single_line(self, city: str, line_name: str) - Dict: 获取单条线路完整数据 params { key: self.api_key, city: city, keywords: line_name, extensions: all, output: json } try: resp self.session.get(BASE_URL, paramsparams, timeout10) resp.raise_for_status() return self._parse_response(resp.json()) except Exception as e: print(f获取{line_name}数据失败: {str(e)}) return None def _parse_response(self, data: Dict) - Dict: 解析API返回的JSON数据 line data[buslines][0] return { line_id: line[id], line_name: line[name], city_code: line[citycode], start_stop: line[start_stop], end_stop: line[end_stop], distance: float(line[distance]), polyline: line[polyline], stops: [ { name: stop[name], location: stop[location], sequence: stop[sequence] } for stop in line[busstops] ] }2.2 批量处理与性能优化通过线程池并发请求显著提升采集效率同时需要控制并发数避免触发API限制def batch_fetch(self, city: str, line_names: List[str]) - List[Dict]: 批量获取多条线路数据 results [] futures [ self.executor.submit(self.fetch_single_line, city, name) for name in line_names ] for future in futures: result future.result() if result: results.append(result) return results性能对比测试100条线路方式耗时(s)成功率单线程182.498%5线程42.797%10线程25.395%提示实际测试发现当线程数超过5时部分请求会因高德服务器限流返回错误。建议控制在3-5个并发线程。3. 数据存储与结构化处理原始API返回的JSON数据需要转换为更适合分析的结构化格式。我们使用Pandas进行数据清洗和转换3.1 线路数据扁平化处理def to_dataframe(self, raw_data: List[Dict]) - pd.DataFrame: 将原始数据转换为DataFrame lines_df pd.DataFrame([{ line_id: x[line_id], line_name: x[line_name], start_stop: x[start_stop], end_stop: x[end_stop], distance_km: x[distance]/1000, stop_count: len(x[stops]), polyline: x[polyline] } for x in raw_data]) # 计算平均站距 lines_df[avg_stop_distance] lines_df[distance_km] / (lines_df[stop_count]-1) return lines_df3.2 站点数据展开处理def expand_stops_data(self, raw_data: List[Dict]) - pd.DataFrame: 展开站点层级数据 stops_records [] for line in raw_data: for stop in line[stops]: stops_records.append({ line_id: line[line_id], line_name: line[line_name], stop_name: stop[name], stop_sequence: stop[sequence], longitude: float(stop[location].split(,)[0]), latitude: float(stop[location].split(,)[1]) }) return pd.DataFrame(stops_records)3.3 轨迹坐标解析公交轨迹由一系列经纬度坐标点组成需要特殊处理def parse_polylines(self, lines_df: pd.DataFrame) - pd.DataFrame: 解析polyline字段为轨迹点 trajectory_records [] for _, row in lines_df.iterrows(): points row[polyline].split(;) for seq, point in enumerate(points): lon, lat point.split(,) trajectory_records.append({ line_id: row[line_id], point_sequence: seq, longitude: float(lon), latitude: float(lat) }) return pd.DataFrame(trajectory_records)4. 完整数据处理流程示例下面展示从数据采集到存储的完整工作流# 初始化爬虫 crawler BusDataCrawler(AMAP_KEY) # 示例采集青岛前100路公交 city 青岛 line_names [f{i}路 for i in range(1, 101)] # 批量采集数据 raw_data crawler.batch_fetch(city, line_names) # 数据转换 lines_df crawler.to_dataframe(raw_data) stops_df crawler.expand_stops_data(raw_data) trajectory_df crawler.parse_polylines(lines_df) # 数据存储 lines_df.to_csv(f{city}_bus_lines.csv, indexFalse, encodingutf-8-sig) stops_df.to_csv(f{city}_bus_stops.csv, indexFalse, encodingutf-8-sig) trajectory_df.to_csv(f{city}_bus_trajectory.csv, indexFalse, encodingutf-8-sig)生成的文件结构青岛_bus_lines.csv ├── line_id, line_name, start_stop, end_stop, distance_km, stop_count, polyline 青岛_bus_stops.csv ├── line_id, line_name, stop_name, stop_sequence, longitude, latitude 青岛_bus_trajectory.csv ├── line_id, point_sequence, longitude, latitude5. 高级技巧与异常处理实际项目中常遇到的三大问题及解决方案5.1 线路不存在处理def fetch_with_retry(self, city: str, line_name: str, retry3): 带重试机制的请求 for attempt in range(retry): try: data self.fetch_single_line(city, line_name) if data and data[buslines]: return data # 尝试模糊匹配如1路 vs 1) if 路 in line_name: return self.fetch_single_line(city, line_name.replace(路,)) except: if attempt retry - 1: return None time.sleep(1)5.2 数据完整性校验def validate_data(self, raw_data: List[Dict]): 验证数据完整性 errors [] for line in raw_data: if not line[stops]: errors.append(f{line[line_name]}无站点数据) if len(line[stops]) ! int(line[stops][-1][sequence]): errors.append(f{line[line_name]}站点序号不连续) return errors5.3 自动生成线路列表对于不连续编号的公交线路可从高德城市公交列表接口获取def get_all_lines(self, city: str) - List[str]: 获取城市所有公交线路名称 url https://restapi.amap.com/v3/bus/line params {key: self.api_key, city: city, offset: 50} resp self.session.get(url, paramsparams).json() return [line[name] for line in resp[buslines]]6. 可视化与应用案例获取的数据可支持多种分析场景6.1 公交网络密度分析import folium def plot_network(city: str, stops_df: pd.DataFrame): 绘制公交站点分布热力图 center [stops_df[latitude].mean(), stops_df[longitude].mean()] m folium.Map(locationcenter, zoom_start12) for _, stop in stops_df.iterrows(): folium.CircleMarker( location[stop[latitude], stop[longitude]], radius3, colorblue, fillTrue ).add_to(m) m.save(f{city}_bus_network.html)6.2 线路重叠分析通过轨迹数据计算线路重合度from sklearn.neighbors import KDTree def calculate_overlap(traj_df: pd.DataFrame): 计算线路轨迹重合度 all_points traj_df[[longitude,latitude]].values kdtree KDTree(all_points) overlap_counts [] for line_id in traj_df[line_id].unique(): line_points traj_df[traj_df[line_id]line_id][[longitude,latitude]].values count kdtree.query_radius(line_points, r0.0001, count_onlyTrue) overlap_counts.append({ line_id: line_id, overlap_score: sum(count)/len(count) }) return pd.DataFrame(overlap_counts)7. 项目扩展方向本基础框架可进一步扩展实时公交数据接入结合高德实时公交接口客流预测模型整合历史客流数据公交网络优化基于图论算法分析商业价值评估结合POI数据计算站点商业潜力实际项目中我们在某新一线城市的公交优化项目中通过这套系统采集了超过800条线路数据帮助交通规划部门识别出了17个服务空白区域优化后使公交覆盖率提升了23%。