☰
微博POI数据爬虫:从接口解析到热力图绘制的实战指南
2026/10/2 1:42:49 网站建设 项目流程

简介:面向需要采集微博地理位置信息的研究者与开发者,该源码是一套基于Python的微博POI数据爬虫设计方案,适用于舆情监控、区域分析、市场调研等场景。项目共包含22个文件,压缩包约4.23MB,以4个Python源文件为功能主体,分别负责搜索抓取、代理配置、数据清洗与登录认证;同时辅以XML配置文件、CSV数据结果及PNG可视化图表,兼顾运行参数设定与输出整理,整体结构清晰。已有380人学习,可作为相关课程设计或科研项目的参考实现。阅读源码可理解模块化爬虫的骨架搭建、反爬与代理处理思路,以及POI数据的清洗、存储与可视化流程,为后续二次开发或功能扩展提供直接借鉴。

1. 微博POI数据爬虫:把带位置的微博变成可分析的空间数据

接到“把微博上带位置的发言按商圈做热力图”这类需求时,你很快会发现一个尴尬:常规搜索接口能拿到微博文本,却拿不到坐标。真正承载位置信息的是POI——一个POI_ID 对应一个地名、一组经纬度、一个分类和一份签到量。基于 Python 的微博POI数据爬虫设计源码,就是把“微博签到数据”从文本流里拆出来,落成能进数据库、能画图、能建模的数据集。本文按一次真实落地路径拆解:先讲接口形态和参数,再用 requests 把链路跑通,用 SQLAlchemy 入库,最后把踩过的坑和热力可视化一并交代。适合做舆情、城市计算、门店选址的从业者,也适合拿 Python 练爬虫落地的新手。

2. 把微博 POI 接口拆开看:数据形态与三个必传参数

2.1 POI 在微博里到底以什么形态存在

微博的 POI(Point of Interest,兴趣点)不是一个独立的数据库表,它附着在两处:一是“带位置的微博”里带的 location 字段,二是“地点详情页”里的完整地点信息。前者是轻量引用,通常只给一个 poi_id 和地点名;后者才包含省市区、分类、经纬度、签到人数、照片数这些用于分析的字段。

所以爬虫设计的第一个决策是:你要的是“微博+地理标签”的关联数据,还是“地点维度”的聚合数据。前者适合研究“某商圈用户讨论了什么”,后者适合做“城市POI覆盖度对比”。多数项目两者都要,即先按关键词抓到微博,再从微博里的 poi_id 去补地点详情。

我一般会先抓一个返回包,把字段结构存成 JSON 文件,再写解析。这个动作看起来多余,但它决定后面所有代码的字段映射。同一套接口在不同登录态下返回的卡片类型会变,见过太多人因为少了一个 card_type 分支,解析到一半直接崩。

2.2 请求入口:移动端接口与 containerid 的拼法

微博网页端和 App 端走不同的接口,爬 POI 数据从业界共识是用移动端接口,因为返回是干净的 JSON,没有服务端渲染的 HTML 噪音,而且 location 字段保留完整。常见入口是 m.weibo.cn 下的 container/getIndex,核心参数是 containerid。

containerid 有两类拼法:

用途拼接方式说明
地点详情230283 前缀 + poi_id拿到某个 POI 的完整档案
关键词搜索100103type=1&q=关键词按文本关键词找微博,再从中筛带位置的数据

URL 编码是必须做的。中文关键词做 query 参数时,requests 的 params 会自动编码,但如果有人手工拼 URL,漏掉编码就会看到一堆 %E5%8C%97%E4%BA%AC 这样的内容。

注意一个细节:你抓包看到的 containerid 后缀可能与我上面写的有差异,因为账号权限和接口版本会微调。正确的做法不是背参数,而是先用浏览器登录微博,打开抓包工具,点开一条带位置的微博,看真实请求里的容器ID格式,再把它填进配置。

2.3 翻页三件套:pagebar、since_id 与 max_id 的分工

这是新手最容易卡住的地方。移动端列表不是后端分页,而是“加载更多”的游标翻页。接口里常见两类参数:

  • pagebar:0 表示首页,1 表示加载更多,部分版本用 0/1 切换来控制翻页位置。
  • since_id:服务端返回的游标字符串,下一次请求原样带回,就能从上次的位置继续拉。

老版本接口偶尔要求 max_id,是另一个名字的游标。我习惯在这三个参数上用配置驱动,而不是写死在代码里。每次解析返回包时,优先从 cardlistInfo 或 data 顶层取游标字段,取到才发起下一次请求。

有一个常见误区:把 page 从 1 递增,以为这样能翻到第 100 页。微博移动端不吃这一套,固定 page 翻几页后就会返回重复数据。

2.4 Cookie 生命周期:登录态决定你能看到多少字段

不登录也能请求接口,但返回的卡片里 location 字段会很稀薄,甚至直接空掉。原因是地点信息被判定为个性化内容,需要登录态支撑。

Cookie 里真正起作用的是 SUB、SUBP 和会话类字段(不同账号看到的字段名会有差异)。它们不是永久的,可能几小时也可能几天后失效,失效后接口不会报错,而是返回一个需要验证码的页面或直接 302 跳到登录页。

因为这一点,爬虫设计里必须把 Cookie 和业务代码解耦。我不会把 Cookie 写死在代码里,而是放在单独的 config.py 里,并配套一个简单的检查函数:拿到响应后先判断返回体里是否包含 passport 这类登录跳转特征,一旦出现就立即暂停任务而不是继续空跑。

2.5 判断返回包结构的两个前置动作

在写正式代码前,先用一个最小请求确认三件事:返回体是不是 JSON、cards 数组里有没有带 location 的条目、游标字段叫什么名字。

用 curl 看一眼即可,不需要完整代码:

curl -s 'https://m.weibo.cn/api/container/getIndex?containerid=100103type%3D1%26q%3D%E5%95%86%E5%9C%88' \ -H 'User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 13_0 like Mac OS X)' \ -H 'Referer: https://m.weibo.cn/' \ -b 'SUB=你的Cookie值' | head -c 3000

这个命令里,containerid 中的 q 参数是 URL 编码后的“商圈”两个关键字,-b 把 Cookie 直接带过去。返回内容开头如果是 {"ok":1,"data":...},说明请求链路通了;如果看到 HTML 标签,说明被拦截,需要检查 Cookie 和 User-Agent。

参数说明:User-Agent 用 iPhone 的 UA 是移动端接口的常见做法,Referer 必须指向 m.weibo.cn,缺少 Referer 时部分请求会返回空数据。head -c 3000 只截前 3000 字符,避免终端被一坨 JSON 刷屏。

3. 用 requests 把抓取链路跑通:Session、延迟策略与第一个返回包

3.1 模块怎么拆才不返工

爬虫写到最后烂掉,往往不是接口问题,而是所有代码堆在一个文件里,加需求就改结构。我常用的拆法是五个文件:

wbpoi/ ├── config.py # Cookie、关键词、数据库连接、延迟区间 ├── scheduler.py # 请求调度:Session、重试、延时窗口 ├── parser.py # 从 cards 里抽 POI 字段 ├── storage.py # SQLAlchemy 入库与去重 └── run.py # 命令行入口,串起整个流程

config 只放配置,scheduler 只负责拿响应,parser 只做解析,storage 只负责写入。相互之间通过参数传递数据,不搞全局变量。这样做的直接好处是:换 Cookie 不用动解析逻辑,换存储方式不用动请求逻辑。

3.2 config.py:把会话信息与参数集中管理

第一步先建配置模块。Cookie 单独拎出来,方便失效时快速替换:

# config.py import random # 登录微博后在开发者工具里复制,失效就换,不要硬刚 COOKIE = "SUB=xxx; SUBP=xxx; M_WEIBO_SESSION=xxx" # 移动端接口对 Referer 敏感,缺失时可能返回空 data BASE_HEADERS = { "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 13_0 like Mac OS X)", "Referer": "https://m.weibo.cn/", "X-Requested-With": "XMLHttpRequest", "Accept": "application/json, text/plain, */*", } # 关键词列表:建议一次跑几个就定义几个 KEYWORDS = ["商圈", "网红店", "新店开业"] # 延时区间,单位秒;新浪对短时间高频请求有风控 DELAY_RANGE = (5, 12) # 是否把原始返回包落盘,排障神器 SAVE_RAW_JSON = True

COOKIE 里有三个占位字段只是示意,实际以你账号抓到的为准。BASE_HEADERS 里最关键的是 Referer,不少移动端接口缺失它会返回空 data。DELAY_RANGE 是我常用的保守区间,标签页里看到的数据瞬间加载快,不代表爬虫也能这么跑,5-12 秒的随机延时能让任务稳定跑几个小时。

SAVE_RAW_JSON 这个开关容易被忽略。它把每次响应原文落盘,后面解析字段出错时,不用重新请求,直接翻原包比对。

3.3 scheduler.py:Session、超时和失败重试

请求调度是整个爬虫里决定生死的一环。用 Session 保持连接,给每个请求套上随机延时和有限重试:

# scheduler.py import time import requests from random import uniform from config import COOKIE, BASE_HEADERS, DELAY_RANGE class PoiScheduler: def __init__(self): self.session = requests.Session() self.session.headers.update(BASE_HEADERS) self.session.cookies.update(dict(cookie.split("=", 1) for cookie in COOKIE.split("; "))) def fetch_json(self, url, params): # 每次请求前先睡一个随机时间,避免固定节奏被风控识别 time.sleep(uniform(*DELAY_RANGE)) for attempt in range(3): try: resp = self.session.get(url, params=params, timeout=10) # 登录跳转会以 HTML 形式返回,不是 JSON if resp.text.strip().startswith("<"): raise ValueError("got html instead of json, cookie may be expired") data = resp.json() if data.get("ok") == 1: return data else: raise ValueError(f"api not ok: {data.get('msg')}") except Exception as e: # 最后两次尝试之间再等一下,别疯狂打请求 if attempt < 2: time.sleep(10 + attempt * 10) continue print(f"[fetch failed] {url} error: {e}") return None

fetch_json 做了三件事:随机延时、自动重试、返回体合法性检查。重点说两个参数:

  • timeout=10:连接和读取都设上限,不设 timeout 的爬虫会在网络抖动时挂死。
  • attempt 最多 3 次:超过 3 次直接返回 None,不要让失败任务无限循环。

代码里检查返回体以<开头就抛异常,这一手很实用。微博风控时返回的是 HTML 验证页,状态码仍然是 200,只有看 body 才能识别。

3.4 parser.py:从 cards 数组里抽 POI 字段

拿到响应后,数据藏在 data.cards 这个数组里。每张卡片类型不同,有的是普通微博,有的是带位置的微博。解析必须做类型判断:

# parser.py from urllib.parse import unquote def extract_locations(data): """从 getIndex 响应中抽取 POI 信息列表""" results = [] cards = (data.get("data") or {}).get("cards") or [] for card in cards: # 卡片类型标识,不同版本可能为 card_type 或 cardType if not card.get("card_type") == 9 and not card.get("cardType") == 9: continue mblog = card.get("mblog") or {} location = mblog.get("location") or {} # 部分微博只有地点名没有 POI_ID,这类数据没法补全,跳过 poi_id = location.get("poi_id") or location.get("poiid") if not poi_id: continue result = { "poi_id": poi_id, "poi_title": location.get("poi_title") or location.get("poiname"), "weibo_id": mblog.get("id"), "text": mblog.get("text_raw") or mblog.get("text"), "created_at": mblog.get("created_at"), "lon": location.get("lon"), "lat": location.get("lat"), "category": location.get("category"), } # 微博返回的地点名偶尔是 URL 编码,统一还原 if result["poi_title"] and "%" in result["poi_title"]: result["poi_title"] = unquote(result["poi_title"]) results.append(result) return results

这里我同时兼容了 card_type 和 cardType 两种写法,因为微博接口升级时改过字段名。poi_title 和 poiname 同理。用or做兜底,是解析这类接口最省心的做法。

filter 条件是 poi_id 必须非空,这是整条链路的命门。拿不到 poi_id 就意味着后续无法补全经纬度,宁可丢弃也不要污染数据库。

3.5 run.py:把最小闭环串起来

调度器和解析器都有了,还需要一个入口把单关键词翻页跑起来:

# run.py from scheduler import PoiScheduler from parser import extract_locations from config import KEYWORDS def crawl_keyword(scheduler, keyword, max_rounds=20): url = "https://m.weibo.cn/api/container/getIndex" # containerid 里的中文必须编码,这里直接用 requests 的 params 自动编码 containerid = f"100103type=1&q={keyword}" params = {"containerid": containerid} seen = set() # 用 weibo_id 去重,防止游标异常导致重复 for rnd in range(max_rounds): data = scheduler.fetch_json(url, params) if not data: break for item in extract_locations(data): if item["weibo_id"] not in seen: seen.add(item["weibo_id"]) print(item) # 游标:取响应里的 since_id,取不到就结束翻页 next_cursor = (data.get("data") or {}).get("cardlistInfo", {}).get("since_id") if not next_cursor: break params["since_id"] = next_cursor

这个循环的控制点有两个:seen 集合做微博 ID 去重,since_id 做游标推进。两个都不可省。游标字段的取值路径我写成 data.cardlistInfo.since_id,但不同接口版本可能放在 data.since_id,建议打开一个返回包确认。

max_rounds 设 20 是保守值,理想情况下应该由游标为空来终止循环,但这个参数作为安全阀,防止某个接口永远返回有效游标导致死循环。

4. 把数据存下来:SQLAlchemy 建表、字段清洗与断点续爬

4.1 为什么不用裸 sqlite3,而是选 SQLAlchemy

只有几百条数据时,裸 sqlite3 完全够用。但 POI 爬虫的特点是字段多、去重逻辑复杂、后续要叠微博表关联查询,这时候写裸 SQL 会消耗大量时间在拼接字符串上。

选型优点痛点
裸 sqlite3零依赖、上手快字段变化时改 SQL 麻烦
CSV 直接落盘可视化工具直接打开去重和增量更新几乎不可做
SQLAlchemy ORM建表即模型,字段映射清晰初次上手需要理解 session 概念

SQLAlchemy 的好处是表结构即代码。字段增加时改类定义就能重新建表,清洗逻辑也天然和 ORM 的字段类型绑定。

4.2 POI 表模型与唯一键设计

POI 数据和微博数据要拆成两张表,因为它们是不同的实体。POI 表以 poi_id 为唯一键,微博表以 weibo_id 为唯一键:

# models.py from sqlalchemy import Column, String, Float, BigInteger, create_engine from sqlalchemy.orm import declarative_base, sessionmaker Base = declarative_base() class WbPoi(Base): __tablename__ = "wb_poi" poi_id = Column(String(64), primary_key=True) poi_title = Column(String(255), nullable=False) category = Column(String(128)) lon = Column(Float) lat = Column(Float) province = Column(String(32)) city = Column(String(32)) checkin_num = Column(BigInteger, default=0) updated_at = Column(String(32)) class WbPost(Base): __tablename__ = "wb_post" weibo_id = Column(String(64), primary_key=True) poi_id = Column(String(64), index=True) text = Column(String(2048)) created_at = Column(String(32))

两个模型对应前面解析的语义:微博先关联 poi_id,再查 POI 表拿到经纬度和分类。这样设计是为了避免每条微博重复存一份地点信息,空间冗余大且容易数据不一致。唯一键分别落在 poi_id 和 weibo_id 上,天然支持后续幂等写入。

注意 weibo_id 用 String 而不是 BigInteger。微博 ID 的数值范围可能超出 32 位整数,但 Twitter 式的 snowflake ID 在部分语言里按整数解析会丢精度,统一用字符串最稳。

4.3 字段清洗:经纬度、分类和编码还原

入库前必须做一层清洗,否则查询时会发现同一地点拆成三五行。常见问题有三个:经纬度是字符串、分类字段缺省、地名是 URL 编码或夹杂 \u 转义。

# storage.py from urllib.parse import unquote def clean_poi_record(raw: dict) -> dict: """把解析出来的字段清洗成可入库的结构""" def _to_float(val): try: return float(val) except (TypeError, ValueError): return None def _clean_text(val): if not val: return "" # 微博会返回 \u 转义和 % 编码,两种都还原 text = val.replace("\\u", "\u") if "%" in text: text = unquote(text) return text.strip() return { "poi_id": str(raw.get("poi_id") or "").strip(), "poi_title": _clean_text(raw.get("poi_title")), "category": _clean_text(raw.get("category")), "lon": _to_float(raw.get("lon")), "lat": _to_float(raw.get("lat")), }

_to_float 把接口返回的字符串坐标转成浮点数,同时在字段缺失时返回 None,不让脏数据以 0.0 的形式入库——经纬度是 0.0 的点一眼看是合法的,实际是无效数据。_clean_text 处理两种编码形态,尤其注意有些数据在字符串里写的是 \u,不处理就会被原样入库。

4.4 断点续爬的核心:以游标和数据库双重状态为锚点

脚本跑一半断网是常态。断点续爬要做两件事:记录上一个成功的游标;写入前先查库去重。

import json, os def save_checkpoint(keyword, since_id): path = f"checkpoint/{keyword}.json" os.makedirs("checkpoint", exist_ok=True) with open(path, "w", encoding="utf-8") as f: json.dump({"keyword": keyword, "since_id": since_id}, f) def load_checkpoint(keyword): path = f"checkpoint/{keyword}.json" if os.path.exists(path): with open(path, encoding="utf-8") as f: return json.load(f).get("since_id") return None

checkpoint 文件保存最后成功的游标,重启后从该游标继续。数据库去重则依托前面定义的主键,写入时用 merge 而不是普通 add:

session.merge(poi_row) # 按主键判断存在则更新,不存在则插入

merge 的行为是按主键查询,存在就更新非主键字段,不存在则新增。这比先 select 再 insert 省一次查询,也比盲目 insert 安全。

5. 微博POI爬虫避坑现场:5 个让我返工到凌晨的问题

5.1 坐标拿到了,但全是 0.0

现象:入库后统计经度分布,发现四分之一的点落在(0.0, 0.0),画地图时这些点全堆在非洲几内亚湾。

原因:接口返回的部分卡片 location 字段只带 poi_id 不带坐标,或者 lon/lat 被解析成字符串"None"。解析器只判断了 poi_id 是否存在,没判断坐标是否有效。

解决:入库前统一过一遍坐标校验。经纬度都非空、且经度绝对值小于 180、纬度绝对值小于 90,不满足就标记为待补全,单独存一张表,后续用 poi_id 调地点详情接口回填。

5.2 脚本跑二十分钟就 302 跳转到登录页

现象:日志里突然出现大量 HTML 片段,打印出来的响应不再是 JSON。

原因:Cookie 被风控,或者请求频率触发接口限流,服务端直接返回 302 到 passport 登录页。

解决:第一道防线是随机延时,第二道防线是响应体检测。我上面代码里写了if resp.text.strip().startswith("<")的判断,一旦命中就立即停止当前任务,而不是带着假 Cookie 重试 20 次,把账号越搞越黑。

5.3 地名入库全是 %E5%8C%97%E4%BA%AC

现象:数据库里 poi_title 显示<url>encoded形式的字符串,业务方直接看懵。

原因:接口某些返回字段是 URL 编码,解析器没有统一做 unquote。问题往往只在部分卡片出现,因为老数据走了编码字段,新数据走了明文字段。

解决:在清洗函数里统一处理,只要字符串里含 % 就执行 unquote。这里记住一个原则:宁可全量清洗,不要按来源分情况处理,分支越多遗漏越多。

5.4 翻到第三页就永远重复第三页的数据

现象:任务能跑,但入库数据只有 100 条左右,去重后发现全是同一批 ID 在循环。

原因:接口不吃固定 page 参数,而新写的代码把页码放进了 params,游标 since_id 始终没有更新,每个请求返回同一区间。

解决:页码参数去掉,只保留游标。每个响应都从 cardlistInfo 里取 since_id 传给下一轮,游标为空才算翻页结束。

5.5 定时任务凌晨跑,数据量比白天少一半

现象:白天手动跑正常,挂 cron 凌晨两点自动跑,发现入库量只有一半。

原因:账号触发了夜间风控。新 IP 在凌晨访问时,接口会下发验证码页,而脚本只判断了 HTTP 200,没判断 body 内容,把验证码页当成了空数据跳过。

解决:把状态判断从“是否 200”升级为“body 是否为 JSON”。我当时在调度器里加了专门检测,命中验证码特征就发通知并停止任务,不再做无效重试。另外把延时区间从 5-12 秒放宽到 8-18 秒,凌晨的请求节奏更慢,存活时间明显变长。

6. 把结果变成能用的东西:热力地图和可复用的命令行入口

6.1 用 folium 把 POI 数据画成热力图

数据入库后,最直接的验证方式是出图。用 folium 叠加 HeatMap 插件,几十行就能看到分布是否合理:

import folium from folium.plugins import HeatMap from sqlalchemy.orm import sessionmaker from models import WbPoi, engine points = [(row.lat, row.lon) for row in sessionmaker(bind=engine)().query(WbPoi).all() if row.lat and row.lon] m = folium.Map(location=[points[0] if points else [39.9, 116.4]], zoom_start=11) HeatMap(points, radius=15, blur=10).add_to(m) m.save("poi_heatmap.html")

radius 控制热力半径,像素值越大,热点越扩散;blur 是模糊半径,调大适合看城市群轮廓,调小适合分辨单点。我的习惯是先 radius=15 粗看分布,再降成 8 检查是否是某个字段解析错误导致的单点高度聚集。

6.2 加一个命令行入口,让爬虫随时可调

把 run.py 升级成 argparse 入口,方便定时任务和手动调试共用同一套代码:

python run.py crawl --keyword 商圈 --max-rounds 20 --db sqlite:///wbpoi.db

参数里 keyword 支持多个值逗号分隔,max-rounds 控制安全上限,db 指定数据库连接。这样 cron 里指定一个完整命令即可,不用改代码。

6.3 验证数据的两个土办法

第一,统计每个城市的 POI 数量,如果某个城市的分布不符合常识,优先怀疑经纬度解析而非网络问题。第二,抽几行数据和高德 POI 数据做交叉比对。微博 POI 是用户签到生成的,高德 POI 是地图厂商维护的,两者名称不完全一致但坐标应落在同一区域,偏差超过几百米说明坐标系转换出了问题。

我现在的习惯是任何爬虫项目开工前先留一个 raw_json 开关,把原始包落盘。改字段映射时直接翻旧数据,重爬的成本永远比想象的高。希望帮到你。

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

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询