迁徙指数爬虫实战：从接口定位到结构化落库全流程解析

简介百度迁徙数据爬取脚本是一个基于Python实现的网络爬虫工具面向爬虫初学者、数据分析人员及人口流动研究者掌握基础Python即可上手解决自动获取百度地图迁徙数据并整理为易分析表格的问题。压缩包共1个文件为Python源码大小仅2KB结构精简适合直接阅读和二次修改。脚本覆盖HTTP请求、BeautifulSoup/lxml解析、数据提取、pandas处理与CSV/Excel导出等完整流程针对动态加载与反爬策略提供了Selenium浏览器自动化、代理IP管理及请求间隔设置等扩展思路同时强调遵守robots协议与数据合规。通过命令行参数指定日期程序即可自动完成数据抓取与表格生成输出指定日期的城市迁入迁出量。资源已有1564人学习下载适合作为爬虫入门实践范例也可为春运等人口迁徙分析提供高效的数据采集基础。1. 从一张迁徙热力榜说起这份资源解决什么某开发者的跨区域人口流动分析报告卡在数据源网页上能看到每个城市迁入、迁出的 top10却没有历史数据下载入口翻来翻去也只是趋势图。我的第一反应是抓包定位接口把迁徙数据按天落成 CSV再继续做分析。这也是本次拆解项目的主线把一套“只展示不下载”的网页指数变成能直接拿来用的结构化数据。做这件事需要一点 Python 基础、会开浏览器开发者工具还要有点对付签名参数的耐心。这篇笔记只讲我在项目里实际验证过的做法接口定位、签名构造、批量任务调度、安全落库以及几个典型踩坑复盘。适合刚接触网络爬虫、想拿真实数据练手的人也适合需要迁流数据做可视化或论文图表的人。文中把目标平台化名为某地图平台接口与参数均已脱敏看懂逻辑后换成自己的目标城市 ID 就能跑。2. 拆解迁徙数据接口从网页到 JSONP 请求2.1 网页观察与接口定位打开迁徙页面F12 切到 Network 面板勾选 XHR然后反复切换迁入、迁出城市。你会发现页面没有整页刷新排行和数值却变了这基本可以断定数据走的是异步接口。接下来按响应体积从大到小排序逐个点开请求预览响应内容搜索move_in、migration、cityrank这类特征词很快就能定位到真正的数据入口。我习惯把这类数据入口的请求完整复制成 cURL 命令存进项目目录的request_example.txt里。后面写 Python 脚本时对照这个原始请求逐字段抄进去比靠记忆重建请求头要可靠得多。尤其是 JS 生成的 Cookie 和签名参数一旦漏掉一个接口要么返回异常码要么返回空数据查起来特别费时间。注意不要直接去 Elements 面板里翻 DOM。页面里展示的“迁徙占比 Top10”往往带百分号和排名提取后还要二次清洗而底层接口返回的才是未格式化的原始数值省去一大堆字符串处理入库和计算都更顺手。提示尽量在浏览器隐身窗口里做抓包避免本地缓存的旧脚本干扰页面逻辑。2.2 请求参数与签名规则从定位到的请求 URL 里挑出每次请求都会变化的参数即可。迁徙指数场景里我整理出来的关键参数如下参数含义示例dt维度类型city 或 provinceid城市或省份编码200001占位type迁徙方向move_in 迁入、move_out 迁出date日期YYYYMMDD20250101callbackJSONP 回调名jQueryxxxsign请求签名由参数排序和盐值计算签名参数是最容易卡住新手的地方。常见实现是把请求参数按 ASCII 码排序拼接成k1v1k2v2形式的字符串再与盐值做哈希把结果拼到 URL 后面。各平台差异在于盐值来源有的是固定字符串有的是时间戳转字符串后截取还有的会混入设备指纹导致每次签名都不一样。我们不需要逆向全部细节只需完成“参数到签名结果”的映射。先在浏览器里连续发两次相同请求对比sign值如果两次完全不同说明签名里掺入了随机数或时间因子。再把sign参数删掉试一次如果数据照常返回说明签名是软校验可以暂时忽略如果立刻报 401 或 403就必须认真实现签名生成器。2.3 用 requests 复现一次最小请求接口入口和参数规则确认后我用 requests 构造最小可运行请求。核心代码大致长这样import hashlib import requests # 站点与盐值示例占位跑通前先抓包确认 API_URL http://xxx.host/migration/cityrank SALT your_salt_placeholder def make_sign(params: dict, salt: str) - str: # 参数按 ASCII 排序后拼接再与盐值做 MD5这是最常见的签名方式 raw .join(f{k}{v} for k, v in sorted(params.items())) raw salt return hashlib.md5(raw.encode(utf-8)).hexdigest() def fetch_city_rank(date: str, city_id: str, move_type: str move_in): params { dt: city, id: city_id, type: move_type, date: date, callback: jQuery_test, } # 多数接口里签名参数名是 sign这里按最普遍的命名演示 params[sign] make_sign(params, SALT) resp requests.get(API_URL, paramsparams, timeout10) resp.raise_for_status() return resp.text # 此时拿到的是 JSONP 字符串第3章再做剥壳处理这段代码有两个关键点。第一make_sign内部用sorted(params.items())统一参数顺序因为签名规则通常规定“先排序后拼接”顺序错一位后端校验就过不了。第二callback参数不能随便起名。有的后端会校验回调名是否在白名单内更稳妥的做法是先用浏览器里实际出现的回调名做冒烟测试确认通了再考虑改写成别的形式。如果单挑一个城市、单日期的请求能正常返回城市排名和指数值说明最小链路已经打通。此时先不要铺开全量任务批量层的事放到第 3 章排错成本会更低。3. 爬虫工程化批量城市、日期轮询与结构化落库3.1 批量任务调度单请求跑通后把循环套上去。迁徙数据的任务可拆成三层日期列表 × 城市列表 × 迁徙方向也就是迁入、迁出两组。这里必须提前估算总请求量30 个城市跑一整年仅迁入就是 10950 个任务再乘迁出直接破两万。设计调度器时要把“已完成子任务”和“总任务”分开维护否则中断重跑会浪费大量重复请求。下面是任务生成器的简化写法from datetime import date, timedelta from typing import Iterator, Dict def gen_tasks(city_ids, start_date: str, end_date: str) - Iterator[Dict[str, str]]: 生成全部任务参数返回迭代器避免一次性占用过多内存 d date.fromisoformat(start_date) stop date.fromisoformat(end_date) while d stop: for cid in city_ids: for move_type in (move_in, move_out): yield { date: d.strftime(%Y%m%d), id: cid, type: move_type, direction: move_type, } d timedelta(days1)gen_tasks是生成器而不是列表这是长周期爬取里很值得坚持的习惯。把两万多个任务一次性 append 进 list内存占用和启动耗时都会变难看生成器每个循环才产生一个任务字典跑起来轻得多也方便中途停止。外层按天推进内层先遍历城市再遍历方向这样后面按日期聚合 CSV 时顺序是连续的。3.2 JSONP 响应标准化解析接口返回的 JSONP 是一段字符串不是可以直接使用的 dict。剥壳办法有两种正则去包裹或者直接做字符串裁剪。正则更通用因为回调名可能带数字、下划线和随机后缀手写前缀裁剪很容易写死。import re import json def parse_jsonp(resp_text: str) - dict: # 匹配最外层的函数调用包裹取出其中的 JSON 文本 m re.search(r^[^(]\((.*)\)\s*;?\s*$, resp_text, re.S) if not m: raise ValueError(非 JSONP 响应先检查请求是否被拦截或签名失效) return json.loads(m.group(1))这里的正则写法比较宽^[^(]\(匹配到第一个左括号(.*)\)贪婪匹配最后一个右括号适配各种回调名。弱点也明显如果响应正文里夹带了大段 HTML 调试信息贪婪匹配可能把多出来的内容也吞进 JSON 里。所以我还会做一层防御先json.loads失败后立刻把原始字符串前 200 个字符打印出来定位是哪段内容混进来了。拿到标准 JSON 后迁徙数据的核心字段通常藏在data.list及类似路径下每条记录是来源城市、目标城市、排名、指数值。不同时间段的响应结构可能微调所以不要凭经验硬写字段路径先json.dumps(data, ensure_asciiFalse, indent2)把结构打出来看一遍再写提取逻辑。3.3 结构化落库CSV 优先数据最终要喂给分析工具我第一版直接存 CSV方便用 pandas 读。字段结构相对扁平用标准库csv逐行写入足够没必要引入数据库。import csv from pathlib import Path FIELD_NAMES [ date, city_id, direction, rank, target_id, target_name, value, ] def save_row(row: dict, output_path: Path): is_new not output_path.exists() with open(output_path, a, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnamesFIELD_NAMES) if is_new: writer.writeheader() writer.writerow({k: row.get(k, ) for k in FIELD_NAMES})这里用utf-8-sig而不是utf-8是踩过坑之后改的。因为办公表格软件直接双击打开 UTF-8 编码的 CSV中文列名会变成乱码带 BOM 头的utf-8-sig兼容性更好也不影响 pandas 读取。rank字段保存目标城市在榜单里的排名后面做“哪些城市长期榜上有名”的统计时直接用省一次排序。如果是几十 GB 的超大量数据CSV 读写会明显变慢但迁徙数据这种单日单城市最多十几条排名的场景CSV 完全够用落库前先做日期分区文件名里带起止日期后面找数据也方便。4. 反爬应对与请求策略频率、Cookie、断点续爬4.1 先把反爬规则摸清地图类平台的反爬通常不是单点设置而是三件事一起起作用请求头校验、频率限制、签名时效。请求头是最容易过的一关只要带上常见浏览器的 UA 和正确的 Referer一般不会被拦。频率限制靠阈值短时间请求过密就会临时封禁。签名时效要求请求发出的那一刻签名有效把几分钟前算好的签名拿来复用大概率会直接失败。我每次写爬虫前会先过一遍体检清单请求头是否包含完整且一致的 UA、Referer、Acceptcookie 或 token 是否过期是否需要定期刷新签名参数与请求参数是否严格对齐单台机器单日请求量是否控制在平台可接受区间。不要迷信“挂代理换 IP 就万事大吉”。有些平台的封禁粒度是账号和设备指纹换 IP 只能拖时间请求频次一旦持续超限照样会被风控规则锤掉而且表现形式很隐蔽HTTP 200 正常返回但data为空。这种静默失败最容易让人在代码里空转排查半天。4.2 统一会话与请求封装我习惯把网络请求收敛进一个带requests.Session的封装函数UA、Referer、Cookie 放在 session 级别不散落在每个请求里。这样改策略只动一个地方可维护性高很多。import time import requests from fake_useragent import UserAgent session requests.Session() session.headers.update({ User-Agent: UserAgent().random, Referer: http://xxx.host/, Accept: application/json, text/javascript, */*; q0.01, }) def safe_get(url: str, params: dict, retries: int 3): for attempt in range(1, retries 1): try: resp session.get(url, paramsparams, timeout10) if resp.status_code 200: return resp.text except requests.RequestException as e: print(f[{attempt}/{retries}] 请求异常: {e}) time.sleep(2 ** attempt) # 指数退避1秒、2秒、4秒 return Nonefake_useragent每次生成随机 UA避免多个请求重复同一个浏览器标识。但这里有个反直觉的点UA 切换太频繁反而容易触发风控。某平台的日志里如果 UA 一直在 Chrome 和 Safari 之间跳跃基本可以断定是爬虫。我个人习惯是先用固定 UA 把 100 条数据刷通确认无异常后才允许随机化。safe_get里的time.sleep(2 ** attempt)是指数退避第一次失败等 2 秒第二次等 4 秒第三次等 8 秒。它解决的不只是限流还兼顾了临时网络抖动。注意重试只对超时和连接错误有意义如果返回 403 或校验失败重试多少次都没用这时候要回头检查签名和 Cookie。4.3 频率控制与断点续爬频率控制最朴素的方案就是 sleep。但 sleep 只解决“爬太快”解决不了“中断后重跑”。两万多条任务跑到一半断电或代码异常从头再来一次谁都受不了。我的项目里加了一个极简进度文件记录每个已完成的“日期_城市_方向”启动时先读这个集合跳过已完成任务。from pathlib import Path PROGRESS_FILE Path(./progress.txt) def load_done() - set: if not PROGRESS_FILE.exists(): return set() lines PROGRESS_FILE.read_text(encodingutf-8).strip().splitlines() return set(lines) def mark_done(task_key: str): with open(PROGRESS_FILE, a, encodingutf-8) as f: f.write(task_key \n)load_done把进度文件读成集合爬虫启动时遍历任务key 在集合里就跳过。这里藏着一个顺序陷阱进度标记必须在“成功保存数据之后”写入。如果先mark_done再save_row一旦保存失败这个任务就被永久跳过了数据缺失且无法感知。正确顺序是先保存后标记。频率控制方面单线程跑两万多个请求每次间隔 0.5 秒大约也要三小时。把间隔压到 0.1 秒容易封号建议保持 0.3 到 0.8 秒的随机间隔。另外对同一个城市的连续请求间隔要再放大一倍因为短时间集中访问同一资源的风控敏感度更高。提示长时间爬取前先用一个城市、三天的数据量做冒烟测试确认签名、解析、落库全部正常后再放开全量。5. 避坑指南五个高频踩坑点5.1 响应层JSONP 回调名与空数据坎一JSONP 回调名被后端校验现象第一批抓 20 个城市返回的响应体里一半直接报“非法回调”另一半正常。原因我把回调名自定义成了callbackmydata而后端白名单只认前端固定使用的回调形式不匹配的请求会在入口处被拒绝。解决从浏览器开发者工具里复制实际回调名原样使用不要为了好看自创名字。如果后端回调名带随机后缀就固定抓当前网页正在用的那个值再在解析层做通用匹配。坎二返回 200 但指数全是 0现象请求成功JSON 里data.list字段也在但所有value值都是 0。原因传的id是省级编码而dtcity要求城市级编码前后端数据结构不匹配时平台不报错只返回占位空数据导致接口状态看起来正常实际上没有拿到任何有效值。解决先拿到平台自带的行政区划对照表把目标城市编码核对一遍再批量跑。别手打城市名部分新一线城市的编码规则和普通地级市不一样同样一个名字在不同维度下可能对应两个 ID。5.2 请求层签名失效、频率封禁与日期偏移坎三签名盐值随版本更新失效现象前一天跑得正常的脚本第二天所有请求都开始返回 403。原因接口升级了鉴权签名算法里新增了参与计算的字段或者把盐值从一个固定字符串改成了“字符串设备号”的拼接老签名自然全部失效。解决重新抓取一次正常请求用本地脚本逐段拼接签名与返回的签名做比对找出差异字段。把盐值和参与签名字段的列表抽成配置常量放到项目配置文件里而不是埋在业务函数中间下次平台更新时改配置就行。坎四请求过于频繁导致 IP 临时受限现象连续跑 20 分钟后所有响应开始返回 429 状态码或被重定向到验证页面。原因请求频次超过接口阈值触发风控规则最直接表现就是 IP 被短时限制。解决调大 sleep 间隔同时去掉不必要的重复请求。被封后不要立刻重试猛刷先停 10 分钟再继续如果项目周期允许把采集窗口拉长到 3 到 7 天每天固定跑一小批比一次性快速拉完要安全得多。坎五日期参数差一天现象爬下来的数据总是比网页展示的少一天且每天的数据日期范围和预期对不上。原因平台按当天零点切分日期而我本地生成日期时用了错误的初始化起点或者时区换算没做把开始日期写成了“昨天”。解决统一在参数生成处使用strftime(%Y%m%d)并在运行时打印一条请求日志记录实际请求的日期字段与网页日历人工对照一遍。更稳妥的做法是解析响应后做“最大日期”检查发现日期范围和预期不一致就立即告警别让整个批次在错误参数下跑完。6. 进阶让爬下来的数据“能直接用”6.1 城市编码映射与缺失值补全数据落库后我发现一个明显不便CSV 里的目标城市是编码不是中文名做图表和人工核对都要来回翻译。于是我把行政区划整理成一个映射字典入库时补充一列中文名。但字典总有滞后性如果平台新增了区划或者调整了名称本地字典就会漏。解决办法是优先使用接口返回的target_name字段它和编码来自同一份数据源永远是最新的。6.2 校验脚本先验证再入库爬取完并不等于数据可用我额外写了一个不带网络请求的校验步骤def validate_day(rows, expected_count: int) - bool: # 检查单日数据条数是否等于 城市数 × 方向数 if len(rows) ! expected_count: return False # 简单抽查排名和指数值不能为空或全 0 if any(r.get(rank) in (None, ) or r.get(value) 0 for r in rows[:20]): return False return True这个脚本踩住了两个最常见的静默错误记录数与任务数不一致以及指数全为 0。每条数据入库前先跑一遍校验不合格就直接标记该任务为失败后续单独补拉而不是混在正常数据里继续往下跑。以前我总把爬虫写好就撒手数据入库后才发现错了一大片回头重新补爬反而耗时更久。后来我强制自己形成一套启动自检习惯先加载已完成集合再单独跑一次签名生成函数最后对前三条响应做字段打点确认返回结构和预期一致才放开全量任务。这套检查也就 10 分钟但能挡住绝大多数返工。希望这篇笔记能帮你在做迁徙数据爬取或类似指数型接口时少走几步弯路。本文还有配套的精品资源点击获取