简介一份基于 Python 的百度迁徙数据爬取脚本面向需要采集人口迁徙流向的研究者、数据分析师及爬虫初学者目标是通过自动化方式替代手动复制网页数据快速获得可用于分析的结构化信息。资源压缩包仅含一个 Python 脚本整体约2KB代码精简但覆盖请求发送、HTML解析、关键字段提取等常用爬虫步骤同时预留日期参数入口方便按指定时段抓取。脚本整体学习门槛不高已有1564人学习下载适合用来理解数据采集到表格化输出的完整流程。运行时可借助 requests、lxml 等常见库完成页面请求与解析再结合 pandas 将结果整理为 CSV 或 Excel 表格针对页面中的动态加载内容读者可自行扩展 Selenium 或代理池应对进一步巩固反爬处理能力是入门 Python 爬虫与数据分析的实用范本也适合作为春运等场景下迁徙数据研究的前置工具。1. 从界面点选到定时拉取地图迁徙指数爬虫到底解决了什么节假日前后打开地图应用迁徙指数热力图总在社交平台刷屏。对做数据分析的人来说新鲜感会过去真正有价值的事是把指数背后的城市间流动数据按天拉下来存成表再拿去和交通、商业、招聘数据做关联。这个项目的定位就是把某个地图平台的迁徙指数抓下来一个不依赖浏览器界面的网络爬虫直接请求后端接口把数据落成结构化 CSV。整个项目用 Python 编写依赖控制在几个库以内核心链路是抓包定位接口 - 构造请求 - 解析 JSON - 增量落盘。每个环节切成独立模块拆也好拆换也好换。对想练手网络爬虫的开发者来说这是一份能跑通完整流程的参考代码要做区域分析、门店选址或者人口流动研究的人也可以拿它当数据底座。我拿到这份源码的第一印象是它没有把抓取逻辑和业务字段写死在一起四个模块各管一件事后续换存储、加省份维度都很直接。这篇解析就按准备 - 运行 - 改造的顺序把它拆开讲你照着走完一遍数据结构长什么样、接口参数怎么组、二次开发从哪里下手心里基本有数。2. 准备与架构设计技术选型、目录结构和接口定位2.1 技术选型requests 够用Selenium 可以放一边先交代技术栈Python 3.9 以上requests 负责网络请求标准库的 csv、sqlite3 负责落盘argparse 处理命令行参数日志走 logging。没有引入 Scrapy 或 Playwright不是它们不好而是这个场景的需求太明确了——拉接口返回的 JSON。数据源本身是接口直接渲染不需要浏览器执行脚本requests 一个 Session 就能完成九成的工作。用 Selenium 打开浏览器渲染页面数据没多拿启动浏览器和等待页面加载的时间反而翻了几倍。如果你经常写爬虫大概率见过这种场面硬要用 Selenium 跑一个纯 JSON 接口结果每次启动都卡在浏览器加载上。我的习惯是先看 DevTools 的 Network 面板确认返回的是接口数据还是页面直出再决定要不要上浏览器自动化工具。这个项目在抓包阶段就确认目标是 XHR 请求所以直接选了轻量方案。依赖少带来的另外一个好处是部署简单找一台装了 Python 的机器克隆代码就能跑不涉及 Redis、消息队列这些基础设施。数据量到这个级别单机单进程绰绰有余。2.2 目录结构一个爬虫项目的文件边界整个项目的文件不多但边界切得很清楚migration_spider/ ├── config.py # 请求头、基础地址、采集日期范围 ├── request_client.py # 请求封装会话、超时、重试 ├── spider_main.py # 入口调度省份和日期串联各模块 ├── parser.py # 解析接口返回的 JSON ├── storage.py # 落盘CSV 增量写入 / SQLite 归档 ├── requirements.txt └── logs/ # 运行日志config.py 只存配置不含业务逻辑request_client.py 只管请求的发出和异常重试parser.py 不关心数据从哪来只负责把 JSON 转成统一结构的记录storage.py 把记录写入文件或数据库。spider_main.py 作为入口把上面几个模块按顺序串起来。这个划分的妙处在于如果你想把输出从 CSV 换成数据库只需要改 storage.py想增加一个城市的采集改 config.py 的城市列表就行。各模块之间通过简单的数据结构通信没有复杂的类继承关系读代码的成本很低。2.3 定位数据接口抓包是第一生产力拿到项目代码之后最先要看懂的是数据从哪来。打开目标网页的迁徙功能页面按 F12 打开 DevTools切到 Network 面板勾选 Fetch/XHR刷新页面然后点一个省份进去。此时你会看到几个新的 JSON 请求冒出来逐个点开看返回内容找到里面包含城市名称和数值的那一个那就是核心数据接口。找到之后右键该请求选择 Copy - Copy as cURL把这段 cURL 信息保持下来。它里面包含了完整的请求地址、Headers、Query 参数这是后续构造代码请求的基准。之后在代码里调试时可以先写死一组参数跑通再改成循环。提示抓包时勾选 Fetch/XHR 能过滤掉图片、CSS 等静态资源请求定位接口更快。2.4 请求参数说明日期、方向、城市编码怎么传从抓包结果里能直接看到这个接口的查询参数常见的有以下几类参数类型含义示例datestring查询日期格式 YYYYMMDD20250101typestring迁入还是迁出方向move_in / move_outcitystring行政区划编码320000pageinteger分页序号1这里有两个容易出错的点。第一参数值必须是行政区划编码而不是汉字城市名传320000而不是江苏否则接口不认。第二date 必须补零成八位传2025-1-1这种格式时部分接口会容错返回空数据表现就是页面有数据但代码请求结果为空。这两个坑在后面的排查章节会重点展开。另外要注意接口返回的 value 是平台定义的迁移指数不是实际人口数量。把它理解成一个反映流动规模相对大小的指标就行分析和呈现时不要写成人次。3. 核心模块拆解请求构造、调度循环与解析落盘的完整链路3.1 请求构造把网页请求翻译成本地代码request_client.py 是所有请求进出的地方。第一步是把抓包得到的 Headers 完整复制到代码里尤其不能漏掉 Referer 和 User-Agent少了这两个字段很容易触发服务端校验。# request_client.py import requests import logging logger logging.getLogger(__name__) class MigrationClient: def __init__(self, base_url: str, headers: dict, timeout: int 10): self.base_url base_url self.timeout timeout self.session requests.Session() self.session.headers.update(headers) def fetch_flow(self, date_str: str, city_code: str, flow_type: str) - dict: params { date: date_str, city: city_code, type: flow_type, } try: resp self.session.get(self.base_url, paramsparams, timeoutself.timeout) resp.raise_for_status() data resp.json() except requests.RequestException as exc: logger.warning(请求失败: date%s city%s type%s err%s, date_str, city_code, flow_type, exc) return {} return data这段代码的逻辑很直白用 requests.Session 维持一个会话对象把所有请求头封装在 session 内部每次调用 fetch_flow 时只需要关注查询参数。Session 与普通 get 请求的区别在于它可以复用底层的 TCP 连接并且自动携带之前响应里 set-cookie 写入的 Cookie这对需要保持会话状态的接口很重要。参数说明base_url 是抓包得到的接口地址headers 是包含完整请求头的字典timeout 设 10 秒是为了避免某个请求卡死拖垮整个任务。返回的数据统一走resp.json()解析成 Python 字典如果请求异常返回一个空字典而不是抛出异常让上层调度不至于因为单次失败直接中断。3.2 调度循环按省份、按日期把任务排开spider_main.py 是入口模块负责把城市列表和日期范围展开成一个个请求任务。代码结构是一个双层循环外层遍历城市编码内层遍历日期# spider_main.py import argparse import logging import time from config import CONFIG from request_client import MigrationClient from parser import parse_flow from storage import CsvStorage def build_date_list(start: str, end: str) - list[str]: 生成 YYYYMMDD 格式的日期序列 from datetime import datetime, timedelta start_dt datetime.strptime(start, %Y%m%d) end_dt datetime.strptime(end, %Y%m%d) dates [] while start_dt end_dt: dates.append(start_dt.strftime(%Y%m%d)) start_dt timedelta(days1) return dates def main(): parser argparse.ArgumentParser() parser.add_argument(--start, requiredTrue) parser.add_argument(--end, requiredTrue) args parser.parse_args() client MigrationClient(CONFIG[base_url], CONFIG[headers]) storage CsvStorage(CONFIG[output_dir]) date_list build_date_list(args.start, args.end) for city_code in CONFIG[city_codes]: for date_str in date_list: payload client.fetch_flow(date_str, city_code, move_in) rows parse_flow(payload, date_str, city_code) storage.save_rows(rows) time.sleep(CONFIG[request_interval]) if __name__ __main__: logging.basicConfig(levellogging.INFO, format%(asctime)s %(levelname)s %(message)s) main()这段代码的设计要点在数据流的单向传递client 拿到字典parser 把字典转成行记录storage 负责写入。每个模块都不感知其他模块内部实现。sleep 放在内层循环末尾是为了控制请求频率避免短时间大量请求触发频控。参数说明--start 和 --end 是必填参数格式 YYYYMMDDbuild_date_list 用 datetime 做日期累加比直接操作字符串更稳妥。CONFIG 里的 city_codes 是你要采集的行政区划编码列表request_interval 是每次请求之间的间隔秒数建议设置在 0.3 到 1 之间根据实际频率限制调整。3.3 数据解析把嵌套 JSON 拍平成记录接口返回的 JSON 通常有两层结构外层是状态码和日期内层才是真正的流动数据。parser.py 做的事情就是把这层嵌套展开成扁平的记录每个记录代表某天某个城市的人口流入量# parser.py def parse_flow(payload: dict, date_str: str, city_code: str) - list[dict]: 把接口返回的 JSON 解析成统一结构的记录列表 rows [] data payload.get(data) or {} flow_list data.get(list) or [] for item in flow_list: rows.append({ date: date_str, src_city: city_code, target: item.get(city_name), value: item.get(value), rank: item.get(rank), }) return rows解析逻辑有三个要点。第一payload.get(data) or {}这种写法是为了处理接口异常时返回空字典的情况避免深层取 key 时报 TypeError。第二字段名的映射写在解析这一层后面存储和业务分析都使用标准字段名即使接口升级改字段名只需要改这一个文件。第三rank 字段保留原始排序后续做 Top 城市分析时能直接用不需要重新排序。3.4 增量落盘CSV 去重和 SQLite 归档的选择storage.py 提供两个写入实现一个是 CSV 追加一个是 SQLite 去重。对于个人研究和轻量分析CSV 够用如果要反复采集同一批数据SQLite 的去重逻辑能显著降低数据清洗成本。# storage.py import csv import os import sqlite3 class CsvStorage: def __init__(self, output_dir: str): self.output_dir output_dir os.makedirs(output_dir, exist_okTrue) def save_rows(self, rows: list[dict]) - None: if not rows: return fieldnames list(rows[0].keys()) file_path os.path.join(self.output_dir, migration_flow.csv) is_new not os.path.exists(file_path) with open(file_path, a, newline, encodingutf-8) as f: writer csv.DictWriter(f, fieldnamesfieldnames) if is_new: writer.writeheader() writer.writerows(rows) class SqliteStorage: def __init__(self, db_path: str): self.db_path db_path self._init_table() def _init_table(self) - None: with sqlite3.connect(self.db_path) as conn: conn.execute( CREATE TABLE IF NOT EXISTS flow ( date TEXT, src_city TEXT, target TEXT, value REAL, rank INTEGER, PRIMARY KEY (date, src_city, target) ) ) def save_rows(self, rows: list[dict]) - None: with sqlite3.connect(self.db_path) as conn: conn.executemany( INSERT OR IGNORE INTO flow (date, src_city, target, value, rank) VALUES (:date, :src_city, :target, :value, :rank) , rows)CSV 实现的细节在追加写入每次跑任务只是往文件尾部追加新记录配合 os.path.exists 判断是否写表头避免重复表头。SQLite 实现的关键是INSERT OR IGNORE它依赖PRIMARY KEY (date, src_city, target)做唯一约束重复拉取同一天、同一个来源城市、同一个目标城市的数据时新记录会被静默忽略。这里的选择建议是如果只是每周拉一次做趋势观察CSV 足够如果打算做长期积累比如连续采集一年SQLite 去重能力能省下很多事后清洗时间。你完全可以在原项目基础上增加一个 MySQL 或 PostgreSQL 的实现只要在 spider_main.py 里换掉 storage 实例即可改动范围不超过十行。4. 常见问题排查四个把爬虫跑挂的坑位与处理记录4.1 请求返回 403请求头里少了一个关键字段现象代码跑起来之后打印的日志里频繁出现 403返回的不是 JSON 而是 HTML 页面。在浏览器里手动刷新页面正常一用代码请求就被拒绝。原因服务端校验了请求来源。用 requests 发请求时默认的 User-Agent 是 python-requests 的标识非常显眼而且我们通常只传了 User-Agent忘了 Referer 字段。服务端看到来源不明、Header 不完整的请求直接拒绝了。解决从 DevTools 复制该请求的 cURL用 curlconverter 转成 Python 代码把里面的 Headers 完整搬到 config.py。重点检查 Referer、Origin、X-Requested-With 这几个字段。之后每次修改 Headers都用浏览器正常访问 - 代码请求各跑一次做对照。4.2 接口不报错但 value 全为 0日期格式暗藏玄机现象请求正常返回 200解析也没有异常但统计出来的 value 全都是 0排名字段也是空的。原因经手这个项目的人里至少三个人踩过同一个坑。配置文件里写的日期是2025-1-5而不是20250105。接口的容错策略是遇到格式不正确的日期不报错而是返回一份全空的数据体。不带前导零的日期串要么被解释成错误的时间点要么直接命中无效时间窗口。解决日期统一用datetime.strftime(%Y%m%d)生成任何来源的日期字符串都先经过 build_date_list 转换一次再做拼接。不要把日期写死在配置文件里这样很难发现少了前导零。4.3 今天的数据拉不到平台数据其实有一天延迟现象任务配置了当天日期跑完之后发现少了今日记录。一开始怀疑是代码逻辑问题调试了半天手动打开页面看到更新中三个字才明白怎么回事。原因迁徙指数不是实时计算的。平台在当天结束后汇总数据次日才会发布完整结果。当天打开页面时看到的往往是部分数据或者提示更新中的占位内容接口返回的 list 为空。解决默认回退一天即date.today() - timedelta(days1)。如果业务上确实需要当天上午的数据把解析逻辑对空 list 的情况做保留标记等次日补拉而不是在空数据基础上直接生成分析报表。4.4 跑到一半请求全部超时触发频率限制了现象任务前几分钟正常后面请求间隔越来越长最后连续出现 timeout 和 500 错误日志刷屏。关掉代码隔一段时间又能跑。原因短时间内的请求频率超过了服务端的阈值服务端对当前会话做了临时限流表现不一定是 403而是超时、返回空列表、随机 500。再往后就是被临时封禁只能等冷却时间。解决在 config.py 里把 request_interval 从 0.1 调整到 0.5 以上采集日期范围长时拆成多个批次运行确保每次跑任务之间留出间隔。一般爬虫类任务低频长跑比高频冲刺更稳。5. 二次开发思路从手动执行到定时任务和联动分析5.1 把采集脚本变成定时任务项目跑到这一步还是手工执行需要手动敲命令才能更新数据。日常使用可以直接把脚本挂到 cron 上每天凌晨拉取前一天的完整数据# 每天 02:30 拉取前一天的迁徙数据 30 2 * * * cd /opt/migration_spider /usr/bin/python3 spider_main.py --start $(date -d yesterday %Y%m%d) --end $(date -d yesterday %Y%m%d) logs/run.log 21cron 的写法需要你按照服务器的实际时间格式调整核心思路是定时执行一次完整的单日拉取。由于 storage 层做了唯一约束重复执行不会产生脏数据可以放心让 cron 每天重复调用。5.2 把命令行参数做成可复用的采集工具如果希望不同部门或同事都能使用把 config.py 里的写死配置改成交互式参数也许会更好用。在 spider_main.py 增加一个 --city 参数允许传入单独城市编码parser.add_argument(--city, defaultNone) city_codes [args.city] if args.city else CONFIG[city_codes]这样既保留了全量采集能力又支持了单城市调试。配合一个简单的 Web 页面或者 Excel 透视表把爬下来的数据按日期 目标城市维度汇总就可以直接画出迁入量趋势折线。5.3 让数据产生更多价值如果从使用者的角度再往前走一步迁徙数据真实有价值的地方在于交叉分析。把这张 CSV 里的 value 列横向叠加到当地交通、酒店商圈数据上可以协助判断节假日客流。把若干天的数据做成热力矩阵能直观看出每个城市互相流动的排名变化。这些玩法都是在现有爬虫产出数据基础上做的核心还是先把稳定可靠的原始数据积累起来。实际改造时我习惯先画一张数据流图确认目标再做代码修改。比如目标是每天凌晨自动拉取写入数据库存到云端那需要改的其实只有 storage 层和加入定时任务两个点其余模块都无需动。说实话项目本身代码量不大但把定位接口 - 构造请求 - 解析 - 落盘 - 定时调度这条链路完整走下来等于把爬虫流程从头到尾过了一遍。从那以后我每次写完爬虫都强制自己走一遍不带 Cookie 的冷启动测试模拟新机器环境把入口跑通再交付这个方法帮我避开好多次数据缺失的现场翻车。希望帮到你。本文还有配套的精品资源点击获取