企业官网建设流程全解析

简介：一份基于 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 请求参数说明：日期、方向、城市编码怎么传

从抓包结果里能直接看到这个接口的查询参数，常见的有以下几类：

这里有两个容易出错的点。第一，参数值必须是行政区划编码而不是汉字城市名，传"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, params=params, timeout=self.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(days=1) return dates def main(): parser = argparse.ArgumentParser() parser.add_argument("--start", required=True) parser.add_argument("--end", required=True) 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(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") main()

这段代码的设计要点在数据流的单向传递：client 拿到字典，parser 把字典转成行记录，storage 负责写入。每个模块都不感知其他模块内部实现。sleep 放在内层循环末尾，是为了控制请求频率，避免短时间大量请求触发频控。

参数说明：--start 和 --end 是必填参数，格式 YYYYMMDD，build_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_ok=True) 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="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) 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(days=1)。如果业务上确实需要当天上午的数据，把解析逻辑对空 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 2>&1

cron 的写法需要你按照服务器的实际时间格式调整，核心思路是"定时执行一次完整的单日拉取"。由于 storage 层做了唯一约束，重复执行不会产生脏数据，可以放心让 cron 每天重复调用。

5.2 把命令行参数做成可复用的采集工具

如果希望不同部门或同事都能使用，把 config.py 里的写死配置改成交互式参数也许会更好用。在 spider_main.py 增加一个 --city 参数，允许传入单独城市编码：

parser.add_argument("--city", default=None) city_codes = [args.city] if args.city else CONFIG["city_codes"]

这样既保留了全量采集能力，又支持了单城市调试。配合一个简单的 Web 页面或者 Excel 透视表，把爬下来的数据按"日期 + 目标城市"维度汇总，就可以直接画出迁入量趋势折线。

5.3 让数据产生更多价值

如果从使用者的角度再往前走一步：迁徙数据真实有价值的地方在于交叉分析。把这张 CSV 里的 value 列横向叠加到当地交通、酒店商圈数据上，可以协助判断节假日客流。把若干天的数据做成热力矩阵，能直观看出每个城市互相流动的排名变化。这些玩法都是在现有爬虫产出数据基础上做的，核心还是先把稳定可靠的原始数据积累起来。

实际改造时，我习惯先画一张"数据流图"确认目标，再做代码修改。比如目标是"每天凌晨自动拉取，写入数据库存到云端"，那需要改的其实只有 storage 层和加入定时任务两个点，其余模块都无需动。

说实话，项目本身代码量不大，但把"定位接口 -> 构造请求 -> 解析 -> 落盘 -> 定时调度"这条链路完整走下来，等于把爬虫流程从头到尾过了一遍。从那以后我每次写完爬虫，都强制自己走一遍不带 Cookie 的冷启动测试，模拟新机器环境把入口跑通再交付，这个方法帮我避开好多次数据缺失的现场翻车。希望帮到你。

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