简介这份资源面向具备Python基础、希望上手网络爬虫与数据处理的开发者提供一份可直接运行的百度迁徙数据抓取脚本解决按日期获取全国热门城市人口迁徙信息的需求。压缩包内共1个文件为单个py脚本整体仅1KB轻量易读便于快速理解与二次修改。脚本涉及requests发送HTTP请求、json解析返回数据、异常捕获处理以及文件保存等环节覆盖爬虫请求、数据解析与本地存储的完整链路适合作为爬虫入门与数据采集练习的参考。目前已有425人学习下载说明其在同类练手项目中具有一定参考价值。读者可从中了解如何按日期参数调用接口、提取来源地、目的地与迁徙规模等字段并在此基础上扩展可视化或批量采集逻辑是学习Python网络数据获取的实用小工具。1. 百度迁徙数据到底能拿到什么从一次失败的抓取说起去年做区域人流分析时我第一反应就是去抓百度迁徙的数据。结果第一次跑脚本返回的是一堆加密字符串连个像样的 JSON 都没有。后来才搞明白百度迁徙的公开页面走的是前端渲染加接口签名直接 requests.get 拿到的 HTML 里根本没有数据。这个标题“python爬取百度迁徙数据.zip”背后其实是一套完整的采集链路定位接口、还原参数、处理反爬、落库存储。它适合做城市间人流趋势分析、节假日出行预测、区域经济活跃度观察的从业者。如果你只是想拿一份静态 CSV那网上流传的旧数据可能更省事但如果你需要按天、按城市对、按迁徙方向持续更新自己写采集脚本是绕不开的。这一章先把数据长什么样、从哪来、值不值得自己抓讲清楚后面再动手。百度迁徙的数据粒度通常是“某天、某城市、迁入/迁出、来源/目的地城市、占比”。它不提供绝对人数只有比例。这意味着你不能直接拿它算“多少人从 A 到了 B”但可以做相对排序和趋势对比。常见做法是每天定时抓取按城市对存成宽表再和统计年鉴、手机信令数据做交叉验证。我一般会先确认三件事目标城市列表、时间范围、是否需要历史回补。这三件事决定了你是写一个一次性脚本还是搭一个小型调度系统。如果只是做一次课题分析脚本加 CSV 就够了如果要持续跑几个月就得考虑去重、断点续抓和存储格式。2. 接口定位与请求参数还原把黑匣子拆开看2.1 从浏览器开发者工具里找到真实请求百度迁徙的页面不会把数据直接写在 HTML 里而是通过 XHR 请求异步加载。打开浏览器开发者工具切到 Network 面板筛选 XHR 或 Fetch然后刷新页面。你会看到类似getMigrationData或migration的请求。重点看三个东西Request URL、Request Headers 里的 Referer 和 User-Agent、以及 Response 返回的结构。常见做法是右键复制为 cURL然后转成 Python 的 requests 调用。注意不要直接复制 Cookie 里的敏感字段很多公开接口只需要 Referer 和 UA 就能过。import requests url https://example-migration-api.com/api/getMigrationData headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Referer: https://example-migration-page.com/, Accept: application/json, text/plain, */* } params { city: 北京, date: 2024-01-01, type: move_in } resp requests.get(url, headersheaders, paramsparams, timeout10) print(resp.status_code) print(resp.text[:500])这段代码的核心是模拟浏览器发起的 XHR 请求。headers里的 Referer 必须和页面域名一致否则接口可能返回 403。params里的city、date、type是常见参数但不同时期字段名可能变化需要以实际抓包为准。timeout建议设 10 秒以内避免卡死。如果返回的是加密字符串说明接口做了响应加密需要进一步分析前端 JS 里的解密逻辑。2.2 参数签名与时间戳的处理很多接口会带一个sign或token参数它通常由前端 JS 用时间戳、城市名和盐值拼接后做 MD5 或 SHA1 生成。你可以在开发者工具的 Sources 面板里搜索sign或md5找到生成函数。常见做法是用 Python 复现这个函数而不是去执行整个 JS 文件。如果 JS 被混淆了可以用 Node.js 跑一个简化版或者用 PyExecJS 调用。但要注意签名算法可能会变所以脚本里最好把签名逻辑单独封装方便后续替换。import hashlib import time def make_sign(city, date, saltabcdef123456): raw f{city}|{date}|{salt} return hashlib.md5(raw.encode(utf-8)).hexdigest() city 上海 date 2024-01-01 sign make_sign(city, date) print(sign)这里salt是示例值实际需要从 JS 里提取。make_sign把城市、日期和盐值用竖线拼接后做 MD5。参数说明city和date必须和请求参数完全一致包括大小写和空格。如果接口用的是时间戳还要把time.time()转成毫秒或秒具体看 JS 里的写法。我一般会先手动构造几个请求确认签名正确后再批量跑。2.3 返回数据的解析与字段映射拿到 JSON 后不要急着往数据库塞。先看结构通常有data数组每个元素包含city_name、value、rank等字段。有些接口返回的是嵌套对象比如data.list。常见做法是写一个parse_response函数把嵌套结构拍平成一行记录。字段映射要保留原始字段名和中文注释方便后续核对。如果返回的是加密字符串先解密再解析。解密逻辑一般在 JS 里可以用crypto-js的对应 Python 库比如pycryptodome。import json def parse_response(text): obj json.loads(text) rows [] for item in obj.get(data, []): rows.append({ city: item.get(city_name), value: item.get(value), rank: item.get(rank), date: item.get(date) }) return rows # 假设 resp.text 是合法 JSON # rows parse_response(resp.text) # print(rows[:3])parse_response把data数组里的每个元素转成字典。get方法避免 KeyError。如果字段名是中文比如城市直接写item.get(城市)即可。注意有些接口的value是字符串需要转成 float 再存。解析完后建议先打印前三条确认字段和值都正常。3. 批量抓取与存储从单次请求到可持续跑3.1 城市列表与日期循环的编排单次请求只能拿一个城市一天的数据。要批量抓就得先准备城市列表和日期范围。城市列表可以从公开的行政区划表里取日期用datetime生成。常见做法是双层循环外层日期内层城市。但要注意请求频率别把接口打挂。我一般会加time.sleep(1)到time.sleep(3)的随机延迟模拟人工操作。如果城市多可以按天分文件存避免内存爆掉。import time import random from datetime import datetime, timedelta cities [北京, 上海, 广州, 深圳] start datetime(2024, 1, 1) end datetime(2024, 1, 3) current start while current end: date_str current.strftime(%Y-%m-%d) for city in cities: # 这里调用前面的请求函数 # resp fetch(city, date_str) # rows parse_response(resp.text) # save(rows) time.sleep(random.uniform(1, 3)) current timedelta(days1)这段代码用while循环遍历日期内层遍历城市。random.uniform(1, 3)生成 1 到 3 秒的随机延迟降低被封风险。save函数需要自己实现可以写 CSV 或插入 SQLite。注意日期格式要和接口要求一致有些接口用20240101有些用2024-01-01以抓包为准。3.2 用 SQLite 做本地去重存储CSV 适合一次性分析但批量抓取时容易重复。我一般用 SQLite 建一张表把city、date、type设成联合唯一索引插入时用INSERT OR IGNORE。这样即使脚本中断重跑也不会产生重复数据。表结构可以设计成id、city、date、direction、target_city、value、rank。direction区分迁入迁出target_city是来源或目的地。import sqlite3 conn sqlite3.connect(migration.db) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS migration ( id INTEGER PRIMARY KEY AUTOINCREMENT, city TEXT, date TEXT, direction TEXT, target_city TEXT, value REAL, rank INTEGER, UNIQUE(city, date, direction, target_city) ) ) def save_rows(rows): for row in rows: cursor.execute( INSERT OR IGNORE INTO migration (city, date, direction, target_city, value, rank) VALUES (?, ?, ?, ?, ?, ?) , (row[city], row[date], row[direction], row[target_city], row[value], row[rank])) conn.commit()CREATE TABLE里的UNIQUE约束保证同一城市、日期、方向、目标城市只存一条。INSERT OR IGNORE在冲突时跳过。save_rows接收解析后的列表逐条插入。注意conn.commit()要放在循环外避免频繁写盘。如果数据量大可以每 100 条提交一次。3.3 断点续抓与日志记录批量抓取最怕跑到一半断了不知道哪些城市哪些日期已经抓过。常见做法是维护一个progress表记录city、date、status。每次开始前先查statusdone的记录跳过已完成的。日志用logging模块写到文件记录请求时间、状态码、异常信息。这样出问题时能快速定位是哪个城市哪个日期翻车了。import logging logging.basicConfig(filenamecrawl.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) def is_done(city, date): cursor.execute(SELECT 1 FROM progress WHERE city? AND date? AND statusdone, (city, date)) return cursor.fetchone() is not None def mark_done(city, date): cursor.execute(INSERT OR REPLACE INTO progress (city, date, status) VALUES (?, ?, done), (city, date)) conn.commit()is_done查进度表返回 True 表示已抓过。mark_done在成功后写入。logging把信息写到crawl.log方便回溯。注意INSERT OR REPLACE会覆盖旧记录适合更新状态。如果抓取失败可以记statusfailed后续重试。4. 避坑与常见问题那些让我重跑整晚的坑4.1 返回 403 或空数据Referer 和 UA 没带对现象请求返回 403或者返回 200 但data为空。原因百度迁徙接口会校验 Referer 和 User-Agent缺失或伪造得不像浏览器就会拒绝。解决把 Referer 设成页面完整 URLUA 用常见 Chrome 的字符串。如果还不行检查是否有 Cookie 里的BAIDUID有些接口需要它。我一般会先用浏览器正常访问一次然后把请求头完整复制到脚本里再逐步删减找到最小必要集合。4.2 签名错误时间戳精度或盐值不对现象接口返回sign error或invalid token。原因签名算法里时间戳是毫秒你用了秒或者盐值是从 JS 里提取的但混淆后看错了。解决在 JS 里搜索sign或md5把生成函数完整抄下来用 Node.js 跑一遍对比结果。如果 JS 被压缩可以用jsbeautifier格式化。注意有些接口的签名还包含请求体GET 和 POST 的签名方式可能不同。4.3 频率过高被封 IP延迟和并发没控制好现象跑了几百个请求后接口开始返回 429 或直接超时。原因请求间隔太短或者用了多线程并发。解决加随机延迟单线程跑。如果必须并发用threading.Semaphore限制并发数在 2 到 3 个并且每个线程独立延迟。我一般会设time.sleep(random.uniform(2, 5))跑一晚上也就几千条足够课题用。如果被封了换 IP 不是长久之计降低频率才是。4.4 数据对不上比例和绝对值的混淆现象抓下来的value加起来不等于 100或者和页面显示不一致。原因百度迁徙的value是占比但可能只返回了前 N 个城市剩下的被截断。解决在解析时保留rank字段只取前 10 或前 20 做分析。如果需要完整分布得看接口是否支持分页。另外迁入和迁出的value是分开算的不要混在一起求和。我一般会在入库后跑一个校验 SQL看每个城市每天的value总和是否在合理范围。4.5 脚本跑久了内存暴涨没及时释放响应对象现象跑了几千个请求后Python 进程占用内存越来越大。原因requests的响应对象没有显式关闭或者把全部数据堆在列表里。解决用with requests.get(...) as resp或者手动resp.close()。数据不要全存内存解析完立刻写库。如果用的是json.loads大响应可以改用ijson流式解析。我一般每处理 100 条就gc.collect()一次虽然有点土但管用。5. 进阶技巧用增量更新和可视化验证数据质量数据抓下来只是第一步怎么确认它靠谱、怎么持续更新才是长期跑的关键。我一般会做两件事增量更新和可视化抽查。增量更新是指每天只抓昨天和今天的数据历史数据不动。这样脚本跑起来很快也不容易触发反爬。具体做法是在progress表里记录最后抓取日期每次启动时从那天开始往后跑。如果发现某天数据缺失再手动补。可视化抽查是用pandas和matplotlib画一张迁徙趋势图看曲线是否平滑、有没有突刺。突刺往往是接口返回了异常值比如value为 0 或 100。这时候要回去看原始响应确认是接口问题还是解析问题。我一般会按城市对画热力图横轴日期纵轴城市颜色深浅表示迁徙强度。如果某一行全白说明那个城市没抓到数据得检查城市名是否写错。import pandas as pd import matplotlib.pyplot as plt conn sqlite3.connect(migration.db) df pd.read_sql(SELECT date, city, value FROM migration WHERE directionmove_in, conn) pivot df.pivot_table(indexcity, columnsdate, valuesvalue, aggfuncsum) plt.figure(figsize(12, 6)) plt.imshow(pivot, aspectauto, cmapYlOrRd) plt.colorbar(labelmigration index) plt.yticks(range(len(pivot.index)), pivot.index) plt.xticks(range(len(pivot.columns)), pivot.columns, rotation45) plt.tight_layout() plt.savefig(migration_heatmap.png)这段代码从 SQLite 读数据用pivot_table转成城市×日期的矩阵然后用imshow画热力图。aggfuncsum是因为同一天同一城市可能有多个目标城市求和得到总迁入指数。cmapYlOrRd是黄到红的渐变颜色越深表示迁入越强。保存成 PNG 后肉眼扫一遍就能发现异常。如果某个城市整列都是白色回去查progress表看是不是漏抓了。还有一个技巧是用schedule库做定时任务每天凌晨跑一次增量。但要注意如果脚本本身不稳定定时任务只会让错误累积。我一般会先手动跑一周确认没问题再上定时。另外SQLite 在并发写入时容易锁库如果定时任务和手动查询同时进行最好加一个timeout参数或者改用 PostgreSQL。但课题级别用 SQLite 足够了别过度设计。最后说一个血泪教训不要相信接口返回的date字段它可能是服务器时间和你请求的日期差一天。我一般会用请求参数里的日期作为主键而不是响应里的日期。这样即使接口时区有问题数据也能对上。希望帮到你。本文还有配套的精品资源点击获取