简介这是一份基于Python的百度迁徙数据爬取脚本目标是以最小代码量实现人口流动数据的自动化采集适合网络爬虫初学者巩固requests、HTML解析与pandas数据处理技能也适合交通、公共健康等领域的分析人员快速获取数据。压缩包仅含1个py文件整体大小只有2KB将春运场景下的数据抓取逻辑集中在一个脚本中便于阅读和二次修改也避免了多文件工程带来的学习负担。目前已有1564人学习与下载关注度良好。脚本不仅演示了根据百度迁徙页面结构定位日期、城市、迁入迁出量等字段的具体方式还结合实例涵盖了正则表达式匹配、字符串清洗、pandas表格导出以及请求间隔设置等反爬应对细节。用户拿到后直接运行即可观察效果也可以自行扩展命令行参数、代理IP管理或历史数据的批量抓取整体是一份麻雀虽小但五脏俱全的实战样例。1. 一个抓节假日人口迁徙数据的爬虫项目它到底能干什么春节前一周你打开某个地图平台的迁徙页面能看到每个城市的人员迁入迁出排行数据每天都在更新。但页面只给你看不给下载。想做返乡趋势分析、城市活力对比或者给论文凑一张“春运人口流动图”你就得手动一条条抄。这个项目就是干这个的一个专门针对百度迁徙数据的数据抓取工具输入日期、城市、迁入或迁出方向它把网页背后 JSONP 接口里的榜单数据自动抓下来清洗成标准 CSV 表格。项目本身不大技术栈也很朴素适合两类人一是想快速拿到城市人口流动数据做分析或可视化的开发者二是刚学完 requests 基础、想找一个“有真实数据源、有完整代码链路”的练手爬虫项目。看完这篇你能跑通一次完整抓取也能理解从 URL 构造、JSONP 解析到数据落盘的全过程最后还能按自己的需求改成批量定时抓取。2. 项目结构与技术选型requests 之外还有哪些关键设计2.1 为什么不选 Scrapy 而是 requests 组合这类轻量级数据抓取项目很多人的第一反应是上 Scrapy。我的判断是没必要。目标接口就一两个返回格式是固定的 JSONP没有登录、没有加密参数、没有翻页逻辑Scrapy 的项目骨架和中间件配置在这里反而是负担。用 requests 做请求、正则加 json 模块做解析、csv 模块做落盘三五百行代码就能把事讲清楚出问题也好排查。处理大规模抓取时scrapy 框架的并发、去重、断点续爬肯定更稳但本项目单次抓取几十个城市的接口requests 加 time.sleep 已经足够。整个项目的核心依赖只有 requests 一个库pandas 只是可选用来做数据清洗不装也能跑。这种“能少依赖就少依赖”的选择让项目在任何一台装好 Python 3.8 以上的机器上都能立刻运行。2.2 数据接口是怎么定位的一次抓包看穿 JSONP这个项目最核心的前期工作不是写代码而是找到数据接口。某地图平台的迁徙页面是单页应用页面数据通过浏览器异步请求加载。打开浏览器开发者工具切到 Network 面板刷新页面过滤 XHR 请求能看到一类名字带有 topflow、flowcity 字样的请求响应内容不是标准 JSON而是被一个回调函数包裹起来的 JSON 文本这就是典型的 JSONP 格式。服务器返回的内容类似callback({...})浏览器通过动态 script 标签加载这段文本并执行回调函数从而拿到数据。接口的关键参数有三个date 表示数据日期格式是 YYYYMMDDcity 表示城市编码type 表示方向迁入迁出分别对应不同取值。不同版本页面参数名可能不同常见做法是打开一个请求复制它的 query string 保存下来作为模板再在代码里动态替换日期和城市。这里不给出固定域名因为它变过好几次正确姿势是每人跑一次抓包把你看到的 URL 填进代码。2.3 目录结构按数据流向拆分层项目拆成四个模块加两个入口目录结构如下baidu_migration/ ├── config.py # 配置城市、日期、方向、输出目录、请求间隔 ├── requester.py # 请求层UA 池、限速、超时、重试 ├── parser.py # 解析层JSONP 清理、榜单数据归一化 ├── storage.py # 存储层CSV/Excel 落盘 ├── main.py # 入口单日单城市抓取 └── run_batch.py # 入口批量时间段抓取分层思路很直接config 管参数requester 管网络parser 管解析storage 管输出入口文件负责串流程。这样设计的好处是哪天接口字段变了你只需要动 parser.py哪天想改输出格式你只需要动 storage.py。爬虫项目最大的维护成本是“数据源变化”把变化点隔离在单一模块里是这一类项目的通用设计思路。3. 核心模块拆解从 URL 构造到 CSV 落盘的数据链路3.1 请求层UA 池与限速别当那台最快最吵的客户端requester.py 的职责是发请求并拿到原始文本。这层做两件事伪装客户端和降低请求频率。import random import time import requests # 常见的 UA 池覆盖桌面端和移动端随机取用 UA_POOL [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36, Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.0 Mobile/15E148 Safari/604.1, ] def fetch_data(url: str, interval: float 1.5, retries: int 3) - str: headers { User-Agent: random.choice(UA_POOL), Accept: */*, Referer: https://qianxi.baidu.com/, # 按你抓包时看到的页面地址填 } for attempt in range(retries): try: time.sleep(interval) # 每次请求前强制等待给服务器留出喘息时间 resp requests.get(url, headersheaders, timeout10) resp.raise_for_status() return resp.text except requests.RequestException as e: if attempt retries - 1: raise time.sleep(2 * (attempt 1)) # 指数退避第一次等2秒第二次等4秒 return 参数说明interval 控制两次请求的最小间隔本项目默认 1.5 秒批量抓取 50 个城市也就是一分钟出头retries 是失败重试次数网络抖动或服务端临时返回 5xx 时能自动恢复Referer 头模拟浏览器来源具体的值以你抓包看到为准不要照抄注释里的占位地址。整体逻辑是先睡够再请求请求失败按指数递增等待后再试避免连续撞同一堵墙。这里用记录目标站点频率限制的经验超过一定频率接口不会直接拒绝而是返回空数据那种“请求成功但结果全是空的”情况比 HTTP 报错更难排查。3.2 解析层JSONP 清理用首尾花括号而不是回调函数名JSONP 格式的文本长这样callbackFn({errno:0,data:{list:[{city_name:上海,value:0.98}]}})最常见且容易采坑的解析方式是用正则精确匹配回调函数名callbackFn后面的内容。但这个函数名是动态生成的今天叫jQuery_123456明天可能叫jsonp_789一旦匹配失败整个解析就翻车。更稳的做法是忽略回调名直接定位文本里第一对首尾花括号import json import re def parse_jsonp(text: str) - dict: 从 JSONP 文本中截取完整 JSON 对象并解析为 dict start text.find({) end text.rfind(}) if start -1 or end -1 or end start: raise ValueError(f无法定位 JSON 边界原始文本前 100 字符: {text[:100]}) return json.loads(text[start:end 1])原理说明JSONP 的固定结构就是“回调名(JSON)”无论回调名怎么变整个 JSON 对象总是从第一个{开始到最后一个}结束。用find和rfind截取就绕开了对动态回调名的依赖。这在接口大概率升级但 JSON 结构保持稳定的场景下特别有用。如果哪天解析失败错误信息里带上原始文本的前 100 个字符能快速确认是接口改了格式还是返回了异常提示这个小习惯能省大量排查时间。3.3 解析层榜单数据归一化为标准行记录接口返回的 JSON 嵌套层数不浅顶层是 errno 和 datadata 里又套 listlist 每一项是一个城市的迁移比例。这个结构是这类地图数据平台的常见做法字段名可能有出入先抓一次原始结果看清楚再定代码。下面是按常见字段写的归一化逻辑def extract_top_cities(data: dict) - list[dict]: 把嵌套 JSON 中的迁徙榜单拍平成行记录 if data.get(errno) ! 0: raise ValueError(f接口返回错误码: {data.get(errno)}) result data.get(data, {}).get(list, []) rows [] for item in result: rows.append({ rank: item.get(rank, ), # 排名部分页面不返回 city: item.get(city_name, ), # 迁入/迁出目标城市名 province: item.get(province_name, ), # 目标省份用于区分同名城市 value: item.get(value, 0), # 迁移指数占比类数值 }) return rows这段代码有三个目的。第一把嵌套 JSON 转成一行一行的记录方便写入表格。第二补全省份字段因为不同省份可能有同名地级市只存城市名后面做数据分析时会遇到歧义。第三把 value 字段从原始值里显式取出来后续无论是直接分析还是导入数据库字段类型都清晰。真实世界里每个榜单大概返回 100 条左右的迁移城市记录value 是比例数值而非实际人数这一点后面还会单独强调。3.4 存储层CSV 落盘UTF-8 带 BOM 防 Excel 乱码storage.py 承担把归一化后的记录写入本地文件的工作。import csv from pathlib import Path def save_rows(rows: list[dict], path: str) - None: 写入 CSV 文件自动创建上级目录使用 utf-8-sig 编码 file_path Path(path) file_path.parent.mkdir(parentsTrue, exist_okTrue) fieldnames [rank, city, province, value] with open(file_path, w, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnamesfieldnames) writer.writeheader() writer.writerows(rows)三个细节用Path.parent.mkdir自动创建输出目录避免因目录不存在直接报错用newline防止 Windows 下 CSV 出现多余空行用utf-8-sig编码而不是utf-8是因为前者写入的 BOM 头能让 Excel 直接双击打开不乱码这是处理中文数据时一个很小的血泪经验。3.5 主流程串联main.py 里的一次完整抓取main.py 把请求、解析、存储串成完整链路from requester import fetch_data from parser import parse_jsonp, extract_top_cities from storage import save_rows def build_url(city_city: str, date: str, move_type: str) - str: 按抓包得到的实际接口模板构造 URL这里只给示意结构 return fhttps://你抓包拿到的域名/topflow?date{date}city{city_city}type{move_type} def main(): url build_url(北京, 20250210, moveIn) text fetch_data(url, interval1.5) data parse_jsonp(text) rows extract_top_cities(data) save_rows(rows, ./output/北京_20250210_迁入TOP100.csv) print(f已保存 {len(rows)} 条记录)整个执行流程是这样的main 把日期、城市和方向加工成完整 URLrequester 带着随机 UA 和限速逻辑请求接口拿到 JSONP 文本后 parser 将其清理成 dict 并拍平为行记录最后 storage 写入带 BOM 的 CSV 文件。一次抓取就是这四个模块的线性协作逻辑非常直白。抓取多城市时只需要循环调用 main 内部的这四个步骤不用改任何模块代码。4. 运行手册环境准备、参数配置与一次完整抓取4.1 环境准备Python 3.8 以上安装 requests在项目根目录创建虚拟环境并安装依赖python3 -m venv venv source venv/bin/activate # Windows 下执行 venv\Scripts\activate pip install requests如果后续要跑批处理脚本里的数据合并逻辑再装 pandas 也不迟单日抓取只需要 requests。项目没有其他第三方依赖这一点对刚接触爬虫的新手比较友好不用为了一两个接口去配置庞大的爬虫框架。4.2 配置项说明改 config.py 一处就够所有抓取参数集中在 config.py 里这是项目的统一入口# 抓取目标城市按需增减 CITY_LIST [北京, 上海, 广州, 深圳, 成都, 重庆, 西安, 武汉] # 抓取数据日期格式必须为 YYYYMMDD DATE_TO_FETCH 20250210 # 迁徙方向moveIn 表示迁入moveOut 表示迁出 MIGRATION_TYPE moveIn # 结果输出目录 OUTPUT_DIR ./output # 请求间隔秒批量抓取时注意不要调太低 REQUEST_INTERVAL 1.5参数说明CITY_LIST 支持任意多个城市接口按城市返回该城市的迁入或迁出榜单DATE_TO_FETCH 需要是接口覆盖范围内的日期春运或节假日是数据回补期非节假日的数据可能需要核对是否发布完全MIGRATION_TYPE 决定榜单方向迁入迁出各一份数据REQUEST_INTERVAL 是限速参数调成 0.2 可能抓得快但容易被服务端限制调成 3 更安全。4.3 执行抓取跑起来看输出从命令行进入项目目录执行python main.py正常输出如下已保存 100 条记录然后打开 output 目录下带日期的 CSV 文件能看到四列数据rank、city、province、value。以“北京 迁入”为例榜单前列应该是周边通勤城市比如保定、廊坊、天津这些城市的 value 占比较高迁出榜单前列类似。看到这种符合地理直觉的数据基本可以判断抓取成功。4.4 验证抓取结果三招自检拿到 CSV 后不要急着做分析先做三件事第一数行数正常应在 100 行左右如果只有几行或为零先处理日期或接口问题第二抽查城市名榜单城市和迁入迁出方向是否匹配比如成都的迁入榜单春节期间应以下沉地级市为主第三检查 value 列数值范围应在 0 到 100 之间的小数如果出现负数或超大值说明解析逻辑匹配错了字段。这三招能过滤掉绝大多数“抓到了但数据是错的”的隐蔽问题比直接拿数据画图省力得多。5. 踩坑排查笔记接口字段、频率限制与“空数据”玄学5.1 现象请求返回 200但解析出来 list 是空的第一次抓取时最容易遇到。代码没有报错打印出 JSON 后发现 data.list 是空数组或者只返回了一个名字为“全国”的汇总项。原因分析接口的数据按日期发布不是所有日期都保证有完整数据。非迁徙高峰期的个别日期平台可能不生成完整的城市榜单更常见的情况是请求参数里的日期格式不对比如把 2025-02-10 按习惯写成了带横杠的格式接口直接忽略并返回空。这是这类接口对参数非法时的常见响应方式静默失败比显式报错更坑。解决办法先手动打开平台页面在页面上把日期切到目标日期如果可以正常展示榜单再抓包看真实请求的 date 参数值照着改成 YYYYMMDD 格式。如果页面本身没有数据就换一个日期再试。从那以后我每次写日期参数都会先打印一遍最终 URL确认没有转义或拼接问题。5.2 现象正则提取 JSONP 内容时频繁报错回调函数名解析失败报类似Expecting value: line 1 column 1的错误。原因分析平台不定期更换 JSONP 回调函数名的生成规则用固定正则r^callback\(去匹配一旦函数名变化就截取不到 JSON。JSONP 本来就是一段“伪装成脚本的 JSON”回调名只是个传输外壳不值得依赖。解决办法切换到首尾花括号定位法即前面 parser.py 里的实现。用text.find({)和text.rfind(})直接截取 JSON 对象完全不关心回调名。这个改动让解析逻辑在接口多次升级后依然稳定属于一劳永逸的解法。如果你还想更保险可以在截取后用json.loads做一次语法验证失败时打印原始文本前 200 个字符供人工确认。5.3 现象连续抓几十个城市后接口开始返回异常数据批量跑 CITY_LIST 里的城市一开始数据正常跑到第 30 个左右突然整批失败或者返回的 value 全部相同。原因分析触发了基于频率的反爬机制。你从浏览器正常访问接口频率一分钟几次脚本接手后变成一秒一次虽然 UA 是随机的但单位时间内的请求数量暴露了自动化特征。服务端不会直接封 IP比较多伪装成“返回一个兜底结果”或者“随机抽掉部分字段”这种隐蔽限制对没有经验的人来说特别像玄学。解决办法把 REQUEST_INTERVAL 调到 2 秒以上并且加上 5.1 节提到的失败重试和指数退避。重试的逻辑不是立即再来一次而是按 2 秒、4 秒、8 秒的间隔逐步拉大让服务端主动把你从限速候补名单里放出来。我从实践中得到的经验是脚本跑批量任务时保留一个“已成功城市清单”每次重跑先跳过已完成的只补抓失败的这样就算中途被限制也能断点续跑。5.4 现象两个城市的榜单里出现同一个区县名称比如榜单里出现“朝阳区”但你可能不确定它属于哪个市。原因分析不少地名在不同省份有同名区县或者同一城市下辖的县级市名称相近。接口返回的城市名是展示名不保证唯一。如果你后续做跨城市对比分析直接拿城市名字段做 join 会出现数据串线。解决办法保存数据时把 province 字段一并写入 CSV分析阶段用“省份 城市”拼接作为唯一主键。这个改动在写 storage.py 时已经预设好了fieldnames 里就有 province 字段。做分析时优先使用拼接列可以绕开绝大多数同名歧义。5.5 现象把 value 当成实际迁徙人数画出的图比例失真这是数据理解层面的踩坑代码层面没有任何报错。原因分析接口返回的 value 是迁徙指数不是真实人头数。它表示的是“该城市对目标城市的相对迁移热度”经过平台归一化处理能用于横向比较不同城市的热度差异但不能直接说“某城市迁入北京 10 万人”。指数和实际客流之间可能存在非线性映射关系。解决办法做趋势分析时只比较同一指标在不同日期的相对变化比如“北京迁出热度在节前一周上升了 20%”而不是换算成绝对人数。如果项目说明里写了“百分比”“比例”字样一律按相对值处理。这也是做数据可视化时最常被低估的一个点。6. 二次开发思路从单日快照到趋势分析小工具跑通单日抓取只是起点。把这个小项目改造成一个“能追踪迁徙趋势”的小工具需要三步扩展难度依次递增。第一步是批量抓取。写一个 run_batch.py遍历日期范围逐日调用同样的抓取逻辑把输出文件名按日期组织import pandas as pd from datetime import datetime, timedelta from requester import fetch_data from parser import parse_jsonp, extract_top_cities from storage import save_rows def run_batch(start_date: str, days: int, move_type: str): date_list [(datetime.strptime(start_date, %Y%m%d) timedelta(daysi)) .strftime(%Y%m%d) for i in range(days)] for date in date_list: try: url build_url(北京, date, move_type) text fetch_data(url, interval2.0) rows extract_top_cities(parse_jsonp(text)) save_rows(rows, f./output/beijing_{date}_{move_type}.csv) except Exception as e: print(f{date} 抓取失败: {e}) continue第二步是合并与分析。把多天的 CSV 读进来按 city 和 date 生成透视图计算每个城市的排名变化幅度。这里可以引入 pandas一行pd.pivot_table就能把“日期 × 城市 × 指数”变成宽表再用移动窗口看出“返乡高峰从哪天开始”。第三步是可视化。用 pyecharts 的地图组件把城市间的迁入迁出画成流线图需要先准备一份城市坐标表或城市编码表把 CSV 里的城市名关联到地图对应区域。如果不想引入前端直接输出按日排名的 Markdown 表格也能满足需求核心是分析思路而不是花哨图表。这个项目的二次开发上限不低可以把抓取逻辑封装成定时任务每天自动跑可以把数据存入 SQLite 做长期累积也可以把 request 层换成 httpx 支持异步并发。我自己的习惯是每次改动后抓一个真实日期的数据对照平台页面核对一遍榜单前几名确认解析和数据两个环节都没被改坏。从那以后我每次做这类外部数据源的抓取脚本都会先在本地保存一份原始响应样例并在代码注释里记录抓包时间方便下次接口升级时快速 diff。希望帮到你。本文还有配套的精品资源点击获取