Python数据抓取与质量校验：以杭州市公交线路为例

参考文章：Python爬取公交站点和线路数据（上下行双向） - 知乎 (zhihu.com)

之前也讲过数据抓取类型的文章很讲究时效性，该篇文章发于2020年，因为高德api策略的更新和网站抓取数据的机制变化等原因，如今脚本已经不能直接使用，我们这篇是对脚本原理的解释和流程的优化，本篇文章讲二个重点：1、解释清楚逻辑并略作优化让脚本可以重新跑起来了；2、讲一下数据质量，本文以杭州市公交线路为例，那么杭州市到底有多少条公交线路呢？

先讲一下方法思路，一共三个步骤；

方法思路

1. 获取公交信息网站——8684网站
2. 获取经纬度——通过调用高德地图API
3. 坐标转换——高德坐标系（GCJ-02） to WGS84

PS：运行前把代码中的url里面的key和安全密钥换成自己，另外把下面的城市参数改成自己需要获取数据的城市；

# 设定参数
citys = ['hangzhou']  # 城市总列表
chinese_city_names = ['杭州']

完整代码#运行环境Python 3.11

# -*- coding: utf-8 -*-import requests
import pandas as pd
import json
import re
import time
from bs4 import BeautifulSoup
import math# 获取城市公交线路的首字母列表
def get_initial_letters(city_name, headers):# 构造请求URLurl = f'https://{city_name}.8684.cn/list1'# 设置请求头headers = {'User-Agent': headers}# 发送GET请求response = requests.get(url, headers=headers)# 解析HTMLsoup = BeautifulSoup(response.text, 'lxml')# 查找包含首字母的标签initial_letters = soup.find_all('div', {'class': 'tooltip-inner'})[3].find_all('a')# 提取每个首字母return [letter.get_text() for letter in initial_letters]# 根据首字母获取公交线路名称
def get_bus_lines_by_letter(city_name, letter, headers, lines):# 构造请求URLurl = f'https://{city_name}.8684.cn/list{letter}'# 设置请求头headers = {'User-Agent': headers}# 发送GET请求response = requests.get(url, headers=headers)# 解析HTMLsoup = BeautifulSoup(response.text, 'lxml')# 查找包含线路名称的标签bus_lines = soup.find('div', {'class': 'list clearfix'}).find_all('a')# 提取每个线路名称lines.extend([line.get_text() for line in bus_lines])# 公交坐标信息转化（GCJ-02到WGS84）
def gcj02_to_wgs84(lng, lat):if out_of_china(lng, lat):  # 如果不在中国境内，则直接返回原坐标return lng, lat# 计算偏移量dlat = transformlat(lng - 105.0, lat - 35.0)dlng = transformlng(lng - 105.0, lat - 35.0)radlat = lat / 180.0 * math.pimagic = math.sin(radlat)magic = 1 - 0.00669342162296594323 * magic * magicsqrtmagic = math.sqrt(magic)dlat = (dlat * 180.0) / ((6378245.0 * (1 - 0.00669342162296594323)) / (magic * sqrtmagic) * math.pi)dlng = (dlng * 180.0) / (6378245.0 / sqrtmagic * math.cos(radlat) * math.pi)mglat = lat + dlatmglng = lng + dlngreturn [lng * 2 - mglng, lat * 2 - mglat]# 辅助函数，计算纬度的偏移量
def transformlat(lng, lat):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 * math.pi) + 20.0 * math.sin(2.0 * lng * math.pi)) * 2.0 / 3.0ret += (20.0 * math.sin(lat * math.pi) + 40.0 * math.sin(lat / 3.0 * math.pi)) * 2.0 / 3.0ret += (160.0 * math.sin(lat / 12.0 * math.pi) + 320 * math.sin(lat * math.pi / 30.0)) * 2.0 / 3.0return ret# 辅助函数，计算经度的偏移量
def transformlng(lng, lat):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 * math.pi) + 20.0 * math.sin(2.0 * lng * math.pi)) * 2.0 / 3.0ret += (20.0 * math.sin(lng * math.pi) + 40.0 * math.sin(lng / 3.0 * math.pi)) * 2.0 / 3.0ret += (150.0 * math.sin(lng / 12.0 * math.pi) + 300.0 * math.sin(lng / 30.0 * math.pi)) * 2.0 / 3.0return ret# 检查坐标是否在中国境内
def out_of_china(lng, lat):if lng < 72.004 or lng > 137.8347:return Trueif lat < 0.8293 or lat > 55.8271:return Truereturn False# 将坐标字符串转换为坐标元组
def parse_coordinates(coord_str):lng, lat = coord_str.split(',')return float(lng), float(lat)# 获取公交站点信息
def get_station_info(city, line_name):# 构造请求URLurl = f'https://restapi.amap.com/v3/bus/linename?s=rsv3&extensions=all&key=你自己的key&jscode=你自己的安全密钥&output=json&city={city}&offset=2&keywords={line_name}&platform=JS'# 发送GET请求response = requests.get(url)# 解析JSON响应data = json.loads(response.text)if data.get('status') == '1' and data.get('buslines'):  # 检查状态码是否为1，并且有数据buslines = data['buslines']if len(buslines) > 0:results = []for busline in buslines:line_name = busline['name']stations = [{'name': station['name'], 'location': station['location'], 'sequence': station['sequence']}for station in busline['busstops']]results.append({'line_name': line_name, 'stations': stations})return pd.DataFrame(results)return None# 获取公交线路数据信息
def get_line_info(city, line_name):# 构造请求URLurl = f'https://restapi.amap.com/v3/bus/linename?s=rsv3&extensions=all&key=d9f6b18f46c968c44a5a14308fae9044&jscode=92c7deb0430bf5a9efaf91dd70e418d5&output=json&city={city}&offset=2&keywords={line_name}&platform=JS'# 发送GET请求response = requests.get(url)# 解析JSON响应data = json.loads(response.text)if data.get('status') == '1' and data.get('buslines'):  # 检查状态码是否为1，并且有数据buslines = data['buslines']if len(buslines) > 0:results = []for busline in buslines:line_name = busline['name']polyline = busline['polyline']results.append({'line_name': line_name, 'polyline': polyline})return pd.DataFrame(results)return None# 爬取公交数据主程序
def get_station_and_line_info(citys, chinese_city_names, headers, file_path):for city, city_name in zip(citys, chinese_city_names):# 创建公交线路空列表lines = []# 获取首字母列表initial_letters = get_initial_letters(city, headers)# 根据首字母列表获取线路名称for letter in initial_letters:get_bus_lines_by_letter(city, letter, headers, lines)print(f'正在爬取{city_name}市的公交线路名称...')# 保存路径station_file = f'{file_path}{city}_station.csv'line_file = f'{file_path}{city}_line.csv'# 爬取公交站点数据print(f'正在爬取{city_name}市的公交站点数据...')station_data = pd.DataFrame()for line in lines:data = get_station_info(city_name, line)if data is not None:station_data = pd.concat([station_data, data], ignore_index=True)# 处理站点坐标processed_stations = []for _, row in station_data.iterrows():line_name = row['line_name']stations = row['stations']coord_x, coord_y = [], []for station in stations:lng, lat = parse_coordinates(station['location'])wlng, wlat = gcj02_to_wgs84(lng, lat)coord_x.append(wlng)coord_y.append(wlat)processed_stations.append({'line_name': [line_name] * len(stations), 'coord_x': coord_x, 'coord_y': coord_y,'station_name': [s['name'] for s in stations],'sequence': [s['sequence'] for s in stations]})# 合并站点数据stations_df = pd.concat([pd.DataFrame(s) for s in processed_stations], ignore_index=True)stations_df.to_csv(station_file, index=False, encoding='utf_8_sig')print(f'已保存{city_name}公交站点数据至 {station_file}')# 爬取公交线路数据print(f'正在爬取{city_name}市的公交线路数据...')line_data = pd.DataFrame()for line in lines:data = get_line_info(city_name, line)if data is not None:line_data = pd.concat([line_data, data], ignore_index=True)# 处理线路坐标processed_lines = []for _, row in line_data.iterrows():line_name = row['line_name']polyline = row['polyline'].split(';')for order, coord in enumerate(polyline):wlng, wlat = gcj02_to_wgs84(*parse_coordinates(coord))processed_lines.append({'line_name': line_name, 'order': order, 'lon': wlng, 'lat': wlat})# 保存线路数据lines_df = pd.DataFrame(processed_lines)lines_df.to_csv(line_file, index=False, encoding='utf_8_sig')print(f'已保存{city_name}公交线路数据至 {line_file}')# 设定参数
citys = ['hangzhou']  # 城市总列表
chinese_city_names = ['杭州']
headers = 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36'  # 浏览器user-agent
file_path = 'D://data//bus_data//'  # 数据存储路径
get_station_and_line_info(citys, chinese_city_names, headers, file_path)

脚本运行完成后会在D://data//bus_data这个路径下生成二个csv，这个有需要可以改成自己的路径，不改也行，会自己生成路径；

hangzhou_line.csv线路表包括线路名称、站点顺序、x坐标、y坐标的标签；

hangzhou_station.csv线路表包括线路名称、站点名称、站点顺序、x坐标、y坐标的标签；

把hangzhou_station.csv导入arcgis/arcgispro中，直接检索【点集转线】，且线字段选择line_name即可连成直线，方法在历史多篇文章里详细说明过，这里不再赘述；

接下来就是校对数据质量环节，我们打开hangzhou_line.csv这张表，把line_name单独复制出来，另存一个sheet ，在Excel中删除重复值，有2128条线路，另外因为是双向的原因，我们除于2，也就是1064条线路，我们再来打开杭州市公交集团官方来一探究竟，看看有多少条线路：集团介绍 | 杭州市公共交通集团有限公司 (hzbus.com.cn)；

介绍里提及，截止2024年6月营运线路1147条，差值83条线路，另外数据包含12条杭州地铁线路。偏差率大约8.28%，但这个偏差率大体上可以接受，这个差值大概是因为8684是第三方网站，数据更新有一定滞后性等原因，但作为学术论文分析等还是有一定的分析价值，可以作为基础数据分析出杭州市的公共交通分布情况、公交线路重复率、公交非直线系数等一些指标。

另外这里可以看一下我这篇文章：利用高德API获取整个城市的公交路线并可视化（一）_实时公交api-CSDN博客

我同样获取了杭州市的所有公交线路，且获取方式基本一致，但是这个数据获取的公交线路去除12条地铁线路，是1122条公交线路，差值仅25条，偏差率大约2.17%，这里姑且留个伏笔，待我研究研究。

文章仅用于分享个人学习成果与个人存档之用，分享知识，如有侵权，请联系作者进行删除。所有信息均基于作者的个人理解和经验，不代表任何官方立场或权威解读。