简介:本资源是一套完整可用的微博数据爬虫实战项目,面向计算机专业本科生、毕业设计与课程设计学生,以及Python/Java双栈开发初学者,解决社交平台公开数据采集、清洗与分析的实际需求。压缩包共190个文件,2.58MB,涵盖34个核心Python脚本(含爬虫主逻辑、反爬绕过、数据解析模块)、22个JavaScript前端交互文件(用于模拟用户行为与页面渲染)、11个JSON格式配置与结果数据、以及C语言底层驱动代码(如ds1302.c、DHT11.c等嵌入式通信模块),体现软硬协同设计思路;另有JPG/PNG图表、CSS/HTML可视化界面及VS工程文件(.sln、.csproj),支持开箱即用与二次开发。已有62人学习下载,资源经严格评审获95分高分,包含完整项目报告、可运行源码、测试截图与部署说明,特别适合课设答辩、毕设原型构建及数据采集类项目实战演练。
1. 微博数据爬虫不是“点开网页右键另存为”——它是一套需绕过动态渲染、反爬策略与接口鉴权的工程化数据采集链路
很多人看到“微博数据爬虫-点赞转发评论全获取”这个标题,第一反应是:不就是用 requests 请求几个 URL 吗?但实际落地时,90% 的人卡在第一步——连微博首页都拿不到完整 HTML。原因很简单:微博 PC 端和移动端均采用 React + SSR 混合渲染,关键字段(如微博 ID、用户 UID、互动数)全部由 JavaScript 动态注入;更关键的是,所有公开接口(如https://weibo.com/ajax/statuses/repostTimeline、https://weibo.com/ajax/statuses/buildComments)均强制校验X-XSRF-TOKEN、Cookie中的SUB和SUHB,且请求头必须携带User-Agent、Referer、X-Requested-With三要素,缺一不可。这不是“写个 for 循环就能跑通”的脚本,而是涉及登录态维持、接口逆向、参数签名、频率控制、异常重试的端到端数据管道。适合需要批量分析竞品运营效果、舆情监测、学术研究中微博传播路径的 IT 工程师、数据分析师与合规合规的数据采集人员——前提是已获得平台公开数据授权或仅用于个人非商业研究场景。
2. 从登录态构建到接口逆向:微博互动数据采集的三层可信链路设计
微博数据采集不能靠模拟浏览器点击,而要建立可复现、可审计、可降级的三层链路:认证层 → 接口层 → 解析层。这三层不是线性流程,而是相互校验的闭环。认证层决定你能否拿到有效 Cookie;接口层决定你能调通哪些 endpoint;解析层决定你是否能稳定提取结构化字段。跳过任一层,都会导致“今天能跑,明天 403”或“返回空数组却无报错”的典型问题。
2.1 认证层:不依赖 Selenium,用 Requests + 手动 Cookie 注入实现轻量登录态复用
微博 Web 端已全面弃用账号密码直登,改用扫码登录或手机短信登录。但对爬虫而言,最稳定的方式是复用已有登录态 Cookie。手动获取方式如下:
- 使用 Chrome 登录微博账号,打开开发者工具(F12)→ Application → Cookies → 复制
SUB、SUBP、SSOLoginState、XSRF-TOKEN四个关键字段; - 构建最小化 Session 对象,显式设置 Referer 和 User-Agent:
import requests session = requests.Session() session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", "Referer": "https://weibo.com/", "X-Requested-With": "XMLHttpRequest" }) cookies = { "SUB": "_2AkMmZqk-f8NhqwJRmPEaxurqaT6HzH7EieOaYJc_EPxMKDNzTnAKqjBtT5QbDvLWfVnKuUwGdNpRlIgSsC1yZi1hIj8_", "SUBP": "u32Cc72-4rM7e--fSiq5ir5g5x2RBj3l1.LR8ZQoA7jQZfQXzqQZ5JpX", "SSOLoginState": "1715823456", "XSRF-TOKEN": "1234567890abcdef" } session.cookies.set_dict(cookies)提示:
XSRF-TOKEN不是固定字符串,它会随每次页面加载刷新,必须从https://weibo.com/响应 HTML 的<script>标签中正则提取(如window.$CONFIG = {.*?xsrfToken: "(.*?)",),否则后续所有接口将返回{"code": 100001, "msg": "invalid token"}。
2.2 接口层:微博互动数据三大核心 endpoint 及其必传参数表
微博未开放官方 API 获取完整互动数据,但其前端 JS 会调用以下三个真实接口,经实测(2024 年 Q2)仍可用,且支持分页与增量拉取:
| 接口地址 | 用途 | 必传 Query 参数 | 是否需 X-XSRF-TOKEN Header |
|---|---|---|---|
https://weibo.com/ajax/statuses/repostTimeline | 获取转发列表 | id(微博 bid)、page(页码,从 1 开始)、count(每页条数,默认 20) | 是 |
https://weibo.com/ajax/statuses/buildComments | 获取评论列表 | id(微博 bid)、page(页码)、count(默认 20)、type(1=普通评论,2=热门评论) | 是 |
https://weibo.com/ajax/statuses/likeCount | 获取点赞总数 | id(微博 bid) | 否(仅 GET,无需 token) |
其中bid是微博唯一标识符,格式为 9~10 位字母数字组合(如KxYzAbC12),不是 mid。它存在于微博详情页 URL 中(https://weibo.com/xxxxxx/KxYzAbC12)或通过https://weibo.com/ajax/statuses/show?id=xxx接口返回的data.bid字段中。
2.2.1 转发数据拉取:带重试与状态码校验的健壮请求封装
def fetch_reposts(session, bid, page=1, count=20): url = "https://weibo.com/ajax/statuses/repostTimeline" params = {"id": bid, "page": page, "count": count} headers = {"X-XSRF-TOKEN": session.cookies.get("XSRF-TOKEN", "")} for attempt in range(3): try: resp = session.get(url, params=params, headers=headers, timeout=10) if resp.status_code == 200: data = resp.json() if data.get("ok") == 1 and "data" in data: return data["data"].get("reposts", []) elif data.get("code") == 100001: raise ValueError("XSRF-TOKEN expired, need refresh") elif resp.status_code == 418: # 微博反爬常见状态码:I'm a teapot time.sleep(2 ** attempt) # 指数退避 continue except Exception as e: print(f"[REPOST] Error on page {page}: {e}") time.sleep(1) return [] # 示例调用 reposts = fetch_reposts(session, bid="KxYzAbC12", page=1) for r in reposts[:3]: print(f"转发者UID: {r['user']['id']}, 内容: {r['text'][:30]}...")注意:
repostTimeline接口返回的user.id是用户 UID(如1234567890),不是昵称;text字段含 HTML 实体(如>),需用html.unescape()解码;created_at为 ISO8601 字符串(如"2024-05-15T14:23:11"),可直接转为datetime对象。
2.3 解析层:从 JSON 响应中稳定提取结构化字段的字段映射表
微博接口返回 JSON 结构嵌套深、字段名不统一(如评论有user和user_info两套字段),必须定义明确的字段映射规则,避免因前端微调导致解析失败:
| 原始 JSON 路径 | 目标字段名 | 类型 | 说明 |
|---|---|---|---|
user.id/user_info.id | uid | str | 用户唯一标识,可用于去重或关联用户资料 |
user.screen_name/user_info.screen_name | nickname | str | 昵称,可能含 emoji,建议 UTF-8 存储 |
text | content | str | 原始文本,含@、#、链接等,需清洗 |
created_at | created_at | datetime | 发布时间,注意时区为 UTC+8 |
source | device | str | 来源设备(如"iPhone客户端"),需正则提取关键词 |
like_count | like_count | int | 当前点赞数(仅转发/评论对象自身,非原博) |
import html from datetime import datetime def parse_repost_item(item): user = item.get("user") or item.get("user_info", {}) return { "uid": str(user.get("id", "")), "nickname": user.get("screen_name", ""), "content": html.unescape(item.get("text", "")), "created_at": datetime.fromisoformat(item["created_at"].replace(" ", "T")), "device": item.get("source", "").strip("来自"), "like_count": item.get("like_count", 0) } # 应用解析 parsed_reposts = [parse_repost_item(r) for r in fetch_reposts(session, "KxYzAbC12")]3. 全量数据采集实战:从单条微博到批量 UID 的可配置 pipeline
单条微博的点赞、转发、评论只是起点。真实业务场景中,你需要:① 批量输入微博 bid 列表;② 自动发现关联用户 UID;③ 按时间范围增量采集;④ 输出标准 CSV/JSONL。这就要求把前述模块组装成可配置 pipeline,而非一次性脚本。
3.1 输入层:支持三种微博 ID 来源的标准化读取器
微博数据采集入口不能只靠人工复制 bid。常见来源有:
- URL 列表:
https://weibo.com/2803301701/KxYzAbC12 - UID + 时间范围:通过
https://weibo.com/ajax/profile/getWeibo?uid=2803301701&page=1&count=10获取某用户近期微博列表 - 关键词搜索结果:调用
https://weibo.com/ajax/side/hotSearch获取热搜,再用https://weibo.com/ajax/searchall?keyword=xxx&page=1获取相关微博
下面是一个通用 bid 提取函数,兼容上述三种输入:
import re from urllib.parse import urlparse, parse_qs def extract_bid_from_input(input_str): """从 URL、UID 或纯 bid 字符串中提取 bid""" # case 1: URL like https://weibo.com/2803301701/KxYzAbC12 url_match = re.search(r"weibo\.com/\d+/([a-zA-Z0-9]{9,10})", input_str) if url_match: return url_match.group(1) # case 2: pure bid (9-10 chars alphanumeric) bid_match = re.match(r"^[a-zA-Z0-9]{9,10}$", input_str.strip()) if bid_match: return input_str.strip() # case 3: UID (pure digits), trigger profile fetch uid_match = re.match(r"^\d{5,12}$", input_str.strip()) if uid_match: return fetch_bids_by_uid(session, uid_match.group(0), limit=5) # 返回 list raise ValueError(f"Cannot extract valid bid from: {input_str}") def fetch_bids_by_uid(session, uid, limit=10): """根据 UID 获取最近 limit 条微博的 bid 列表""" url = f"https://weibo.com/ajax/profile/getWeibo" params = {"uid": uid, "page": 1, "count": limit} resp = session.get(url, params=params) if resp.status_code == 200: data = resp.json() return [item["bid"] for item in data.get("data", {}).get("list", [])[:limit]] return []3.2 控制层:可中断、可续采、带进度日志的采集调度器
为避免网络波动或反爬限流导致全量任务失败,必须实现断点续采。核心是记录每个 bid 的采集状态(pending/done/failed)到本地 SQLite 数据库:
import sqlite3 from pathlib import Path DB_PATH = Path("weibo_crawl.db") def init_db(): conn = sqlite3.connect(DB_PATH) conn.execute(""" CREATE TABLE IF NOT EXISTS crawl_status ( bid TEXT PRIMARY KEY, status TEXT DEFAULT 'pending', last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP, error_msg TEXT ) """) conn.commit() conn.close() def mark_as_done(bid): conn = sqlite3.connect(DB_PATH) conn.execute("INSERT OR REPLACE INTO crawl_status (bid, status) VALUES (?, 'done')", (bid,)) conn.commit() conn.close() def get_pending_bids(limit=100): conn = sqlite3.connect(DB_PATH) cursor = conn.execute("SELECT bid FROM crawl_status WHERE status = 'pending' LIMIT ?", (limit,)) bids = [row[0] for row in cursor.fetchall()] conn.close() return bids # 主采集循环 init_db() pending_bids = get_pending_bids(limit=50) for bid in pending_bids: try: print(f"[START] Processing bid {bid}") reposts = fetch_reposts(session, bid, page=1) comments = fetch_comments(session, bid, page=1) like_count = fetch_like_count(session, bid) # 保存到 CSV(按 bid 分文件) save_to_csv(f"output/{bid}_reposts.csv", reposts) save_to_csv(f"output/{bid}_comments.csv", comments) save_to_json(f"output/{bid}_meta.json", {"bid": bid, "like_count": like_count}) mark_as_done(bid) print(f"[DONE] bid {bid} saved") except Exception as e: print(f"[FAIL] bid {bid}: {e}") mark_as_failed(bid, str(e))提示:
save_to_csv()应使用csv.DictWriter并指定fieldnames为预定义字段列表(如["uid","nickname","content","created_at"]),确保列顺序一致;文件名含bid可避免多任务写入冲突;CSV 编码必须为utf-8-sig,否则 Excel 打开中文乱码。
3.3 输出层:生成含统计摘要与原始明细的双层报告结构
“含全部资料+报告.zip”中的“报告”,不应是 Word 或 PDF,而应是机器可读、人可查的结构化产物。推荐输出以下两类文件:
- 明细层:
{bid}_reposts.csv、{bid}_comments.csv、{bid}_meta.json—— 原始数据,字段严格对齐 2.3 节映射表; - 汇总层:
summary_report.json—— 包含总微博数、总互动数、用户去重数、时间分布直方图(按小时聚合)、高频设备统计等。
# summary_report.json 示例结构 { "crawl_time": "2024-05-18T10:23:45+08:00", "total_weibos": 12, "total_reposts": 342, "total_comments": 1897, "unique_uids": 1243, "top_devices": [ {"device": "iPhone客户端", "count": 872}, {"device": "Android客户端", "count": 651} ], "hourly_distribution": { "00": 12, "01": 8, ..., "23": 21 } }该报告可直接导入 BI 工具(如 Metabase、Superset)做可视化,也可用 Pandas 快速生成折线图/词云。
4. 避坑指南:微博反爬升级后最常触发的 5 类错误及对应解法
微博反爬策略持续迭代,2024 年以来新增了多项检测机制。以下 5 类错误出现频率最高,且均有明确解法,无需更换 IP 或购买代理池。
4.1 错误类型 1:{"code":100001,"msg":"invalid token"}—— XSRF-TOKEN 过期
现象:所有带X-XSRF-TOKEN的接口均返回此错误,但 Cookie 中XSRF-TOKEN字段存在。
根因:微博服务端会校验XSRF-TOKEN与当前SUBCookie 的绑定关系,且该绑定 30 分钟失效。
解法:每次请求前,先 GEThttps://weibo.com/,从响应 HTML 中用正则提取最新 token:
def refresh_xsrf_token(session): resp = session.get("https://weibo.com/", timeout=5) match = re.search(r'xsrfToken\s*:\s*"([^"]+)"', resp.text) if match: session.headers["X-XSRF-TOKEN"] = match.group(1) session.cookies.set("XSRF-TOKEN", match.group(1)) else: raise RuntimeError("Failed to extract XSRF-TOKEN from homepage")4.2 错误类型 2:status_code == 418—— 请求被识别为自动化流量
现象:请求返回 HTTP 418(I'm a teapot),无 JSON body,headers 中x-rate-limit-remaining为 0。
根因:微博服务端基于请求指纹(User-Agent + Referer + 请求间隔)判定为 bot。
解法:
- 在
session.headers中固定User-Agent(不要随机); - 所有接口请求
Referer必须为https://weibo.com/(不是微博某条具体博文页); - 加入
time.sleep(random.uniform(1.5, 3.0)),避免固定间隔; - 关键:禁用 connection pooling,每次请求新建 TCP 连接:
from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry session.mount("https://", HTTPAdapter( pool_connections=1, pool_maxsize=1, max_retries=Retry(total=0) ))4.3 错误类型 3:评论返回{"ok":0,"msg":"暂无数据"}—— 评论开关关闭或权限不足
现象:buildComments接口返回空数据,但微博页面可见评论。
根因:原博主设置了“仅好友可见评论”或“关闭评论”。
解法:先调用https://weibo.com/ajax/statuses/show?id={bid},检查返回 JSON 中data.comments_count是否 > 0;若为 0,则跳过评论采集,记录reason: "comments_disabled"。
4.4 错误类型 4:UnicodeEncodeError: 'gbk' codec can't encode char—— Windows 下 CSV 写入中文失败
现象:save_to_csv()报编码错误,尤其在 Windows 系统。
解法:显式指定encoding='utf-8-sig',并用newline=''避免空行:
with open(filename, "w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(data)4.5 错误类型 5:bid解析失败导致repostTimeline返回{"ok":0,"msg":"参数错误"}
现象:接口返回参数错误,但 bid 看似合法。
根因:微博 bid 实际是 base62 编码的 mid,长度固定为 9 或 10,但部分旧微博(2018 年前)使用 8 位 bid,已被新接口废弃。
解法:增加 bid 格式校验:
def validate_bid(bid): if not isinstance(bid, str): return False if not re.match(r"^[a-zA-Z0-9]{9,10}$", bid): return False # 额外校验:尝试访问 show 接口 resp = session.get(f"https://weibo.com/ajax/statuses/show?id={bid}") return resp.status_code == 200 and resp.json().get("ok") == 15. 进阶技巧:用 UID 反查微博账号基础信息,构建用户画像关联网络
仅采集微博互动数据是单维视角。真正有价值的是将uid(如1234567890)映射到用户实体,获取其昵称、所在地、简介、关注数、粉丝数、认证信息等,从而构建“谁在转发谁”的传播图谱。微博提供https://weibo.com/ajax/profile/info?uid={uid}接口,无需额外鉴权,但需注意字段稳定性。
5.1 UID 基础信息接口字段可靠性排序(按 2024 年实测)
| 字段名 | 是否稳定 | 说明 | 替代方案(若缺失) |
|---|---|---|---|
user.screen_name | ★★★★★ | 昵称,100% 存在 | — |
user.description | ★★★★☆ | 简介,95% 存在 | 从用户主页 HTML 提取 |
user.followers_count | ★★★★☆ | 粉丝数,90% 存在 | 用https://weibo.com/ajax/friendships/friends/active?uid={uid}补充 |
user.location | ★★☆☆☆ | 所在地,仅 40% 用户填写 | 用 NLP 从简介中抽城市名 |
user.verified_type | ★★★★★ | 认证类型(-1=未认证,0=个人,2=企业,3=政府) | — |
user.avatar_hd | ★★★★☆ | 高清头像 URL | — |
5.2 批量 UID 查询的并发控制与缓存策略
为避免频繁请求触发限流,必须实现两级缓存:
- 内存缓存:用
functools.lru_cache(maxsize=1000)缓存最近查询的 UID 结果; - 磁盘缓存:用 SQLite 存储
uid → profile_json,设置 TTL 7 天。
import sqlite3 from functools import lru_cache DB_CACHE = "uid_profile_cache.db" def init_cache_db(): conn = sqlite3.connect(DB_CACHE) conn.execute(""" CREATE TABLE IF NOT EXISTS uid_profile ( uid TEXT PRIMARY KEY, profile_json TEXT NOT NULL, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.execute("CREATE INDEX IF NOT EXISTS idx_updated ON uid_profile(updated_at)") conn.commit() conn.close() @lru_cache(maxsize=1000) def get_user_profile_cached(uid): # 先查内存缓存(lru_cache) # 再查磁盘缓存(SQLite) conn = sqlite3.connect(DB_CACHE) cursor = conn.execute("SELECT profile_json FROM uid_profile WHERE uid = ? AND updated_at > datetime('now', '-7 days')", (uid,)) row = cursor.fetchone() conn.close() if row: return json.loads(row[0]) # 调用 API resp = session.get(f"https://weibo.com/ajax/profile/info?uid={uid}") if resp.status_code == 200 and resp.json().get("ok") == 1: profile = resp.json()["data"] # 写入磁盘缓存 conn = sqlite3.connect(DB_CACHE) conn.execute("INSERT OR REPLACE INTO uid_profile (uid, profile_json) VALUES (?, ?)", (uid, json.dumps(profile, ensure_ascii=False))) conn.commit() conn.close() return profile return None5.3 构建传播关系 CSV:从互动数据到有向图边集
最终输出的interaction_edges.csv应包含以下字段,可直接导入 Gephi 或 NetworkX 做社区发现:
| 字段 | 类型 | 说明 |
|---|---|---|
source_uid | str | 转发者/评论者 UID |
target_uid | str | 原博主 UID(从微博user.id获取) |
interaction_type | str | repost/comment/like |
timestamp | datetime | 互动发生时间 |
weight | int | 固定为 1(或按互动深度加权) |
生成逻辑示例:
def build_edge_list(bid, reposts, comments, original_user_id): edges = [] # 转发边 for r in reposts: edges.append({ "source_uid": r["uid"], "target_uid": original_user_id, "interaction_type": "repost", "timestamp": r["created_at"], "weight": 1 }) # 评论边 for c in comments: edges.append({ "source_uid": c["uid"], "target_uid": original_user_id, "interaction_type": "comment", "timestamp": c["created_at"], "weight": 1 }) return edges # 调用示例 original_user_id = "2803301701" # 从 show 接口获取 edges = build_edge_list("KxYzAbC12", reposts, comments, original_user_id) save_to_csv("output/interaction_edges.csv", edges, fieldnames=["source_uid","target_uid","interaction_type","timestamp","weight"])这一张 CSV,就是微博传播网络的原子级表达——它不依赖任何第三方平台,完全由你掌控数据主权,且可无限扩展至百万级 UID 关系分析。
本文还有配套的精品资源,点击获取