简介一份以城市公交站点及线路爬取为主题的Python实战资料面向爬虫初学者、数据采集爱好者及交通数据研究者解决从公交网站中自动提取站点和线路信息的关键问题。内容围绕requests与BeautifulSoup等常用库展开详细说明如何分析目标网站HTML结构、利用标签与类名编写选择器、解析站点和线路数据并借助csv模块将结果存储为结构化文件为后续数据分析与可视化打下基础。资源包大小约39.37MB文件总数与具体类型未明确标注整体以代码示例和讲解为主适合边学边练。已有678人学习本资源读者可从中获得完整的爬取思路和可复用脚本并延伸到pandas数据清洗、地图API公交线路绘制等进阶应用同时强化合规爬取意识尊重网站robots协议与数据版权。1. 城市公交站点及线路爬取先把“站点”和“线路”的数据模型想清楚城市公交站点及线路爬取看起来是把某座城市的线路、站点列表抓下来实际操作半小时后就会发现这更像一个建数据模型的问题而不是发请求熟练度的问题。一条线路有编号、有上下行、有首末站和首末班时间一个站点有名称、有经纬度还可能同时被十几条线路共用二者之间还要有一张带排序的关联表才能还原公交车“先到哪、后到哪”。很多人在这个环节没想清楚导致抓下来的数据里同一站名散落着多个坐标、同一线路缺方向标记后面做换乘计算或画线路图时全暴露出来。这篇从数据选型、抓包方式、写入 SQLite、并发与断点续传讲到数据质量校验覆盖城市范围跑通一套公交爬取链路的完整取舍。适合做公交路网分析、到站预测或 LBS Demo 的开发者参考。2. 数据源选型从网页、地图平台拿到线路与站点匹配什么样的爬取策略2.1 常见做法先去抓异步 JSON 接口而不是解析 HTML城市公交查询页大多不是服务端渲染页面骨架是 HTML 文件线路列表和站点顺序由浏览器在 JS 里二次请求得到。用 Python 的 requests 直接拉回来的 HTML 源码里往往连站名都搜不全更不用说站序。用 bs4 去解析这类动态页面是典型误入的方向因为真正有价值的数据藏在 XHR 响应里不看网络面板直接写解析代码大概率会卡在“取到的字段全为空”。打开浏览器开发者工具切到 Network 面板并筛选 XHR在公交查询框输入一条线路号随后能看到一个返回 JSON 数据的请求里面带着线路名称、站点列表、首末班时间等字段。先手动复制这个请求的 URL 和请求头再用一个小脚本验证它是否能被直接访问import requests url https://map.example.com/service/bus/line params { city: beijing, keywords: 1路, offset: 0, limit: 20, } headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36, Referer: https://map.example.com/bus, } resp requests.get(url, paramsparams, headersheaders, timeout10) print(resp.status_code) print(resp.text[:500])这里把 URL 做了脱敏因为各平台接口路径不同真实抓包时只需要把浏览器看到的地址照搬下来。请求头里的 User-Agent 和 Referer 同时带上很多接口会校验来源页面缺失时直接拒签。timeout 参数必须设置否则某个请求卡住会把整批任务拖死。2.2 四个可用数据源怎么选完整度、时效和坑的对比数据源覆盖完整度更新时效上手难度典型问题地图平台网页异步接口高主要城市基本齐全随地图数据更新时效最好中需要抓包处理请求头与分页接口参数会变限频明显城市开放数据平台低仅部分城市中取决于更新周期低有文档有域名白名单字段不全格式偏老OpenStreetMap中高城市间差异大靠志愿者维护中需要理解 node/relation 模型站名可能和实际站牌不一致公交公司官网单城市最准最快高多为页面渲染且无公开接口无反爬但有滚动加载和验证码风险我一般优先选择地图平台网页异步接口作为主力原因很直接覆盖度够、字段里带了经纬度、线路站点顺序完整。OSM 的问题在于站点名称不一定等于官方站名有些小城市甚至整条线路的关系都没人画城市开放数据平台则适合作为补充数据源来校准坐标而不是当作唯一的抓取对象。选完数据源还要评估一下反爬强度。公交查询接口通常不会要求登录也不会强制滑块但会做频率限制。单线程按每秒一个请求的节奏跑几百条线路能在一小时内完成一旦想加快就进入第 4 章的并发与重试控制范畴。记住一个原则先用小样本量验证接口返回结构再放开全量不要第一天就把任务数拉满。2.3 动手前定三个字段级约束无论选哪个源写代码前先定约束后面可以少返工。第一线路表必须留 direction 字段同一线路号的上行、下行当成两条记录不能混进同一个数组里。第二站点坐标单独存不要只存站点名站点重名率比想象中高经纬度才是最终去重依据。第三线路与站点关联表必须有 sort_no 序号它记录的是车辆途经顺序环线还会出现同一站点在线路中出现两次的情况没有序号就无法区分首末站。这三条约束确定后数据表结构、抓取代码和后续质检脚本可以一次对齐。3. 用 Python 爬取公交线路与站点数据并落库3.1 抓包确认关键特征动态页面不等于动态渲染有些开发者把“动态页面”等同于“用 Selenium 渲染”这其实是两个概念。公交查询站里的线路数据是异步返回的 JSON页面结构是联动变化的但数据本身不依赖浏览器渲染。确认方法很简单在 Network 面板里找到那个 JSON 请求点击 Preview如果能看到可读的 line_list、station_list 之类字段就说明请求接口已经暴露。这一步还要确认分页参数。公交查询页翻页时请求参数里通常有 offset 和 limit每次把偏移量加固定值即可取到下一页。接口返回里可能有 total 或 total_count 字段用来判断是否还有下一页。不要靠解析 HTML 里“下一页按钮”是否存在来判断接口字段更可靠。3.2 最小实现requests 拉取线路 JSON 并还原站点顺序以两个核心请求为例按线路名搜索得到初步线路列表再按线路 ID 获取该线路的完整站点序列。写成最小可运行版本如下import requests import time BASE_URL https://map.example.com/service/bus headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36, Referer: https://map.example.com/bus, } def search_line(city, keyword): params {city: city, keywords: keyword, offset: 0, limit: 20} resp requests.get(f{BASE_URL}/line, paramsparams, headersheaders, timeout10) resp.raise_for_status() data resp.json() return data.get(line_list, []) def get_line_detail(city, line_id): params {city: city, id: line_id} resp requests.get(f{BASE_URL}/line_detail, paramsparams, headersheaders, timeout10) resp.raise_for_status() detail resp.json() return detail.get(line, {})search_line 负责按关键词搜索返回的 line_list 里每条记录有线路名、线路 ID、首末站get_line_detail 则通过线路 ID 拿到这条线的完整站点数组。拆成两个函数的目的是把“搜索”和“详情”分开缓存搜索列表可能变化而线路详情可以长期复用。两个函数之间加 time.sleep(0.3)用来维持低频请求。随后从 detail 中提取站点顺序并按顺序输出def extract_stations(line_detail): stations [] for idx, st in enumerate(line_detail.get(stations, [])): stations.append({ station_id: st[station_id], station_name: st[station_name], lng: st[lng], lat: st[lat], sort_no: idx, }) return stations站点数组在接口里的顺序就是车辆实际经过顺序直接按 enumerate 的 index 存成 sort_no。不要自己去排序更不能把多趟线路的站点混到一个集合里再排序。这样得到的 stations 列表可以直接喂给下一节的 SQLite 写入逻辑。3.3 存储三张表一次事务写入站点和线路是多对多关系至少要拆三张表。线路表存线路元数据站点表存站点坐标中间表 line_station 存线路与站点的关联关系及顺序。建表语句见第 1 章提到的模型落地版CREATE TABLE IF NOT EXISTS line ( line_id TEXT PRIMARY KEY, line_name TEXT NOT NULL, direction INTEGER, start_station TEXT, end_station TEXT, first_bus_time TEXT, last_bus_time TEXT ); CREATE TABLE IF NOT EXISTS station ( station_id TEXT PRIMARY KEY, station_name TEXT NOT NULL, lng REAL, lat REAL ); CREATE TABLE IF NOT EXISTS line_station ( line_id TEXT, station_id TEXT, sort_no INTEGER, PRIMARY KEY (line_id, station_id, sort_no) );需要留意中间表主键里的 sort_no。环线公交的首末站常是同一个站点同一线路里同一站可能出现两次如果不带 sort_no第二次插入会直接违反主键约束。实际写入时可以先把整个线路的事务包起来import sqlite3 conn sqlite3.connect(bus.db) cur conn.cursor() line_stmt INSERT OR REPLACE INTO line VALUES (?,?,?,?,?,?,?) station_stmt INSERT OR REPLACE INTO station VALUES (?,?,?,?) rel_stmt INSERT OR REPLACE INTO line_station VALUES (?,?,?) for line_id, line_meta, stations in all_lines: cur.execute(line_stmt, ( line_id, line_meta[line_name], line_meta.get(direction, 0), line_meta.get(start_station, ), line_meta.get(end_station, ), line_meta.get(first_bus_time, ), line_meta.get(last_bus_time, ), )) for st in stations: cur.execute(station_stmt, ( st[station_id], st[station_name], st[lng], st[lat], )) cur.execute(rel_stmt, ( line_id, st[station_id], st[sort_no], )) conn.commit() conn.close()写入顺序分两层先写 line再循环写 station 和 line_station。使用 executemany 可以更高效但此处按单条 execute 写更容易阅读全城几千条线路、几万个站点时这个量级差异并不明显。还要注意 INSERT OR REPLACE 的粒度它按主键覆盖因此重新爬取同一线路时不会产生重复行。4. 让全城爬取不崩限速、重试、断点续传与去重控制4.1 全量任务拆分按关键词表分任务包而不是单线程循环城市公交线路不止“1路”到“999路”还有“快速公交”“机场专线”“夜班车”“社区微循环”等专名。常见做法是把关键词表拆成两种任务数字线路名用循环 range(1, 1000) 生成专名线路手动整理一张关键词表。两种任务合并后用 ThreadPoolExecutor 限制并发。from concurrent.futures import ThreadPoolExecutor, as_completed def run_task(keyword): try: lines search_line(beijing, keyword) return keyword, lines except Exception: return keyword, [] keywords [f{i}路 for i in range(1, 1000)] with ThreadPoolExecutor(max_workers8) as pool: futures [pool.submit(run_task, kw) for kw in keywords] for fut in as_completed(futures): kw, lines fut.result() if lines: print(kw, len(lines))并发线程数 8 是相对安全的起点公交接口不像电商页面那么严但也不要一口气开到 50。每个任务里保留 try/except单条线路失败不能影响整个任务池。关键词任务的返回结果不直接入库先存到内存队列再统一落库方便统计哪些关键词没搜到数据。4.2 限速与重试控制频率不靠“感觉”公交爬取任务最常见的失败就是请求过快触发频控。响应里遇到 429 或 503第一时间不是改代码而是确认是否触发了频率限制。下面这张参数表是常见配置参考具体数值要根据目标接口的容忍度再调参数建议值作用线程数8I/O 密集任务不需要开更多线程请求间隔0.3 到 0.8 秒每个 worker 完成一次请求后 sleep单次超时10 秒超时就放弃本次连接不无限等待最大重试5 次超过 5 次标记任务失败进入失败队列配合指数退避写一个带重试的请求函数比到处写 try/except 要省事得多import time import requests def fetch_json_with_retry(url, params, max_retries5, base_interval1.0): for attempt in range(max_retries): try: resp requests.get(url, paramsparams, headersheaders, timeout10) if resp.status_code 200: return resp.json() if 400 resp.status_code 500: return None except (requests.ConnectionError, requests.Timeout): pass time.sleep(base_interval * (2 ** attempt)) return None4xx 错误里的 403 和 429 都发生在服务端客户端重试意义不大5xx 和网络超时则可以重试。指数退避的时间序列是 1 秒、2 秒、4 秒、8 秒、16 秒重试期间如果连续失败问题多半是接口参数过期而不是网络抖动。把 fetcher 和业务逻辑分离后需要调整频率时只改 base_interval不需要动任务代码。4.3 断点续传把进度写进任务表全城爬取耗时可能数小时一次跑完不现实。断点续传不是把已经入库的数据重新下载一遍而是把每个关键词任务标记成 pending、done、failed 三种状态下次从 failed 接着跑。任务表结构越简单越好CREATE TABLE IF NOT EXISTS crawl_task ( task_id INTEGER PRIMARY KEY AUTOINCREMENT, keyword TEXT NOT NULL UNIQUE, status TEXT DEFAULT pending, retry_count INTEGER DEFAULT 0 );每次启动爬虫前从 crawl_task 里取 status in (pending, failed) 的记录组成新的任务列表全部跑完后更新状态。这样即使爬到一半宕掉也不需要人工排查跑到哪里重新启动后自动补齐缺失部分。4.4 站点去重同名同坐标的合并边界公交站点重名非常普遍比如“人民医院”在不同城区各自存在。入库后跑一条聚合 SQL能快速看出哪些站名出现频率最高SELECT station_name, COUNT(*) AS cnt FROM station GROUP BY station_name HAVING cnt 3 ORDER BY cnt DESC;查出来以后不要直接按站名合并。同一站名上行方向和下行方向的站点坐标可能相差几十米如果业务只做线路级分析可以按“站名相同 距离 100 米以内”合并如果做进站到站预测必须保留两个点位。合并动作放在入库后执行不要在爬取过程中边抓边合否则会丢失原始数据。先保留完整数据再用派生表做业务需要的“干净站点集合”这是最灵活的姿势。5. 数据质检用坐标和换乘关系校验爬取结果5.1 三条快速校验规则爬取完成不等于数据可用。最基础的三条规则可以在不依赖任何大型框架的情况下快速定位问题。规则一是线路闭合校验对每条 line_station 按 sort_no 排序检查第一站的 station_name 是否等于 line 表的 start_station最后一站是否等于 end_station。不一致大概率是线路详情请求拿错了线路 ID或者站序被覆盖。规则二是坐标域校验把全部站点的经纬度画出散点如果大批点位落在城市边界之外优先检查是不是经纬度写反了。经纬度反了是公交爬取任务里最隐蔽的错误接口里一般先给经度再给纬度接收时如果统一按 lat、lng 的顺序处理城市内点位会全部跑偏。规则三是相邻站点距离校验正常公交站距 300 米到 1 公里个别郊区线路会有两三公里的长站距但如果同一线路相邻两站超过 20 公里基本可以确定是数据关联错误。5.2 写一个不依赖第三方库的校验函数相邻站距校验可以写成轻量函数用 haversine 公式计算距离不引入额外依赖from math import radians, cos, sin, asin, sqrt def haversine(lng1, lat1, lng2, lat2): r 6371.0 d_lat radians(lat2 - lat1) d_lng radians(lng2 - lng1) a sin(d_lat / 2) ** 2 cos(radians(lat1)) * cos(radians(lat2)) * sin(d_lng / 2) ** 2 return r * 2 * asin(sqrt(a)) def check_line_stations(stations, threshold_km20): report [] for i in range(len(stations) - 1): dist haversine( stations[i][lng], stations[i][lat], stations[i 1][lng], stations[i 1][lat] ) if dist threshold_km: report.append((i, i 1, round(dist, 1))) return report调用时从 line_station 表按线路读取站点列表传入这个函数即可。输出结果里如果某条线路连续多段距离超限优先怀疑线路 ID 被串如果全城线路的相邻站距都偏大则检查坐标字段位的解析顺序。整体跑完这个函数再决定哪些线路需要重新抓取补数据通常比重跑全城便宜得多。本文还有配套的精品资源点击获取 城市公交线路站点爬取：数据建模、并发控制与质量校验
简介一份以城市公交站点及线路爬取为主题的Python实战资料面向爬虫初学者、数据采集爱好者及交通数据研究者解决从公交网站中自动提取站点和线路信息的关键问题。内容围绕requests与BeautifulSoup等常用库展开详细说明如何分析目标网站HTML结构、利用标签与类名编写选择器、解析站点和线路数据并借助csv模块将结果存储为结构化文件为后续数据分析与可视化打下基础。资源包大小约39.37MB文件总数与具体类型未明确标注整体以代码示例和讲解为主适合边学边练。已有678人学习本资源读者可从中获得完整的爬取思路和可复用脚本并延伸到pandas数据清洗、地图API公交线路绘制等进阶应用同时强化合规爬取意识尊重网站robots协议与数据版权。1. 城市公交站点及线路爬取先把“站点”和“线路”的数据模型想清楚城市公交站点及线路爬取看起来是把某座城市的线路、站点列表抓下来实际操作半小时后就会发现这更像一个建数据模型的问题而不是发请求熟练度的问题。一条线路有编号、有上下行、有首末站和首末班时间一个站点有名称、有经纬度还可能同时被十几条线路共用二者之间还要有一张带排序的关联表才能还原公交车“先到哪、后到哪”。很多人在这个环节没想清楚导致抓下来的数据里同一站名散落着多个坐标、同一线路缺方向标记后面做换乘计算或画线路图时全暴露出来。这篇从数据选型、抓包方式、写入 SQLite、并发与断点续传讲到数据质量校验覆盖城市范围跑通一套公交爬取链路的完整取舍。适合做公交路网分析、到站预测或 LBS Demo 的开发者参考。2. 数据源选型从网页、地图平台拿到线路与站点匹配什么样的爬取策略2.1 常见做法先去抓异步 JSON 接口而不是解析 HTML城市公交查询页大多不是服务端渲染页面骨架是 HTML 文件线路列表和站点顺序由浏览器在 JS 里二次请求得到。用 Python 的 requests 直接拉回来的 HTML 源码里往往连站名都搜不全更不用说站序。用 bs4 去解析这类动态页面是典型误入的方向因为真正有价值的数据藏在 XHR 响应里不看网络面板直接写解析代码大概率会卡在“取到的字段全为空”。打开浏览器开发者工具切到 Network 面板并筛选 XHR在公交查询框输入一条线路号随后能看到一个返回 JSON 数据的请求里面带着线路名称、站点列表、首末班时间等字段。先手动复制这个请求的 URL 和请求头再用一个小脚本验证它是否能被直接访问import requests url https://map.example.com/service/bus/line params { city: beijing, keywords: 1路, offset: 0, limit: 20, } headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36, Referer: https://map.example.com/bus, } resp requests.get(url, paramsparams, headersheaders, timeout10) print(resp.status_code) print(resp.text[:500])这里把 URL 做了脱敏因为各平台接口路径不同真实抓包时只需要把浏览器看到的地址照搬下来。请求头里的 User-Agent 和 Referer 同时带上很多接口会校验来源页面缺失时直接拒签。timeout 参数必须设置否则某个请求卡住会把整批任务拖死。2.2 四个可用数据源怎么选完整度、时效和坑的对比数据源覆盖完整度更新时效上手难度典型问题地图平台网页异步接口高主要城市基本齐全随地图数据更新时效最好中需要抓包处理请求头与分页接口参数会变限频明显城市开放数据平台低仅部分城市中取决于更新周期低有文档有域名白名单字段不全格式偏老OpenStreetMap中高城市间差异大靠志愿者维护中需要理解 node/relation 模型站名可能和实际站牌不一致公交公司官网单城市最准最快高多为页面渲染且无公开接口无反爬但有滚动加载和验证码风险我一般优先选择地图平台网页异步接口作为主力原因很直接覆盖度够、字段里带了经纬度、线路站点顺序完整。OSM 的问题在于站点名称不一定等于官方站名有些小城市甚至整条线路的关系都没人画城市开放数据平台则适合作为补充数据源来校准坐标而不是当作唯一的抓取对象。选完数据源还要评估一下反爬强度。公交查询接口通常不会要求登录也不会强制滑块但会做频率限制。单线程按每秒一个请求的节奏跑几百条线路能在一小时内完成一旦想加快就进入第 4 章的并发与重试控制范畴。记住一个原则先用小样本量验证接口返回结构再放开全量不要第一天就把任务数拉满。2.3 动手前定三个字段级约束无论选哪个源写代码前先定约束后面可以少返工。第一线路表必须留 direction 字段同一线路号的上行、下行当成两条记录不能混进同一个数组里。第二站点坐标单独存不要只存站点名站点重名率比想象中高经纬度才是最终去重依据。第三线路与站点关联表必须有 sort_no 序号它记录的是车辆途经顺序环线还会出现同一站点在线路中出现两次的情况没有序号就无法区分首末站。这三条约束确定后数据表结构、抓取代码和后续质检脚本可以一次对齐。3. 用 Python 爬取公交线路与站点数据并落库3.1 抓包确认关键特征动态页面不等于动态渲染有些开发者把“动态页面”等同于“用 Selenium 渲染”这其实是两个概念。公交查询站里的线路数据是异步返回的 JSON页面结构是联动变化的但数据本身不依赖浏览器渲染。确认方法很简单在 Network 面板里找到那个 JSON 请求点击 Preview如果能看到可读的 line_list、station_list 之类字段就说明请求接口已经暴露。这一步还要确认分页参数。公交查询页翻页时请求参数里通常有 offset 和 limit每次把偏移量加固定值即可取到下一页。接口返回里可能有 total 或 total_count 字段用来判断是否还有下一页。不要靠解析 HTML 里“下一页按钮”是否存在来判断接口字段更可靠。3.2 最小实现requests 拉取线路 JSON 并还原站点顺序以两个核心请求为例按线路名搜索得到初步线路列表再按线路 ID 获取该线路的完整站点序列。写成最小可运行版本如下import requests import time BASE_URL https://map.example.com/service/bus headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36, Referer: https://map.example.com/bus, } def search_line(city, keyword): params {city: city, keywords: keyword, offset: 0, limit: 20} resp requests.get(f{BASE_URL}/line, paramsparams, headersheaders, timeout10) resp.raise_for_status() data resp.json() return data.get(line_list, []) def get_line_detail(city, line_id): params {city: city, id: line_id} resp requests.get(f{BASE_URL}/line_detail, paramsparams, headersheaders, timeout10) resp.raise_for_status() detail resp.json() return detail.get(line, {})search_line 负责按关键词搜索返回的 line_list 里每条记录有线路名、线路 ID、首末站get_line_detail 则通过线路 ID 拿到这条线的完整站点数组。拆成两个函数的目的是把“搜索”和“详情”分开缓存搜索列表可能变化而线路详情可以长期复用。两个函数之间加 time.sleep(0.3)用来维持低频请求。随后从 detail 中提取站点顺序并按顺序输出def extract_stations(line_detail): stations [] for idx, st in enumerate(line_detail.get(stations, [])): stations.append({ station_id: st[station_id], station_name: st[station_name], lng: st[lng], lat: st[lat], sort_no: idx, }) return stations站点数组在接口里的顺序就是车辆实际经过顺序直接按 enumerate 的 index 存成 sort_no。不要自己去排序更不能把多趟线路的站点混到一个集合里再排序。这样得到的 stations 列表可以直接喂给下一节的 SQLite 写入逻辑。3.3 存储三张表一次事务写入站点和线路是多对多关系至少要拆三张表。线路表存线路元数据站点表存站点坐标中间表 line_station 存线路与站点的关联关系及顺序。建表语句见第 1 章提到的模型落地版CREATE TABLE IF NOT EXISTS line ( line_id TEXT PRIMARY KEY, line_name TEXT NOT NULL, direction INTEGER, start_station TEXT, end_station TEXT, first_bus_time TEXT, last_bus_time TEXT ); CREATE TABLE IF NOT EXISTS station ( station_id TEXT PRIMARY KEY, station_name TEXT NOT NULL, lng REAL, lat REAL ); CREATE TABLE IF NOT EXISTS line_station ( line_id TEXT, station_id TEXT, sort_no INTEGER, PRIMARY KEY (line_id, station_id, sort_no) );需要留意中间表主键里的 sort_no。环线公交的首末站常是同一个站点同一线路里同一站可能出现两次如果不带 sort_no第二次插入会直接违反主键约束。实际写入时可以先把整个线路的事务包起来import sqlite3 conn sqlite3.connect(bus.db) cur conn.cursor() line_stmt INSERT OR REPLACE INTO line VALUES (?,?,?,?,?,?,?) station_stmt INSERT OR REPLACE INTO station VALUES (?,?,?,?) rel_stmt INSERT OR REPLACE INTO line_station VALUES (?,?,?) for line_id, line_meta, stations in all_lines: cur.execute(line_stmt, ( line_id, line_meta[line_name], line_meta.get(direction, 0), line_meta.get(start_station, ), line_meta.get(end_station, ), line_meta.get(first_bus_time, ), line_meta.get(last_bus_time, ), )) for st in stations: cur.execute(station_stmt, ( st[station_id], st[station_name], st[lng], st[lat], )) cur.execute(rel_stmt, ( line_id, st[station_id], st[sort_no], )) conn.commit() conn.close()写入顺序分两层先写 line再循环写 station 和 line_station。使用 executemany 可以更高效但此处按单条 execute 写更容易阅读全城几千条线路、几万个站点时这个量级差异并不明显。还要注意 INSERT OR REPLACE 的粒度它按主键覆盖因此重新爬取同一线路时不会产生重复行。4. 让全城爬取不崩限速、重试、断点续传与去重控制4.1 全量任务拆分按关键词表分任务包而不是单线程循环城市公交线路不止“1路”到“999路”还有“快速公交”“机场专线”“夜班车”“社区微循环”等专名。常见做法是把关键词表拆成两种任务数字线路名用循环 range(1, 1000) 生成专名线路手动整理一张关键词表。两种任务合并后用 ThreadPoolExecutor 限制并发。from concurrent.futures import ThreadPoolExecutor, as_completed def run_task(keyword): try: lines search_line(beijing, keyword) return keyword, lines except Exception: return keyword, [] keywords [f{i}路 for i in range(1, 1000)] with ThreadPoolExecutor(max_workers8) as pool: futures [pool.submit(run_task, kw) for kw in keywords] for fut in as_completed(futures): kw, lines fut.result() if lines: print(kw, len(lines))并发线程数 8 是相对安全的起点公交接口不像电商页面那么严但也不要一口气开到 50。每个任务里保留 try/except单条线路失败不能影响整个任务池。关键词任务的返回结果不直接入库先存到内存队列再统一落库方便统计哪些关键词没搜到数据。4.2 限速与重试控制频率不靠“感觉”公交爬取任务最常见的失败就是请求过快触发频控。响应里遇到 429 或 503第一时间不是改代码而是确认是否触发了频率限制。下面这张参数表是常见配置参考具体数值要根据目标接口的容忍度再调参数建议值作用线程数8I/O 密集任务不需要开更多线程请求间隔0.3 到 0.8 秒每个 worker 完成一次请求后 sleep单次超时10 秒超时就放弃本次连接不无限等待最大重试5 次超过 5 次标记任务失败进入失败队列配合指数退避写一个带重试的请求函数比到处写 try/except 要省事得多import time import requests def fetch_json_with_retry(url, params, max_retries5, base_interval1.0): for attempt in range(max_retries): try: resp requests.get(url, paramsparams, headersheaders, timeout10) if resp.status_code 200: return resp.json() if 400 resp.status_code 500: return None except (requests.ConnectionError, requests.Timeout): pass time.sleep(base_interval * (2 ** attempt)) return None4xx 错误里的 403 和 429 都发生在服务端客户端重试意义不大5xx 和网络超时则可以重试。指数退避的时间序列是 1 秒、2 秒、4 秒、8 秒、16 秒重试期间如果连续失败问题多半是接口参数过期而不是网络抖动。把 fetcher 和业务逻辑分离后需要调整频率时只改 base_interval不需要动任务代码。4.3 断点续传把进度写进任务表全城爬取耗时可能数小时一次跑完不现实。断点续传不是把已经入库的数据重新下载一遍而是把每个关键词任务标记成 pending、done、failed 三种状态下次从 failed 接着跑。任务表结构越简单越好CREATE TABLE IF NOT EXISTS crawl_task ( task_id INTEGER PRIMARY KEY AUTOINCREMENT, keyword TEXT NOT NULL UNIQUE, status TEXT DEFAULT pending, retry_count INTEGER DEFAULT 0 );每次启动爬虫前从 crawl_task 里取 status in (pending, failed) 的记录组成新的任务列表全部跑完后更新状态。这样即使爬到一半宕掉也不需要人工排查跑到哪里重新启动后自动补齐缺失部分。4.4 站点去重同名同坐标的合并边界公交站点重名非常普遍比如“人民医院”在不同城区各自存在。入库后跑一条聚合 SQL能快速看出哪些站名出现频率最高SELECT station_name, COUNT(*) AS cnt FROM station GROUP BY station_name HAVING cnt 3 ORDER BY cnt DESC;查出来以后不要直接按站名合并。同一站名上行方向和下行方向的站点坐标可能相差几十米如果业务只做线路级分析可以按“站名相同 距离 100 米以内”合并如果做进站到站预测必须保留两个点位。合并动作放在入库后执行不要在爬取过程中边抓边合否则会丢失原始数据。先保留完整数据再用派生表做业务需要的“干净站点集合”这是最灵活的姿势。5. 数据质检用坐标和换乘关系校验爬取结果5.1 三条快速校验规则爬取完成不等于数据可用。最基础的三条规则可以在不依赖任何大型框架的情况下快速定位问题。规则一是线路闭合校验对每条 line_station 按 sort_no 排序检查第一站的 station_name 是否等于 line 表的 start_station最后一站是否等于 end_station。不一致大概率是线路详情请求拿错了线路 ID或者站序被覆盖。规则二是坐标域校验把全部站点的经纬度画出散点如果大批点位落在城市边界之外优先检查是不是经纬度写反了。经纬度反了是公交爬取任务里最隐蔽的错误接口里一般先给经度再给纬度接收时如果统一按 lat、lng 的顺序处理城市内点位会全部跑偏。规则三是相邻站点距离校验正常公交站距 300 米到 1 公里个别郊区线路会有两三公里的长站距但如果同一线路相邻两站超过 20 公里基本可以确定是数据关联错误。5.2 写一个不依赖第三方库的校验函数相邻站距校验可以写成轻量函数用 haversine 公式计算距离不引入额外依赖from math import radians, cos, sin, asin, sqrt def haversine(lng1, lat1, lng2, lat2): r 6371.0 d_lat radians(lat2 - lat1) d_lng radians(lng2 - lng1) a sin(d_lat / 2) ** 2 cos(radians(lat1)) * cos(radians(lat2)) * sin(d_lng / 2) ** 2 return r * 2 * asin(sqrt(a)) def check_line_stations(stations, threshold_km20): report [] for i in range(len(stations) - 1): dist haversine( stations[i][lng], stations[i][lat], stations[i 1][lng], stations[i 1][lat] ) if dist threshold_km: report.append((i, i 1, round(dist, 1))) return report调用时从 line_station 表按线路读取站点列表传入这个函数即可。输出结果里如果某条线路连续多段距离超限优先怀疑线路 ID 被串如果全城线路的相邻站距都偏大则检查坐标字段位的解析顺序。整体跑完这个函数再决定哪些线路需要重新抓取补数据通常比重跑全城便宜得多。本文还有配套的精品资源点击获取