简介：这份资源是一个用Python爬取百度迁徙数据的脚本压缩包，面向具备Python基础、希望入门网络爬虫与数据解析的开发者，解决按日期获取全国热门城市人口迁徙数据的问题。包内共1个文件，为单个py脚本，压缩包约1KB，轻量易读，可直接运行并修改日期参数。脚本涉及requests发送HTTP请求、json解析返回数据、异常捕获处理以及文件保存等环节，帮助读者理解从接口请求到数据落地的完整流程。已有424人学习，说明该案例在爬虫入门场景中具有一定参考价值。通过阅读与调试这份代码，读者可以掌握按日期抓取迁徙数据的方法，并在此基础上扩展数据可视化或批量采集，适合作为爬虫练手与数据获取的起点。

1. 百度迁徙数据到底能拿到什么：先搞清楚数据边界再动手

很多人第一次搜「python爬取百度迁徙数据」，脑子里想的是把全国人口流动的原始记录整表拉下来。实际打开百度迁徙页面会发现，它对外展示的是指数化的迁徙规模，不是绝对人次。页面按城市对城市给出迁入迁出比例，按日期切换，背后是一套归一化后的相对值。这意味着你能做的是趋势对比、城市热度排序、节前节后流向变化，而不是精确统计「今天有多少人从A市到了B市」。

这个区别决定了整个爬取方案的设计。如果目标是做春运人口流动可视化、区域经济分析、或者给选址模型加一个「人口活跃度」特征，指数数据完全够用。但如果要做精确客流预测，这套数据只能当辅助信号。我一般会先跟需求方确认这一点，避免后面返工。

适合读这篇的人有三类：会一点 Python 基础、想拿真实数据练手爬虫的；做数据分析需要人口流动特征的；以及想搭一个定时采集管道、长期跟踪城市迁徙趋势的。下面从接口结构讲到落地代码，再到长期跑会遇到的问题，按能复现的路径走。

2. 接口结构与请求参数：把迁徙数据的入口摸清楚

2.1 页面数据是怎么加载出来的

百度迁徙的页面本身不直接嵌数据，前端通过异步请求拿 JSON 再渲染。用浏览器开发者工具的 Network 面板，切换到 XHR/Fetch 过滤，刷新页面就能看到返回迁徙数据的请求。常见做法是观察请求 URL 里的几个关键参数：城市代码、日期、迁移方向（迁入/迁出）、时间粒度。

这里有个容易翻车的点：页面上的城市名和接口里的城市代码不是一回事。接口用的是内部编码，你得先拿到一份城市名到代码的映射。这个映射通常也能从页面加载的某个配置请求里拿到，或者从城市选择列表的接口里提取。我一般会先把这份映射抓下来存成本地 JSON，后面所有请求都基于它，避免每次硬编码。

请求里通常还带一个时间戳或签名参数，这类参数往往和当前时间挂钩。如果直接复制浏览器里的 URL 去请求，过一段时间就会失效。所以方案不能是「抄一个 URL 反复用」，而要把参数拼装逻辑还原出来。

2.2 用 requests 拼出第一个可用请求

先装依赖，再写最小请求。下面这段是探路用的，目的是确认参数拼对了、能拿到 JSON。

import requests import json import time # 城市代码映射，实际应从配置接口获取，这里示意结构 CITY_CODE = { "北京": "110000", "上海": "310000", "广州": "440100", } def fetch_migration(city_code, date_str, direction="in"): """ city_code: 城市内部编码 date_str: 形如 20240101 direction: in 表示迁入，out 表示迁出 """ url = "https://huiyan.baidu.com/migration/cityrank.jsonp" params = { "dt": "city", "id": city_code, "type": "move", # move 表示迁徙 "date": date_str, "direction": direction, } headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)", "Referer": "https://huiyan.baidu.com/migration/", } resp = requests.get(url, params=params, headers=headers, timeout=10) resp.raise_for_status() text = resp.text # jsonp 需要剥掉回调函数外壳 start = text.find("(") end = text.rfind(")") data = json.loads(text[start + 1:end]) return data if __name__ == "__main__": result = fetch_migration(CITY_CODE["北京"], "20240101", "in") print(json.dumps(result, ensure_ascii=False)[:500]) time.sleep(1)

逻辑说明：这段代码做了三件事——拼参数、发请求、剥 jsonp 外壳。params里的dt、type、direction是接口区分数据维度的关键，改direction就切换迁入迁出，改date就切换日期。headers里带Referer是因为部分接口会校验来源，不带可能返回空数据。

参数说明：timeout=10防止请求卡死；time.sleep(1)是给后续批量请求留的节流位，单次请求看不出作用，批量时是保命的。jsonp的剥壳逻辑用find和rfind而不是固定切片，是因为回调函数名可能变化，固定位置容易翻车。

提示：接口地址和参数名可能随页面改版变化，上面是结构示意。实际动手时以你在 Network 面板里看到的真实请求为准，把参数名对齐。

2.3 城市代码映射怎么批量拿

单个城市能跑通后，下一步是把城市列表拿全。常见做法是找到城市选择列表对应的接口，它一般返回一个带层级结构的 JSON，包含省、市、区。解析时注意两点：一是直辖市和省的结构可能不同，二是有些城市有别名。我一般会把原始返回存一份，再写一个清洗函数生成{城市名: 代码}的扁平字典，存成city_code.json。

def build_city_map(raw_list): mapping = {} for province in raw_list: for city in province.get("city", []): name = city.get("name") code = city.get("code") if name and code: mapping[name] = code return mapping # 假设 raw 是从城市列表接口拿到的原始结构 # with open("city_code.json", "w", encoding="utf-8") as f: # json.dump(build_city_map(raw), f, ensure_ascii=False, indent=2)

这段的价值在于把「每次手动查代码」变成「一次生成、长期复用」。参数上没什么可调的，重点是清洗时对空值和层级差异做防御，否则后面按城市名取代码时会 KeyError。

3. 批量采集与存储：从单次请求到可复用数据集

3.1 按日期和城市双维度循环

单城市单日期跑通后，真正的数据集需要「多城市 × 多日期」。这里要设计循环顺序和节流策略。我一般外层循环日期、内层循环城市，因为同一日期的请求参数结构一致，便于复用 session。

import requests import json import time import random session = requests.Session() session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)", "Referer": "https://huiyan.baidu.com/migration/", }) def fetch_one(city_code, date_str, direction): url = "https://huiyan.baidu.com/migration/cityrank.jsonp" params = { "dt": "city", "id": city_code, "type": "move", "date": date_str, "direction": direction, } resp = session.get(url, params=params, timeout=10) text = resp.text data = json.loads(text[text.find("(") + 1:text.rfind(")")]) return data def crawl_range(city_map, dates, direction="in"): records = [] for date_str in dates: for name, code in city_map.items(): try: data = fetch_one(code, date_str, direction) records.append({ "date": date_str, "city": name, "code": code, "direction": direction, "payload": data, }) except Exception as e: print(f"fail {name} {date_str}: {e}") # 随机节流，降低被限流的概率 time.sleep(random.uniform(0.8, 1.8)) return records

逻辑说明：用Session复用连接，比每次新建请求快。异常捕获放在内层，单个城市失败不影响整体。random.uniform做随机节流，比固定 sleep 更接近人工节奏。

参数说明：节流区间0.8~1.8秒是我常用的起点，城市数量多时可以适当放大。direction作为参数传入，方便同一套逻辑跑迁入和迁出两遍。

3.2 存成什么格式：JSON 还是 CSV

原始返回是嵌套 JSON，直接存 JSON 最省事，但后续分析不方便。我的做法是两层存储：原始层存 JSON 保底，分析层拍平成 CSV。

拍平的时候，把嵌套里的城市对、比例值展开成列。下面是一个拍平函数示意。

import pandas as pd def flatten(records): rows = [] for rec in records: payload = rec["payload"] # 具体字段名以实际返回为准，这里示意 for item in payload.get("data", {}).get("list", []): rows.append({ "date": rec["date"], "city": rec["city"], "direction": rec["direction"], "target_city": item.get("city_name"), "value": item.get("value"), }) return pd.DataFrame(rows) # df = flatten(records) # df.to_csv("migration_flat.csv", index=False, encoding="utf-8-sig")

encoding="utf-8-sig"是为了 Excel 打开不乱码，这个细节很多人第一次导出 CSV 时会踩。字段名city_name、value是示意，实际以接口返回为准，拍平前先打印一条完整记录看清楚结构。

3.3 断点续采：别让一次中断白跑

批量跑几百个城市乘几十天，中途网络抖动或限流很常见。没有断点机制的话，一次失败就得从头来。我的习惯是每采完一个日期就把该日期的结果落盘，并在一个进度文件里记录已完成日期。

import os def save_progress(done_dates, path="progress.json"): with open(path, "w", encoding="utf-8") as f: json.dump({"done": done_dates}, f) def load_progress(path="progress.json"): if os.path.exists(path): with open(path, encoding="utf-8") as f: return json.load(f).get("done", []) return []

启动时先读进度，跳过已完成日期。这个机制看起来简单，但在长时间采集里是后悔药级别的存在。参数上没什么玄学，关键是落盘时机要够频繁，我一般按日期粒度落，不按城市粒度，平衡 IO 和恢复成本。

4. 避坑与排查：迁徙数据采集里最容易翻车的几件事

4.1 请求返回空数据或结构突变

现象：某天开始，原本正常的请求返回空 list，或者 JSON 结构里字段名变了。

原因：页面改版或接口调整，参数名、返回字段都可能变。这类接口没有对外承诺稳定性。

解决：不要假设结构不变。在拍平函数里对字段做防御性取值，用.get()加默认值；同时保留原始 JSON，结构变了可以重新解析历史数据，不用重采。发现空数据时先手动在浏览器里复现一次请求，对比参数差异。

4.2 高频请求被限流

现象：跑了几十个城市后，后续请求开始超时或返回错误码。

原因：请求频率过高触发服务端限流。

解决：加大随机节流区间，必要时在每采完一个日期后插入更长停顿。我一般会在连续失败达到阈值时自动暂停几分钟再继续。别用固定 sleep，固定节奏反而容易被识别。

4.3 城市代码对不上导致数据错位

现象：采集回来的数据里，城市名和预期不符，或者某些城市一直失败。

原因：城市映射里存在别名、层级差异，或者代码过期。

解决：采集前先做一次映射校验，拿几个已知城市试请求，确认返回的城市名和映射一致。映射文件单独维护，发现对不上就更新，不要在每个脚本里硬编码。

4.4 日期格式和时区问题

现象：某些日期的数据查不到，或者查到的日期和预期差一天。

原因：接口对日期格式敏感，且数据可能有更新延迟。

解决：统一用YYYYMMDD字符串，采集前先确认目标日期数据已更新。做时间序列分析时，把采集日期和数据日期分开存，避免混淆。

4.5 数据指数化带来的误读

现象：拿指数值直接当人次做加总或预测，结论离谱。

原因：指数是归一化后的相对值，不同日期、不同城市的基准可能不同。

解决：只用它做同城市跨时间的趋势对比，或同时间跨城市的排序对比。要做跨维度分析，先确认归一化基准是否一致。这一点在需求沟通阶段就要说清楚，否则后面分析全白做。

5. 让数据真正可用：校验、增量更新与一个长期习惯

采集只是第一步，数据能不能用，取决于你有没有做校验和增量维护。我一般会在拍平后跑几个检查：记录数是否和预期城市数乘日期数吻合、有没有大段缺失日期、同一城市同一日期的迁入迁出是否都齐。缺失的地方标记出来，而不是直接丢进分析。

增量更新是长期跑的关键。我的做法是维护一个「已采集日期」清单，每次运行只采新增日期，历史数据不动。这样每天或每周跑一次，成本很低。下面是一个增量入口的示意。

def incremental_dates(all_dates, done_dates): done = set(done_dates) return [d for d in all_dates if d not in done] # dates = ["20240101", "20240102", "20240103"] # done = load_progress() # todo = incremental_dates(dates, done) # records = crawl_range(city_map, todo) # save_progress(done + todo)

参数上，all_dates可以按需生成，比如最近 30 天。done_dates从进度文件读。这个模式跑顺之后，你手里就有了一份持续更新的城市迁徙数据集，比一次性采集有价值得多。

还有一个我踩过的坑：早期我没做数据版本管理，后来接口字段变了，想回溯旧数据发现原始 JSON 被覆盖了。从那以后我养成的习惯是原始层按日期分文件、只增不改，分析层可以随时从原始层重建。这个习惯让我在后来的多次接口变动里都没慌过。

如果你打算把这个方向做成长期项目，建议先把城市映射和进度机制搭好，再考虑加定时任务。数据本身不难采，难的是稳定地采、可复现地存、清楚地知道它的边界在哪。希望帮到你。

本文还有配套的精品资源，点击获取