高德地图API V3公交数据采集实战Python批量获取1000线路坐标与站点信息的工程化实践公交数据作为城市交通规划的重要基础信息其采集效率和质量直接影响交通分析的准确性。传统人工采集方式耗时费力而高德地图API V3为开发者提供了高效获取公交数据的官方渠道。本文将深入讲解如何通过Python实现公交数据的自动化采集、清洗与存储构建一套完整的工程化解决方案。1. 高德地图API环境准备与配置在开始采集前需要完成高德开放平台的开发者账号注册和API Key申请。登录高德开放平台后进入控制台-应用管理创建新应用选择Web服务类型获取Key。这个Key将作为所有API请求的身份凭证。关键配置参数说明参数示例值说明keya5b7479db5b24fd68cedcf24f482c156开发者密钥city北京城市编码或中文名称extensionsall返回全部扩展信息outputJSON返回数据格式建议将API Key存储在环境变量中而非代码里避免泄露风险import os from dotenv import load_dotenv load_dotenv() AMAP_KEY os.getenv(AMAP_KEY)对于大规模采集任务需要特别注意高德API的调用限制注意公交线路查询API的免费版每日调用限额为1000次/Key超过限制会导致请求失败。企业级应用建议购买更高配额的服务套餐。2. 公交线路数据采集核心逻辑实现高德地图的公交线路查询接口为/v3/bus/linename通过构造特定参数可以获取线路的详细信息。以下代码展示了如何封装一个健壮的请求函数import requests import json from retrying import retry retry(stop_max_attempt_number3, wait_fixed2000) def fetch_bus_line(city, line_name, api_key): url fhttps://restapi.amap.com/v3/bus/linename?srsv3extensionsallkey{api_key}outputjsoncity{city}keywords{line_name} try: response requests.get(url, timeout10) response.raise_for_status() data response.json() if data[status] 1 and int(data[count]) 0: return data[buslines][0] # 返回第一条匹配线路 return None except Exception as e: print(f请求失败: {str(e)}) raise对于大规模采集我们需要设计批量获取策略。中国城市公交线路编号通常采用数字路的格式如1路可以通过遍历数字编号获取def batch_fetch_lines(city, max_line1000): results [] for i in range(1, max_line 1): line_name f{i}路 line_data fetch_bus_line(city, line_name, AMAP_KEY) if line_data: results.append(parse_line_data(line_data)) print(f已获取 {city}{line_name}) else: print(f{city}{line_name} 不存在) return results异常处理机制设计网络请求重试使用retrying库实现自动重试数据校验检查返回状态码和count字段速率控制添加适当的延时避免触发限流import time def safe_fetch_with_delay(city, line_name, delay0.2): time.sleep(delay) # 请求间隔 return fetch_bus_line(city, line_name, AMAP_KEY)3. 数据结构解析与关键信息提取高德API返回的公交线路数据包含丰富信息需要从中提取关键字段并结构化存储。典型响应数据结构如下{ buslines: [{ name: 1路, id: 110100000000, polyline: 116.404,39.915;116.404,39.916..., busstops: [{ name: 北京西站, location: 116.404,39.915, sequence: 1 }], start_stop: 北京西站, end_stop: 四惠枢纽站, distance: 12.5, start_time: 05:00, end_time: 23:00, company: 北京公交集团, basic_price: 2.0, total_price: 4.0 }] }数据解析函数示例def parse_line_data(line_data): return { line_id: line_data[id], line_name: line_data[name], start_stop: line_data[start_stop], end_stop: line_data[end_stop], distance_km: float(line_data[distance]), polyline: [list(map(float, point.split(,))) for point in line_data[polyline].split(;)], stops: [{ name: stop[name], location: [float(x) for x in stop[location].split(,)], sequence: int(stop[sequence]) } for stop in line_data[busstops]], operation_hours: f{line_data[start_time]}-{line_data[end_time]}, price_info: { basic: line_data.get(basic_price), total: line_data.get(total_price) } }对于大规模数据处理建议使用Pandas进行高效操作import pandas as pd def create_dataframe(lines_data): df pd.DataFrame([{ 线路ID: line[line_id], 线路名称: line[line_name], 站点数: len(line[stops]), 运营里程: line[distance_km], 首班车: line[operation_hours].split(-)[0], 末班车: line[operation_hours].split(-)[1] } for line in lines_data]) return df4. 数据存储方案设计与实现根据数据规模和使用场景可以选择不同的存储方案1. CSV文件存储适合中小规模数据def save_to_csv(lines_data, filename): df create_dataframe(lines_data) df.to_csv(filename, indexFalse, encodingutf-8-sig) print(f数据已保存至 {filename})2. SQLite数据库存储适合本地复杂查询import sqlite3 def init_db(db_pathbus_data.db): conn sqlite3.connect(db_path) c conn.cursor() c.execute(CREATE TABLE IF NOT EXISTS bus_lines (line_id TEXT PRIMARY KEY, name TEXT, start_stop TEXT, end_stop TEXT, distance REAL, stop_count INTEGER)) c.execute(CREATE TABLE IF NOT EXISTS bus_stops (stop_id INTEGER PRIMARY KEY AUTOINCREMENT, line_id TEXT, name TEXT, longitude REAL, latitude REAL, sequence INTEGER, FOREIGN KEY(line_id) REFERENCES bus_lines(line_id))) conn.commit() return conn def save_to_db(conn, lines_data): c conn.cursor() for line in lines_data: # 插入线路数据 c.execute(INSERT OR REPLACE INTO bus_lines VALUES (?, ?, ?, ?, ?, ?), (line[line_id], line[line_name], line[start_stop], line[end_stop], line[distance_km], len(line[stops]))) # 插入站点数据 for stop in line[stops]: c.execute(INSERT INTO bus_stops (line_id, name, longitude, latitude, sequence) VALUES (?, ?, ?, ?, ?), (line[line_id], stop[name], stop[location][0], stop[location][1], stop[sequence])) conn.commit()3. MongoDB存储适合非结构化大数据from pymongo import MongoClient def save_to_mongodb(lines_data, db_namebus_db, collection_namelines): client MongoClient(localhost, 27017) db client[db_name] collection db[collection_name] result collection.insert_many(lines_data) print(f插入了 {len(result.inserted_ids)} 条文档)5. 高级功能扩展与性能优化对于超大城市如北京、上海等公交线路超过1000条的情况需要采用更高效的采集策略1. 多线程采集加速from concurrent.futures import ThreadPoolExecutor def parallel_fetch(city, max_line1000, workers5): with ThreadPoolExecutor(max_workersworkers) as executor: futures [] for i in range(1, max_line 1): line_name f{i}路 futures.append(executor.submit( safe_fetch_with_delay, city, line_name)) results [] for future in futures: try: data future.result() if data: results.append(parse_line_data(data)) except Exception as e: print(f处理失败: {str(e)}) return results2. 断点续采机制import pickle def resume_fetch(city, max_line1000, checkpoint_filecheckpoint.pkl): try: with open(checkpoint_file, rb) as f: completed pickle.load(f) except FileNotFoundError: completed set() results [] for i in range(1, max_line 1): line_name f{i}路 if line_name in completed: continue data safe_fetch_with_delay(city, line_name) if data: results.append(parse_line_data(data)) completed.add(line_name) # 定期保存进度 if i % 100 0: with open(checkpoint_file, wb) as f: pickle.dump(completed, f) return results3. 数据质量校验def validate_line_data(line_data): required_fields [name, id, polyline, busstops] if not all(field in line_data for field in required_fields): return False if len(line_data[busstops]) 2: return False try: float(line_data[distance]) except ValueError: return False return True6. 数据分析与应用场景获取的公交数据可以支持多种分析场景1. 公交网络拓扑分析import networkx as nx def build_bus_network(lines_data): G nx.Graph() for line in lines_data: stops line[stops] for i in range(len(stops)-1): stop1 stops[i][name] stop2 stops[i1][name] G.add_edge(stop1, stop2, lineline[line_name], distanceline[distance_km]/(len(stops)-1)) return G def analyze_network(G): print(f网络包含 {G.number_of_nodes()} 个站点和 {G.number_of_edges()} 条连接) print(站点连接数TOP5:) degree sorted(G.degree(), keylambda x: x[1], reverseTrue) for node, count in degree[:5]: print(f{node}: {count}条连接)2. 站点可达性分析def calculate_accessibility(G, max_transfers2): accessibility {} for node in G.nodes: reachable set() for _, target in nx.bfs_edges(G, sourcenode, depth_limitmax_transfers): reachable.add(target) accessibility[node] len(reachable) return sorted(accessibility.items(), keylambda x: x[1], reverseTrue)3. 可视化展示使用folium库创建交互式地图import folium def visualize_lines(lines_data, city_center, zoom_start12): m folium.Map(locationcity_center, zoom_startzoom_start) for line in lines_data[:20]: # 限制显示数量 points line[polyline] folium.PolyLine( points, colorblue, weight2, opacity0.8, popupf{line[line_name]}: {line[start_stop]}-{line[end_stop]} ).add_to(m) for stop in line[stops]: folium.CircleMarker( locationstop[location], radius3, colorred, fillTrue, fill_colorred, popupstop[name] ).add_to(m) return m7. 工程化部署与自动化将采集脚本工程化可以创建完整的命令行工具import argparse def main(): parser argparse.ArgumentParser(description高德公交数据采集工具) parser.add_argument(city, help城市名称) parser.add_argument(--max-line, typeint, default1000, help最大线路编号) parser.add_argument(--output, choices[csv, sqlite, mongodb], defaultcsv, help输出格式) parser.add_argument(--threads, typeint, default5, help并发线程数) args parser.parse_args() print(f开始采集 {args.city} 公交数据...) if args.threads 1: data parallel_fetch(args.city, args.max_line, args.threads) else: data batch_fetch_lines(args.city, args.max_line) print(f共获取 {len(data)} 条线路数据) if args.output csv: save_to_csv(data, f{args.city}_公交数据.csv) elif args.output sqlite: conn init_db(f{args.city}_bus.db) save_to_db(conn, data) conn.close() else: save_to_mongodb(data, f{args.city}_bus) print(数据采集完成) if __name__ __main__: main()对于定期更新的需求可以设置定时任务Linux crontab示例0 3 * * * /usr/bin/python3 /path/to/bus_collector.py 北京 --output sqlite /var/log/bus_collector.log 218. 最佳实践与经验分享在实际项目中我们总结了以下关键经验API调用优化将频繁使用的静态数据如线路基础信息缓存到本地减少API调用异常处理针对高德API可能返回的各种异常情况如限流、临时故障建立重试机制数据去重不同线路可能共享相同站点建立站点唯一标识避免重复增量更新记录数据采集时间戳下次只获取变更部分合规使用严格遵守高德API的使用条款避免高频请求一个典型的错误处理增强版请求函数如下def robust_fetch(city, line_name, max_retries3, cacheNone): if cache and line_name in cache: return cache[line_name] last_error None for attempt in range(max_retries): try: data fetch_bus_line(city, line_name, AMAP_KEY) if data and validate_line_data(data): if cache is not None: cache[line_name] data return data except requests.exceptions.RequestException as e: last_error e wait_time (attempt 1) * 5 # 指数退避 time.sleep(wait_time) print(f获取 {line_name} 失败: {str(last_error)}) return None对于需要长期维护的项目建议建立数据质量监控机制def data_quality_report(df): report { total_lines: len(df), missing_values: df.isnull().sum().to_dict(), avg_stops: df[站点数].mean(), avg_distance: df[运营里程].mean(), duplicate_lines: df.duplicated(subset[线路ID]).sum() } return report