简介这是一份面向Python爬虫与数据可视化学习者的实战项目以北京地铁客流量为分析对象完整覆盖数据抓取、清洗、存储到前端图表展示的流程。资源共10个文件包含2个Python脚本负责客流数据采集与时间处理、1个JavaScript数据文件、1个HTML页面与1个JS图表文件另有说明文档、许可协议和忽略规则文件压缩包仅125KB结构紧凑适合快速上手。目前已有1409人学习下载。通过该项目可掌握requests/BeautifulSoup等爬虫基础、pandas数据预处理思路以及ECharts/Highcharts类库绘制交互式统计图的方法项目还涉及HTML/CSS/JavaScript的页面整合和Git版本管理适合用于课程设计或作品集积累。所有代码与数据文件均按功能分离便于逐模块对照学习是一份综合性较强的入门级实战资料。1. 北京地铁客流量统计在做什么从一天的客流到一条趋势线北京地铁每天上午发布的昨日客流数字是通勤族和运营人员都盯着的指标路网总客运量多少万人次、哪条线最挤、跟前一天比是涨是跌。手动查一天两天可以连续查一个月就成了机械劳动而且数据散落在不同页面想画趋势还得自己往 Excel 里粘。这个标题里的项目就是把这条线走通Python 爬虫负责定时抓取清洗成统一格式再用 JS 统计图渲染成浏览器里能直接看的页面。一个容易误判的点是爬虫在这个项目里只占两成工作量剩下八成在数据清洗、时间轴对齐和数值自检上。数字抓回来带着千分位、带着「万人次」后缀还有断更的日期和偶尔写错的合计值不处理干净前端画出来的图就是错的。适合刚学完 requests 想拿真实数据练手的人也适合做通勤分析、需要攒历史客流数据的工程师。全程不依赖付费接口一台能跑 Python 的机器加上一个静态文件服务器就够。2. 用 Python 爬虫抓北京地铁客流先定数据源再写 requests 代码2.1 客流数据长什么样字段和单位要先于代码定下来北京地铁的每日客流并不是开放接口直接给的公开渠道里最常见的是两类官方微博账号每天发布的「昨日路网客运量」带总量和进站量地铁客流爱好者和聚合站点维护的历史分线数据按日期列出每条线路的客运量。第一类字段少、更新固定适合画总量趋势第二类有分线数据能画「哪条线最挤」的对比图。我一般建议两个都抓字段按下面这张表归一到本地记录里。有些博主只发公众号但公众号图文页带登录态校验抓取成本高不如直接换网页源。字段示例值说明落地类型date2024-05-20数据对应的日期统一成 YYYY-MM-DDstringline10号线线路名总量记录填「路网」stringpassenger_flow_10k168.3客运量统一换算成万人次floatentry_flow_10k96.7进站量没有就留空floatupdate_time2024-05-21 09:30抓取时间用于日志排查string单位是第一个坑有的页面写「1250.6 万人次」有的写「12,506,000 人」还有的写「万乘次」。爬虫层只负责把数字字符串原样提取出来单位判断放清洗阶段统一做。字段定得越死后面画图越省事。2.2 requests BeautifulSoup 抓表格页面的最小实现大多数聚合站点的历史数据是静态 HTML 表格用 requests 拿页面、BeautifulSoup 解析就够了这也是 requests 爬虫里最常见的形态。先写抓页面这一段import requests from bs4 import BeautifulSoup # 示例地址替换成你实际找的数据源页面 METRO_FLOW_URL https://example.com/subway/daily HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36, Referer: https://example.com/subway/, Accept: text/html,application/xhtmlxml, } def fetch_html(url: str) - str: resp requests.get(url, headersHEADERS, timeout10) resp.raise_for_status() # 非 2xx 直接抛异常避免拿半截页面 resp.encoding resp.apparent_encoding or utf-8 return resp.texttimeout10是必给的不设 timeout 的请求可能挂几分钟放进定时任务后这是最让人头疼的现场之一。apparent_encoding用来兜底页面声明编码和实际不一致的情况北京本地历史站点 GBK 和 UTF-8 混用很常见这一行能省掉一大半乱码问题。接着解析表格def parse_flow_table(html: str) - list[dict]: soup BeautifulSoup(html, html.parser) records [] for row in soup.select(table tr): # 所有数据行 cells [c.get_text(stripTrue) for c in row.find_all([td, th])] if len(cells) 3: continue # 过滤空行和表尾说明行 records.append({ date: cells[0], line: cells[1], raw_flow: cells[2], }) return recordssoup.select(table tr)拿到页面里所有表格行get_text(stripTrue)把单元格里的换行和空格清掉。len(cells) 3的过滤很关键很多表格末尾有空行和「数据来源」说明行靠列数先粗过滤一轮比写复杂正则可靠。如果一次会话要抓多个页面用 Session 复用连接更划算requests 会自动维护连接池from requests.adapters import HTTPAdapter def make_session() - requests.Session: session requests.Session() adapter HTTPAdapter(pool_connections5, pool_maxsize10, max_retries2) session.mount(https://, adapter) session.headers.update(HEADERS) return sessionpool_maxsize10表示同一域名下最多复用 10 个连接max_retries2让瞬时网络错误自动重试两次页面 4xx/5xx 会在raise_for_status()那里暴露出来不会静默吞掉。提示抓取前确认目标页面允许被抓取。这类站点一天更新一次爬虫一天跑一次就够不需要也不需要高频请求。2.3 页面是 JS 渲染的怎么办先找 XHR再考虑浏览器方案如果 requests 拿回来的 HTML 里没有表格、只在浏览器里能看到数据说明数据是前端 JS 动态渲染的。这时候的正确顺序是打开开发者工具的 Network 面板筛选 XHR/Fetch看表格数据是不是通过一个独立接口加载的。绝大多数这类页面的数据来自一个 JSON 接口直接在 Python 里请求那个接口反而比解析 HTML 更省事。这算入门级的 JS 反爬应对思路不跟渲染对抗去找数据真正的家这也是 JS 逆向里最基础的一课。只有接口也做了加密、或者数据直接写死在 JS 脚本变量里才需要上 Playwright / Selenium 这类浏览器自动化。像本项目这种一天一次的客流统计浏览器方案成本和维护量都偏高第一选择永远是静态 HTML 或 XHR 接口。找不到就换数据源比硬啃逆向划算得多。3. 清洗和转换把「1,250.6万」变成 ECharts 能直接消费的 JSON3.1 统一数字格式去千分位、去单位、补空值爬下来的raw_flow是字符串画图前必须转成 float。三类常见写法和对应的处理规则摆在一起看最清楚原始文本清洗后处理步骤1,250.6万人次1250.6去逗号 → 去「万人次」→ float12,506,000人1250.6去逗号 → 去「人」→ float → 除以 10000— 或 None直接置空画图时显示断点168.3万乘次168.3去「万乘次」→ float对应的清洗函数很直白import re def normalize_flow(raw: str) - float | None: text re.sub(r[,\s], , raw) # 去掉千分位逗号和空白 if not text or text in {-, —, --}: return None m re.search(r([\d.]), text) # 提取第一个数字 if not m: return None value float(m.group(1)) if 万 in text: return value # 本来就是万人次 if 人 in text or 乘次 in text: return value / 10000 # 原单位是人换算成万人次 return valuere.search(r([\d.]), text)只提取数字部分单位靠关键字判断这样「万」「万人」「万乘次」三种写法都能落到同一个单位。注意负数和带括号的写法个别页面会用「(1,250.6)」表示负值或修正值遇到这种情况先打日志人工确认不要默认当正数处理。3.2 按日期补齐时间轴缺的日期用空值而不是跳过客流数据偶尔会断更一两天。如果直接按抓到的日期画图X 轴会少点折线会把两个不相邻的日期连起来看起来像「暴跌」其实是「没数据」。常见做法是生成完整日期序列再对齐from datetime import datetime, timedelta def fill_date_axis(dates: list[str]) - list[str]: start datetime.strptime(dates[0], %Y-%m-%d) end datetime.strptime(dates[-1], %Y-%m-%d) axis [] cur start while cur end: axis.append(cur.strftime(%Y-%m-%d)) cur timedelta(days1) return axis对齐的时候缺失日期在线路数据里写NoneECharts 的折线默认对 null 值断开而不是自动连线图上看到的缺口就是真实的「那天没数据」。日期字符串统一用strftime生成零填充的YYYY-MM-DD避免「2024-5-1」和「2024-05-01」并存导致排序错乱字符串比较在 ISO 格式下就是时间序比较后面做自检也方便。3.3 设计 JSON 结构数组对齐下标前端零组装数据要同时支持「总量趋势折线」和「分线对比」JSON 最好组织成 X 轴数组加若干 series 数组的形式def build_payload(records: list[dict]) - dict: # 按 (date, line) 聚合成 {date: {line: flow}} daily: dict[str, dict] {} line_names: set set() for rec in records: daily.setdefault(rec[date], {})[rec[line]] rec[flow] line_names.add(rec[line]) days fill_date_axis(sorted(daily.keys())) # 每条线一个数组按下标与 days 对齐缺的日期给 None lines {name: [daily.get(d, {}).get(name) for d in days] for name in line_names} # 每日总量 当日各线之和约等于官方发布的路网客运量 total [round(sum(daily[d].values()), 1) if d in daily else None for d in days] return { meta: {unit: 万人次, update_time: date.today().isoformat()}, days: days, total: total, lines: lines, }输出用标准库就能完成import json def dump_payload(payload: dict, path: str) - None: with open(path, w, encodingutf-8) as f: json.dump(payload, f, ensure_asciiFalse, indent2)ensure_asciiFalse必须写否则中文线路名会变成\uXXXX转义序列前端显示没问题但人读 JSON 排错时非常痛苦。3.3.1 为什么是 JSON 不是 CSVCSV 适合给人看和进 Excel但前端 fetch 拿 CSV 还得自己 split、转类型而且日期缺口的表示很别扭。JSON 数组天然匹配 ECharts 的 series.datameta里还能带单位和更新时间页面标题、Y 轴单位都可以从同一个文件读换数据源时不用改 JS。这个项目规模小一个flow.json就是全部数据层不需要数据库。3.4 数值自检总量和分线之和要对得上地铁客流有个内在约束路网客运量等于各线路客运量之和每条线的客运量含进站和换乘。这个约束就是最好的自检断言def verify_total(total: float, lines_sum: float, tolerance: float 0.03) - bool: if total 0: return False diff abs(total - lines_sum) / total return diff tolerance偏差超过 3% 就说明分线数据抓漏了或者总量记录的是修正值打印出各线数字人工核对。这个检查放在每天定时任务的末尾比画完图发现少了一条线再回头排查快得多。自检逻辑也可以在 build_payload 之后单独调一次把失败信息写进日志文件。4. 用 JS 统计图把客流画出来ECharts 折线图与柱状图的落地参数4.1 为什么是 ECharts中文场景下的默认选项客流统计图的核心诉求是时间轴趋势、多系列对比、可缩放ECharts 在这三件事上都是开箱即用折线图带 dataZoom 缩放tooltip 支持 axis 触发中文文档和示例齐全。Chart.js 更轻但缩放和中文生态都要自己补。所以这个项目用 ECharts固定一个 5.x 发行版直接下载echarts.min.js放到本地static/目录页面离线也能打开不依赖 CDN 可用性。版本宁旧勿杂换版本前先看 breaking changes。4.2 页面骨架和最小配置fetch JSON 后 setOption页面只需要一个div加一段脚本!DOCTYPE html html langzh-CN head meta charsetUTF-8 title北京地铁客流量统计/title script srcstatic/echarts.min.js/script /head body div idflow-chart stylewidth:100%;height:600px;/div script fetch(data/flow.json) .then(resp resp.json()) .then(payload { const chart echarts.init(document.getElementById(flow-chart)); chart.setOption(buildOption(payload)); window.addEventListener(resize, () chart.resize()); }); function buildOption(payload) { return { tooltip: { trigger: axis }, legend: { data: [路网客运量] }, grid: { left: 60, right: 30, top: 50, bottom: 70 }, xAxis: { type: category, data: payload.days }, yAxis: { type: value, name: payload.meta.unit }, dataZoom: [ { type: inside }, { type: slider, height: 18, bottom: 10 } ], series: [{ name: 路网客运量, type: line, data: payload.total, smooth: true, areaStyle: { opacity: 0.15 } }] }; } /script /body /htmltooltip.trigger: axis让鼠标滑到某一天时同时显示所有系列该天的值比默认的 item 触发更适合时间轴数据。grid四个边距是给 Y 轴单位和底部 dataZoom 留空间不设的话缩放条会压到 X 轴文字。smooth: true在日频数据上看着舒服但如果要严格展示真实数值就不要开平滑是曲线拟合会轻微改变顶点位置。注意 fetch 在file://协议下会被 CORS 拦截直接用 Python 起一个静态服务在项目根目录运行python -m http.server 8000然后访问http://localhost:8000。这一步是绝大多数新手被卡住的地方不是代码错了是协议不对。4.3 必调参数dataZoom、tooltip 和空值断点参数推荐值作用踩坑点dataZoom.typeinside slider 双配鼠标滚轮缩放 底部滑块拖拽只配 inside 时新手找不到拖拽入口tooltip.triggeraxis时间轴联动显示所有系列item 模式只看单条线series.data 中的 null保留 null断点不连线空缺可见用 0 填充会被当成真实谷值xAxis.boundaryGaptruecategory 默认折线点落在刻度中间柱状图建议改 false 让柱子靠边空值处理值得单独强调清洗阶段置成的None在 JSON 里序列化成nullECharts 默认对 null 断线。不要为了「美观」把 null 填成 00 会画出一个真实的低谷看图的人会误以为当天客流崩了。数据缺口宁可明晃晃地断在那里也不要伪造一个数。4.4 叠加分线对比多系列数组和可滚动图例总量之外再加几条重点线路对比series 扩展成数组即可const focusLines [10号线, 1号线, 4号线]; const lineSeries focusLines .filter(name payload.lines[name]) .map(name ({ name: name, type: line, data: payload.lines[name], smooth: true, lineStyle: { width: 2 } })); chart.setOption({ legend: { type: scroll, data: [路网客运量, ...focusLines] }, series: [ { name: 路网客运量, type: line, data: payload.total, lineStyle: { width: 3 } }, ...lineSeries ] });线路多的时候图例撑满顶部legend.type: scroll让图例可以横向滚动。每周哪条线客流最大、哪条线周末降幅最明显这类结论用多系列折线一眼就能看出来。如果后续想把总量和分线增长率画在一起才需要双 Y 轴在yAxis传一个数组给增长率系列配yAxisIndex: 1。本项目单位一致不加第二根轴图面更干净。5. 定时更新与全链路自检让统计脚本自己跑不用人盯着5.1 一个命令把抓取、清洗、输出三步串起来把前面章节的模块串成run_pipeline.py入口只做三件事抓页面、清洗、写 JSON。再加一个sys.argv接收日期参数方便手动补数import sys from datetime import date def main() - None: # 不带参数默认今天实际数据是昨天发布所以通常会传昨天 target sys.argv[1] if len(sys.argv) 1 else date.today().isoformat() html fetch_html(METRO_FLOW_URL) records parse_flow_table(html) normalized [normalize_flow(r[raw_flow]) for r in records] payload build_payload(records) # 含清洗、日期对齐、总量自检 dump_payload(payload, data/flow.json) print(fupdated to {target}, lines{len(payload[lines])}) if __name__ __main__: main()python run_pipeline.py 2024-05-20可以手动补历史某天打印的lines数量是最快的健康检查少了几条线一眼就能看出来。这里给 Python 脚本传参跟python -m http.server 8000传端口号是同一个机制sys.argv从 1 开始取argv[0]是脚本自身路径。5.2 多个页面并发抓ThreadPoolExecutor 的边界在哪里数据源如果拆成多个页面总量一个、分线一个、历史页一个串行抓会很慢。这个量级用concurrent.futures.ThreadPoolExecutor就够了from concurrent.futures import ThreadPoolExecutor, as_completed URLS [METRO_FLOW_URL, METRO_LINE_URL, METRO_HISTORY_URL] def fetch_all(urls: list[str]) - dict[str, str]: results {} with ThreadPoolExecutor(max_workers3) as pool: future_map {pool.submit(fetch_html, u): u for u in urls} for future in as_completed(future_map): url future_map[future] try: results[url] future.result() except Exception as exc: print(ffailed: {url} - {exc}) return resultsmax_workers3不是越大越好目标站点通常是小规模服务器三个并发足够且不构成压力。并发设计的真正考点在这里先想清楚要抓多少个源、源的更新频率是多少再决定串行还是并发。几十个页面以内、一天跑一次ThreadPoolExecutor 就是正确答案几百上千个源才需要考虑分布式爬虫和消息队列那是另一个量级的事别在这个项目里提前造。5.3 自检脚本加 crontab数据新鲜度、结构完整度一次测完最后写一个healthcheck.py每天跑完管道后自动验证JSON 里最新日期是不是前一天、分线数量对不对、数据源页面结构有没有变。页面结构检查用字符串包含比整页解析更快更抗变化import json from datetime import date, timedelta def healthcheck() - int: with open(data/flow.json, encodingutf-8) as f: payload json.load(f) yesterday (date.today() - timedelta(days1)).isoformat() if payload[days][-1] yesterday: print(latest date is stale:, payload[days][-1]) return 1 html fetch_html(METRO_FLOW_URL) if 昨日客运量 not in html: # 相当于 JS 里 string.includes(...) print(page schema changed, check selector) return 1 print(healthcheck ok, lines:, len(payload[lines])) return 0 if __name__ __main__: raise SystemExit(healthcheck())payload[days][-1] yesterday用字符串比较日期ISO 格式下字典序就是时间序不需要转 datetime。页面改版时表头措辞一变昨日客运量 not in html立刻报警比等图花了再排查快一整天。这套自检放在定时任务末尾输出到日志即可。定时调度用系统 crontab 就够了上午客流数据发布后跑一次比在 Python 里再塞一个调度器少一个常驻进程0 10 * * * cd /path/to/metro-flow python run_pipeline.py logs/run.log 21 python healthcheck.py logs/check.log 21保证上一步成功才执行自检日志落文件方便回查。数据文件用 git 管理每次跑完提交一次一年之后这份 JSON 就是带日期的客流时间序列接下周客流预测模型时它就是最值钱的输入。爬虫技术和 AI 结合的价值也得从这样一份被验证过的历史数据开始。本文还有配套的精品资源点击获取