简介面向需要采集微博地理位置信息的研究者与开发者该源码是一套基于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 在微博里到底以什么形态存在微博的 POIPoint 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 的完整档案关键词搜索100103type1q关键词按文本关键词找微博再从中筛带位置的数据URL 编码是必须做的。中文关键词做 query 参数时requests 的 params 会自动编码但如果有人手工拼 URL漏掉编码就会看到一堆 %E5%8C%97%E4%BA%AC 这样的内容。注意一个细节你抓包看到的 containerid 后缀可能与我上面写的有差异因为账号权限和接口版本会微调。正确的做法不是背参数而是先用浏览器登录微博打开抓包工具点开一条带位置的微博看真实请求里的容器ID格式再把它填进配置。2.3 翻页三件套pagebar、since_id 与 max_id 的分工这是新手最容易卡住的地方。移动端列表不是后端分页而是“加载更多”的游标翻页。接口里常见两类参数pagebar0 表示首页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?containerid100103type%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 SUBxxx; SUBPxxx; M_WEIBO_SESSIONxxx # 移动端接口对 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 TrueCOOKIE 里有三个占位字段只是示意实际以你账号抓到的为准。BASE_HEADERS 里最关键的是 Referer不少移动端接口缺失它会返回空 data。DELAY_RANGE 是我常用的保守区间标签页里看到的数据瞬间加载快不代表爬虫也能这么跑5-12 秒的随机延时能让任务稳定跑几个小时。SAVE_RAW_JSON 这个开关容易被忽略。它把每次响应原文落盘后面解析字段出错时不用重新请求直接翻原包比对。3.3 scheduler.pySession、超时和失败重试请求调度是整个爬虫里决定生死的一环。用 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, paramsparams, timeout10) # 登录跳转会以 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(fapi 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 Nonefetch_json 做了三件事随机延时、自动重试、返回体合法性检查。重点说两个参数timeout10连接和读取都设上限不设 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_rounds20): url https://m.weibo.cn/api/container/getIndex # containerid 里的中文必须编码这里直接用 requests 的 params 自动编码 containerid f100103type1q{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_keyTrue) poi_title Column(String(255), nullableFalse) category Column(String(128)) lon Column(Float) lat Column(Float) province Column(String(32)) city Column(String(32)) checkin_num Column(BigInteger, default0) updated_at Column(String(32)) class WbPost(Base): __tablename__ wb_post weibo_id Column(String(64), primary_keyTrue) poi_id Column(String(64), indexTrue) 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 fcheckpoint/{keyword}.json os.makedirs(checkpoint, exist_okTrue) with open(path, w, encodingutf-8) as f: json.dump({keyword: keyword, since_id: since_id}, f) def load_checkpoint(keyword): path fcheckpoint/{keyword}.json if os.path.exists(path): with open(path, encodingutf-8) as f: return json.load(f).get(since_id) return Nonecheckpoint 文件保存最后成功的游标重启后从该游标继续。数据库去重则依托前面定义的主键写入时用 merge 而不是普通 addsession.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 显示urlencoded形式的字符串业务方直接看懵。原因接口某些返回字段是 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(bindengine)().query(WbPoi).all() if row.lat and row.lon] m folium.Map(location[points[0] if points else [39.9, 116.4]], zoom_start11) HeatMap(points, radius15, blur10).add_to(m) m.save(poi_heatmap.html)radius 控制热力半径像素值越大热点越扩散blur 是模糊半径调大适合看城市群轮廓调小适合分辨单点。我的习惯是先 radius15 粗看分布再降成 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 开关把原始包落盘。改字段映射时直接翻旧数据重爬的成本永远比想象的高。希望帮到你。本文还有配套的精品资源点击获取
简介面向需要采集微博地理位置信息的研究者与开发者该源码是一套基于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 在微博里到底以什么形态存在微博的 POIPoint 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 的完整档案关键词搜索100103type1q关键词按文本关键词找微博再从中筛带位置的数据URL 编码是必须做的。中文关键词做 query 参数时requests 的 params 会自动编码但如果有人手工拼 URL漏掉编码就会看到一堆 %E5%8C%97%E4%BA%AC 这样的内容。注意一个细节你抓包看到的 containerid 后缀可能与我上面写的有差异因为账号权限和接口版本会微调。正确的做法不是背参数而是先用浏览器登录微博打开抓包工具点开一条带位置的微博看真实请求里的容器ID格式再把它填进配置。2.3 翻页三件套pagebar、since_id 与 max_id 的分工这是新手最容易卡住的地方。移动端列表不是后端分页而是“加载更多”的游标翻页。接口里常见两类参数pagebar0 表示首页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?containerid100103type%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 SUBxxx; SUBPxxx; M_WEIBO_SESSIONxxx # 移动端接口对 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 TrueCOOKIE 里有三个占位字段只是示意实际以你账号抓到的为准。BASE_HEADERS 里最关键的是 Referer不少移动端接口缺失它会返回空 data。DELAY_RANGE 是我常用的保守区间标签页里看到的数据瞬间加载快不代表爬虫也能这么跑5-12 秒的随机延时能让任务稳定跑几个小时。SAVE_RAW_JSON 这个开关容易被忽略。它把每次响应原文落盘后面解析字段出错时不用重新请求直接翻原包比对。3.3 scheduler.pySession、超时和失败重试请求调度是整个爬虫里决定生死的一环。用 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, paramsparams, timeout10) # 登录跳转会以 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(fapi 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 Nonefetch_json 做了三件事随机延时、自动重试、返回体合法性检查。重点说两个参数timeout10连接和读取都设上限不设 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_rounds20): url https://m.weibo.cn/api/container/getIndex # containerid 里的中文必须编码这里直接用 requests 的 params 自动编码 containerid f100103type1q{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_keyTrue) poi_title Column(String(255), nullableFalse) category Column(String(128)) lon Column(Float) lat Column(Float) province Column(String(32)) city Column(String(32)) checkin_num Column(BigInteger, default0) updated_at Column(String(32)) class WbPost(Base): __tablename__ wb_post weibo_id Column(String(64), primary_keyTrue) poi_id Column(String(64), indexTrue) 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 fcheckpoint/{keyword}.json os.makedirs(checkpoint, exist_okTrue) with open(path, w, encodingutf-8) as f: json.dump({keyword: keyword, since_id: since_id}, f) def load_checkpoint(keyword): path fcheckpoint/{keyword}.json if os.path.exists(path): with open(path, encodingutf-8) as f: return json.load(f).get(since_id) return Nonecheckpoint 文件保存最后成功的游标重启后从该游标继续。数据库去重则依托前面定义的主键写入时用 merge 而不是普通 addsession.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 显示urlencoded形式的字符串业务方直接看懵。原因接口某些返回字段是 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(bindengine)().query(WbPoi).all() if row.lat and row.lon] m folium.Map(location[points[0] if points else [39.9, 116.4]], zoom_start11) HeatMap(points, radius15, blur10).add_to(m) m.save(poi_heatmap.html)radius 控制热力半径像素值越大热点越扩散blur 是模糊半径调大适合看城市群轮廓调小适合分辨单点。我的习惯是先 radius15 粗看分布再降成 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 开关把原始包落盘。改字段映射时直接翻旧数据重爬的成本永远比想象的高。希望帮到你。本文还有配套的精品资源点击获取
