公交数据可视化实战：Python爬虫+Flask+SQLite+ECharts全链路解析

简介这是一套基于Python与Flask的公交数据可视化分析大作业源码及文档是期末大作业开发中获97分的高分设计项目适合作为课程设计、期末大作业或毕设参考。项目以公交数据为对象覆盖爬虫数据采集、数据清洗存储、Flask后端接口和可视化大屏展示等完整环节配有丰富的代码注释基础薄弱也能读懂并支持二次开发。压缩包共2000个文件以1890个Python源码文件为主涵盖爬虫、Flask后端与数据分析模块另有txt说明、json数据、md文档及少量css/js前端资源帮助理解配置与前端交互整体大小62.36MB结构清晰。目前已有323人学习/下载。对于想要快速搭建数据可视化项目、梳理爬虫Flask完整开发流程的读者这是一份可直接运行的高分设计参考。1. 公交数据可视化大作业真正卡进度的是数据不是图表公交数据可视化这个题目听上去像是一个前端图表项目但实际做下来你会发现另一个结论ECharts 画图只需要几十分钟难的是把线路、站点、车辆运行时间这些数据稳定抓下来清洗成一张能查、能聚类的表。这套技术栈的典型分工是 Python 写爬虫抓数据Flask 把数据暴露成 JSON 接口SQLite 管存储前端用 ECharts 消费接口。对准备课程设计或求职作品集的人它是一条完整可讲的链路对需要快速搭内部数据看板的工程师它也是一个轻量可复制的模板。需要提醒的是抓取数据请遵守目标站点的 robots 约定控制请求频率并对下载内容做合规检查。2. 用 requests BeautifulSoup 搭一个可重试、可限速的公交数据爬虫2.1 数据源选型静态 HTML、JSON 接口和文档型 PDF公交数据一般来自三个渠道抓取策略完全不同。城市公共交通集团官网的线路列表多是静态页面适合 requests 直接抓 HTML 再解析部分数据开放平台会暴露 JSON API可以直接解析结构体还有一类是 PDF 发布的站点时刻表则需要先转文本再用正则做信息抽取。我一般会先把目标页面下载到本地再做解析而不是每次跑都打线上接口。表格里列一下常见选择和对应工具数据源形态典型位置解析方案适用场景静态 HTML 页面公交集团官网线路页requests BeautifulSoup(lxml)线路列表、站点顺序、首末班时间JSON API市级交通开放平台requests response.json()实时车辆位置、到站预估PDF / 图片官网公告附件pdfplumber 转文本 正则站点更名、临时调整第三方地图 Web 页地图公交详情不建议抓取数据版权和反爬成本都高确定了数据源之后第二步才是写代码。如果站点本身有开放数据接口优先走接口没有接口才退回到 HTML 解析路线这个顺序能省掉大量清洗工作。2.2 带重试机制的 Session 封装requests 爬虫不是一个请求函数爬虫最常见的问题是“跑一半挂了”。一个请求超时整个脚本崩溃第二次跑又从第一条开始。常见做法是给 requests 封装一层 Session把重试和连接池复用的逻辑统一收口。import random import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry UA_POOL [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36, ] def build_retry_session(retries: int 3, backoff_factor: float 0.6): retry Retry( totalretries, readretries, connectretries, status_forcelist[500, 502, 503, 504], backoff_factorbackoff_factor, ) session requests.Session() session.headers[User-Agent] random.choice(UA_POOL) session.headers[Referer] https://bus.example.cn/ adapter HTTPAdapter(max_retriesretry, pool_connections10, pool_maxsize20) session.mount(http://, adapter) session.mount(https://, adapter) return session这里status_forcelist指定哪些 HTTP 状态码触发重试500 和 502 这类服务端错误重试才有意义backoff_factor让每次重试等待时间递增0.6 表示第一次等待 0.6 秒、第二次 1.2 秒避免重试风暴。UA 池随机切换是为了减少同一 User-Agent 在日志里出现频率过高的问题但这不是反爬的万能药真正更有效的方案是控制抓取节奏。2.3 解析与入库先落 HTML 再清洗避免反复抓站建议把每次抓到的原始 HTML 存成文件文件名带抓取时间之后再从本地文件解析进数据库。这样做有两个好处同一个页面调试解析逻辑时不产生二次请求原始报文留存方便回溯字段问题。解析代码直接操作 BeautifulSoup 对象from bs4 import BeautifulSoup from pathlib import Path def parse_line_page(html_path: str): html Path(html_path).read_text(encodingutf-8) soup BeautifulSoup(html, lxml) line_list [] for item in soup.select(ul.line-list li): name item.select_one(.line-name).get_text(stripTrue) href item.select_one(a).get(href) line_list.append({name: name, detail_url: href}) return line_listlxml解析速度明显快于标准库html.parser在列表页成百上千个节点的情况下差距能到十倍级别。抓取循环里加随机 sleepimport time for page in range(1, total_pages 1): url fhttps://bus.example.cn/list?page{page} resp session.get(url, timeout10) if resp.status_code 200: raw_dir Path(data/raw) raw_dir.mkdir(parentsTrue, exist_okTrue) (raw_dir / flist_{page}.html).write_text(resp.text, encodingutf-8) time.sleep(random.uniform(1.0, 2.5))timeout10必须写否则 DNS 解析卡住会占用整个 workerrandom.uniform(1.0, 2.5)让请求间隔在 1 到 2.5 秒之间抖动比固定 sleep 更贴近人工浏览行为。数据抓回来看一眼 HTML 再写解析规则这个顺序不要颠倒。3. Flask 提供数据接口SQLite 建模公交线路与到站数据3.1 为什么选 SQLite 而不是 MySQL 或 JSON 文件公交可视化的数据规模一般是万级线路、十万级站点记录SQLite 完全撑得住而且交付时不用部署数据库服务。对比用 JSON 文件直接交给前端SQLite 的好处是可以在 SQL 层做聚合排序比如“某条线路 8 点到 9 点的平均到站间隔”一条 COUNT 加 GROUP BY 就完成。相比之下如果爬虫需要频繁增量更新数据JSON 文件很难做并发写入SQLite 的事务机制能保证同一时间只有一份版本在生效。选择 SQLite 需要把一个点提前想清楚它的强项是本地读写不适合多个服务进程同时高并发写。大作业这种单机爬虫加单机 Flask 的架构远没有触到它的上限。如果后续要支撑更大并发可以把sqlite3.connect(..., check_same_threadFalse)替换成 SQLAlchemy 的连接池做平滑过渡但初期没必要引入额外依赖。3.2 表设计与建表语句线路、站点、GPS 记录分三张这三张表是公交可视化最常见的模型。第一张bus_lines存线路基本信息第二张bus_stops存站点排序第三张bus_gps存车辆位置或到站记录。站点和线路是多对多关系所以第三张表也承担关联查询的职责。CREATE TABLE bus_lines ( line_id TEXT PRIMARY KEY, line_name TEXT NOT NULL, start_stop TEXT, end_stop TEXT, first_bus TEXT, last_bus TEXT ); CREATE TABLE bus_stops ( stop_id INTEGER PRIMARY KEY AUTOINCREMENT, line_id TEXT NOT NULL REFERENCES bus_lines(line_id), seq_no INTEGER NOT NULL, stop_name TEXT NOT NULL, lng REAL, lat REAL ); CREATE TABLE bus_gps ( id INTEGER PRIMARY KEY AUTOINCREMENT, line_id TEXT NOT NULL, run_time TEXT NOT NULL, lng REAL, lat REAL, status TEXT ); CREATE INDEX idx_stops_line ON bus_stops(line_id, seq_no); CREATE INDEX idx_gps_line_time ON bus_gps(line_id, run_time);seq_no表示站点在线路中的顺序可视化画路线时必须按它排序不然线段会乱跳。run_time建议统一存成YYYY-MM-DD HH:MM:SS的字符串浏览数据时不用转换就能看懂Python 侧构造也方便。索引加在查询频率最高的联合字段上一张百万行级别的 GPS 表没有索引可以慢到几百毫秒加了之后基本稳定在 10 毫秒内。3.3 按版本暴露 JSON APIFlask 只做接口层Flask 在这个项目里的角色是纯接口层不做模板渲染。路由设计成带版本号的结构前端升级时后端可以平行兼容。from flask import Flask, jsonify, request from flask_cors import CORS import sqlite3 app Flask(__name__) CORS(app) def query_db(sql: str, args: tuple ()): conn sqlite3.connect(bus_data.db) conn.row_factory sqlite3.Row rows conn.execute(sql, args).fetchall() conn.close() return [dict(row) for row in rows] app.get(/api/v1/lines) def list_lines(): keyword request.args.get(keyword, ).strip() if keyword: rows query_db( SELECT * FROM bus_lines WHERE line_name LIKE ? LIMIT 20, (f%{keyword}%,), ) else: rows query_db(SELECT * FROM bus_lines LIMIT 100) return jsonify({code: 0, data: rows})CORS(app)解决前端本地调试时的跨域问题否则从 5500 端口打开页面访问 Flask 的 5000 端口会被浏览器拦截。LIKE模糊查询用于线路名搜索注意参数化查询用?占位符不要把字符串直接拼进 SQLFlask 不会自动处理这种拼接风险。row_factory sqlite3.Row让查询结果可以按列名取值序列化成 JSON 时中文比较整洁。3.4 前端消费 APIECharts 画时段折线与线路热度接口准备好之后前端用原生 fetch 拿 JSON图表用 ECharts 渲染。为了离线演示稳定建议把echarts.min.js下载到本地 static 目录而不是引用 CDN避免答辩现场没有外网导致图表空白。下面是画一个线路全天到站次数折线图的最小示例fetch(/api/v1/gps_histogram?line_id822) .then((res) res.json()) .then((json) { const data json.data; const chart echarts.init(document.getElementById(histogram)); chart.setOption({ xAxis: { type: category, data: data.hours }, yAxis: { type: value }, series: [{ type: line, data: data.counts, areaStyle: {} }], }); });接口返回的字段名需要在后端定义时保持稳定前端所有图表都依赖这个契约。如果后期改了字段名图表会静默失败控制台只报 undefined排查起来比较费时间。4. 源码工程化与文档说明从单脚本改成分层项目4.1 分层结构 models / services / api / spiders 各管什么大多数大作业源码是从爬虫脚本一路叠加写出来的一个 main.py 里混了 requests、数据库、Flask 路由和 HTML 字符串。跑起来没问题改一个参数要翻遍全部代码。一般我会把项目拆成下面这种结构拆完以后每一层都可以单独测试。bus-vis/ ├── app.py ├── config.py ├── requirements.txt ├── spiders/ │ ├── __init__.py │ └── crawl_bus.py ├── services/ │ ├── __init__.py │ └── line_service.py ├── models/ │ ├── __init__.py │ └── db.py ├── api/ │ ├── __init__.py │ └── routes.py ├── static/ │ └── echarts.min.js ├── templates/ │ └── index.html └── data/ └── raw/spiders只负责请求网页和解析结构返回 Python 字典或列表models封装数据库的初始化、连接和底层查询业务逻辑不写 SQLservices是中间层从爬虫拿原始数据调用 models 写入并做清洗api只注册路由把 services 的返回结果转成 JSON。Flask 的入口app.py里只做三件事创建 Flask 实例、注册蓝图、调用create_all建表。4.2 配置外置与依赖锁定改参数不用动业务代码数据库路径、爬虫请求间隔、UA 池这些应放到 config.py 里这样改配置不用碰逻辑代码也让文档说明和实际运行参数对得上。class Config: DATABASE_PATH bus_data.db REQUEST_TIMEOUT 10 CRAWL_SLEEP_RANGE (1.0, 2.5) MAX_PAGES 50运行入口读配置app Flask(__name__) app.config.from_object(Config)requirements.txt锁定主要依赖版本尽量写“”而不是“”这样既能保持兼容又不会把已有环境锁死在一个过旧版本上。依赖列表里除了 Flask 和 requests还需要lxmlHTML 解析引擎和flask-cors跨域支持。4.3 README 文档说明的五个要素文档说明不写成了使用手册把下面五件事写清楚就够了项目是干什么的、目录结构是什么、环境要求是什么、运行顺序是什么、爬虫抓的是什么数据合法吗。README 里贴出运行顺序是关键按下面的最小模板来python -m venv .venv source .venv/bin/activate pip install -r requirements.txt python app.py先执行爬虫再启动 Flask或者直接做成 Flask CLI 命令这一步在第五章展开。文档里还要写清楚数据更新方式是手动重跑爬虫还是定时任务大作业阶段手动重跑已经足够不要加 cron 把线上站点抓出问题。4.4 从大作业源码到企业级数据可视化还差哪几步这部分是对有经验的读者说的。大作业代码能跑但距离可交付还有四条明显的差距。第一是缺少日志写个简单装饰器把所有关键步骤打到 logs 目录比后期靠 print 排查靠谱得多。第二是没有限流前后两次抓取如果并发数超过网站阈值IP 会被封requests 只能帮你重试不能帮你预防。第三是没有监控指标加一个/api/v1/health路由返回最后抓取时间、数据库连接状态和记录数。第四是配置外部化把密钥、数据库路径放进环境变量而不是写死在 config.py 里。做项目展示时把这几点主动讲出来比跑通 20 张图表更能体现工程能力。5. 用 Flask CLI 一键完成抓取、入库与可视化刷新5.1 自定义 flask crawl-update 命令串起整条链路Flask 的 CLI 体系可以挂自定义命令把爬虫、清洗、入库、缓存清理串成一条命令。这样文档说明里就不用写“先运行 python crawl.py再运行 python app.py”而是flask crawl-update一条路径走完也方便部署脚本调用。import click from flask.cli import with_appcontext def register_commands(app): app.cli.command(crawl-update) with_appcontext def crawl_update_command(): 抓取最新公交数据并刷新接口缓存。 from spiders.crawl_bus import run_spider from services.line_service import load_lines_into_db raw_data run_spider() load_lines_into_db(raw_data) app.config[LAST_UPDATE] datetime.now().isoformat() click.echo(update finished)with_appcontext保证命令执行时可以访问 app 配置和数据库连接click.echo输出统一格式的成功信息配合日志文件可以快速确认执行状态。Flask 2.3 及以上版本用flask --app app crawl-update触发命令。5.2 用 run_time 目录做数据快照可视化随时回溯每次抓取的数据先落到data/raw/20250311/这种按日期命名的目录再在bus_gps表里用run_time字段记录抓取批次时间。这样做的好处是如果某天抓到的数据明显异常可以直接从旧目录恢复到上一个批次不用重新请求站点。数据快照配合接口里的时间参数还能做“不同日期的线路运力对比”这是可视化看图时最加分的功能。5.3 一条命令启动 Flask 并验证接口开发阶段用 127.0.0.1 启动方便调试演示时可以绑定局域网 IP 让其他设备访问flask --app app run --host0.0.0.0 --port5000 curl http://127.0.0.1:5000/api/v1/lines?keyword822curl 返回 JSON 数组后再用浏览器打开http://127.0.0.1:5000看可视化页面。如果接口返回空列表优先检查bus_data.db是否生成、run_spider()是否正常完成而不是先怀疑图表代码。整条链路从抓取到展示都跑通之后再回头看 README 里的运行说明把命令替换成 Flask CLI 方式文档和代码就不会出现两套不一致的启动路径。本文还有配套的精品资源点击获取