简介这份资源是一个用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, directionin): 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, paramsparams, headersheaders, timeout10) 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_asciiFalse)[:500]) time.sleep(1)逻辑说明这段代码做了三件事——拼参数、发请求、剥 jsonp 外壳。params里的dt、type、direction是接口区分数据维度的关键改direction就切换迁入迁出改date就切换日期。headers里带Referer是因为部分接口会校验来源不带可能返回空数据。参数说明timeout10防止请求卡死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, encodingutf-8) as f: # json.dump(build_city_map(raw), f, ensure_asciiFalse, indent2)这段的价值在于把「每次手动查代码」变成「一次生成、长期复用」。参数上没什么可调的重点是清洗时对空值和层级差异做防御否则后面按城市名取代码时会 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, paramsparams, timeout10) text resp.text data json.loads(text[text.find(() 1:text.rfind())]) return data def crawl_range(city_map, dates, directionin): 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(ffail {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。存储层格式用途注意点原始层JSON留档、重解析按日期分文件避免单文件过大分析层CSV直接进 pandas字段拍平一行一条记录映射层JSON城市代码复用单独维护不混在数据里拍平的时候把嵌套里的城市对、比例值展开成列。下面是一个拍平函数示意。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, indexFalse, encodingutf-8-sig)encodingutf-8-sig是为了 Excel 打开不乱码这个细节很多人第一次导出 CSV 时会踩。字段名city_name、value是示意实际以接口返回为准拍平前先打印一条完整记录看清楚结构。3.3 断点续采别让一次中断白跑批量跑几百个城市乘几十天中途网络抖动或限流很常见。没有断点机制的话一次失败就得从头来。我的习惯是每采完一个日期就把该日期的结果落盘并在一个进度文件里记录已完成日期。import os def save_progress(done_dates, pathprogress.json): with open(path, w, encodingutf-8) as f: json.dump({done: done_dates}, f) def load_progress(pathprogress.json): if os.path.exists(path): with open(path, encodingutf-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 被覆盖了。从那以后我养成的习惯是原始层按日期分文件、只增不改分析层可以随时从原始层重建。这个习惯让我在后来的多次接口变动里都没慌过。如果你打算把这个方向做成长期项目建议先把城市映射和进度机制搭好再考虑加定时任务。数据本身不难采难的是稳定地采、可复现地存、清楚地知道它的边界在哪。希望帮到你。本文还有配套的精品资源点击获取