简介：这是一份面向Python初学者与数据采集实践者的新浪微博爬虫项目实战资源，聚焦社交媒体数据获取与基础网络爬虫开发，适用于用户行为分析、舆情监测、课程设计等场景。资源共77个文件，包含44个核心Python脚本（如spider.py、parser.py、user.py）、14篇Markdown文档（涵盖cookie配置、自动化流程、学术使用说明等）、11个HTML示例页及配置类文件（yml/json/conf），整体仅106KB，轻量易读，结构清晰——主模块weibo_spider下分层组织下载器、解析器、日志与配置，配套README、FAQ与贡献指南完善。已有1190人学习下载，提供从requests请求发送、BeautifulSoup网页解析、反爬应对（headers/代理基础设置）到数据存储（CSV/JSON）的完整链路代码与注释，附带测试用例、用户ID列表模板及可直接运行的__main__.py入口，助读者快速上手并理解爬虫工程化组织逻辑。

1. 微博数据采集不是“爬虫玩具”，而是真实业务中高频复用的数据基建模块

某高校实验室做舆情分析时，需要连续三个月抓取某类话题下微博的原始发帖、转发链、评论情感倾向和用户画像标签——他们试过现成的 API 工具，但调用量受限、字段残缺、无法获取已删除内容；也试过浏览器自动化方案，结果跑两天就因验证码拦截全军覆没。最后落地的，是一个基于 Requests + Selenium 混合调度、带登录态持久化、支持断点续爬、能自动识别滑块验证的 Python 爬虫工程。它不炫技，不刷量，不碰敏感边界，只专注把「一条微博从网页 DOM 到本地 JSONL 文件」这条链路跑稳、跑全、跑可维护。如果你要的是：能进生产环境跑半年不崩的日志体系、能按关键词/时间/地域/用户分层采集的策略配置、能导出结构化数据供 Pandas 直接分析的输出格式——那这个项目就是你该下载并跑起来的第一个真实可用样本。它不是教学 Demo，是某开发者在三个实际项目中反复迭代出的最小可行数据采集基座。

2. 技术栈与架构设计：为什么选 Requests + Selenium 而非纯 Selenium 或 Scrapy？

这个项目没有堆砌框架，所有技术选型都来自真实踩坑后的收敛。核心逻辑是：Requests 负责绝大多数静态接口请求（快、可控、易调试），Selenium 只在必须渲染的场景兜底（登录、滑块验证、动态加载页）。这比纯 Selenium 方案快 3～5 倍，比纯 Requests 方案稳定 10 倍以上——后者在微博反爬升级后几乎必然失效。

2.1 模块分层：从入口到落盘的五层职责划分

整个工程按数据流向划分为五个物理模块，全部位于src/目录下：

• core/：核心调度器，含CrawlerManager类，负责任务分发、线程池管理、异常重试策略（指数退避）、全局 session 复用；
• auth/：登录认证模块，封装了扫码登录、Cookie 注入、Token 自动续期逻辑，关键在于WeiboAuthenticator类中对login.sina.com.cn和passport.weibo.com的双域 Cookie 同步处理；
• fetcher/：数据获取层，包含SearchFetcher（关键词搜索）、UserFetcher（用户主页）、CommentFetcher（评论树）三个子类，每个类都实现fetch_page()和parse_page()两个抽象方法；
• parser/：解析层，全部使用lxml.etree+ XPath，拒绝正则匹配 HTML（玄学匹配在微博 DOM 结构频繁变动时必翻车），所有 XPath 表达式都带 fallback 机制；
• storage/：存储层，支持 JSONL（默认）、SQLite、CSV 三种后端，JsonlWriter类内置字段标准化逻辑（如时间转 ISO8601、URL 去重归一化、文本清洗去广告语）。

提示：storage/模块设计为插件式，新增 MySQL 支持只需继承BaseWriter并实现write_batch()方法，无需改动任何上层逻辑。

2.2 请求调度策略：如何让爬虫既快又不被封？

关键不在并发数，而在「请求节奏控制」。项目采用三级节流：

1. 域名级限频：对weibo.com主域设置min_interval=1.2s（防触发服务端 QPS 熔断）；
2. 接口级权重：搜索接口权重为 1，用户主页为 2，评论接口为 3（因后者更重，需更多资源）；
3. 动态退避：当 HTTP 状态码为 418（I'm a teapot，微博反爬常用伪装码）或响应体含"请稍后重试"字样时，当前任务暂停random.uniform(30, 90)秒，并记录到logs/throttle.log。

# src/core/scheduler.py class RequestScheduler: def __init__(self): self.domain_limits = {"weibo.com": 1.2} self.interface_weights = {"search": 1, "user": 2, "comment": 3} self.last_request_time = defaultdict(float) def can_proceed(self, domain: str, interface: str) -> bool: now = time.time() min_next = self.last_request_time[domain] + self.domain_limits[domain] * self.interface_weights[interface] if now < min_next: sleep_time = min_next - now time.sleep(sleep_time) self.last_request_time[domain] = time.time() return True

这段代码看似简单，却是整个项目能长期存活的核心——它不追求极限速度，而用确定性节流换取稳定性。我一般会把domain_limits值写死在配置文件里，上线前用ab工具压测确认阈值，而不是凭经验硬写。

2.3 登录态持久化：为什么不用“每次启动都扫码”？

扫码登录只在首次运行时触发，后续全部依赖 Cookie + Token 双保险。auth/模块会在成功登录后执行：

• 将SUB,SUBP,ALF,SSOLoginState四个关键 Cookie 写入config/auth_cookies.json；
• 同时调用https://login.sina.com.cn/sso/prelogin.php获取 RSA 公钥，加密生成servertime,nonce,rsakv参数，存入config/login_params.json；
• 每次请求前，CrawlerManager会自动注入 Cookie，并在 Header 中添加X-Requested-With: XMLHttpRequest和Referer: https://weibo.com/—— 缺一不可，否则返回 403。

这种设计让爬虫可以 7×24 小时运行，且支持多实例共享同一套登录态（只要auth_cookies.json文件不被覆盖）。某次线上事故中，一个实例因网络中断掉线，另一实例立刻接管任务，零数据丢失。

3. 目录结构与核心配置：5 分钟看懂项目骨架

项目解压后共 12 个文件/目录，结构极简，无冗余：

weibo-crawler/ ├── config/ │ ├── auth_cookies.json # 登录态 Cookie（首次运行后生成） │ ├── login_params.json # RSA 加密参数缓存 │ └── settings.yaml # 全局配置（见下表） ├── logs/ # 运行日志（按天滚动） ├── output/ # 数据输出目录（JSONL 默认在此） ├── src/ │ ├── __init__.py │ ├── core/ # 调度主逻辑 │ ├── auth/ # 认证相关 │ ├── fetcher/ # 各类数据抓取器 │ ├── parser/ # 解析规则 │ └── storage/ # 存储后端 ├── requirements.txt ├── run_crawler.py # 入口脚本（可直接 python run_crawler.py） └── README.md

3.1settings.yaml配置详解：改这 7 个字段就能跑通

这是项目唯一需要人工编辑的文件。以下是关键字段说明（其他字段保持默认即可）：

注意：start_date和end_date控制的是微博搜索页的「时间筛选器」，不是程序运行时间。微博前端只允许最大跨度 30 天，所以若需抓取三个月数据，需在配置中拆成三个任务分别运行。

3.2run_crawler.py入口逻辑：三行代码启动全流程

这个文件只有 23 行，却完整串联了初始化、认证、采集、落盘四步：

# run_crawler.py from src.core.crawler_manager import CrawlerManager from src.auth.weibo_authenticator import WeiboAuthenticator from config.settings import load_settings if __name__ == "__main__": settings = load_settings() # 1. 加载 YAML 配置 auth = WeiboAuthenticator(settings) # 2. 初始化认证器（自动检查 Cookie 有效性） crawler = CrawlerManager(settings, auth) # 3. 构建爬虫管理器 crawler.start() # 4. 启动主循环：关键词 → 搜索页 → 微博列表 → 详情页 → 评论页 → 存储

crawler.start()内部执行的是一个状态机：先调用SearchFetcher.fetch()获取微博 ID 列表，再并发调用UserFetcher.fetch()和CommentFetcher.fetch()补全元数据，最后统一交由JsonlWriter.write_batch()落盘。整个流程不依赖外部消息队列，靠内存队列 + 线程安全锁控制并发。

3.3output/目录下的数据结构：开箱即用的分析友好格式

默认输出为 JSONL（每行一个 JSON 对象），文件名按search_{keyword}_{date}.jsonl命名。每条记录结构如下（已脱敏）：

{ "id": "KxYzAbC123", "text": "今天试用了新发布的大模型，效果惊艳！#AI#", "created_at": "2024-01-15T14:22:33+08:00", "user_id": "U789def456", "user_nickname": "科技观察员", "reposts_count": 12, "comments_count": 47, "attitudes_count": 213, "source": "iPhone客户端", "pic_urls": ["https://wx1.sinaimg.cn/large/abc123.jpg"], "video_url": null, "is_retweet": false, "retweet_id": null, "retweet_user": null }

所有时间字段均为 ISO8601 标准格式，数字字段均为整型（非字符串），布尔字段明确为true/false。这意味着你可以直接用 Pandas 加载：

import pandas as pd df = pd.read_json("output/search_人工智能_2024-01-01.jsonl", lines=True) print(df[["text", "created_at", "reposts_count"]].head())

不需要任何预处理，开箱即用。这也是为什么某导师让学生用这个项目做课程设计——数据质量高，分析门槛低。

4. 核心代码逐行解析：从搜索页抓取到微博正文解析的完整链路

我们以最常用的「关键词搜索」功能为例，拆解从发出请求到解析出 20 条微博的全过程。关键路径在src/fetcher/search_fetcher.py和src/parser/search_parser.py。

4.1SearchFetcher.fetch_page()：如何构造合法的搜索请求？

微博搜索接口是 POST 请求，地址为https://weibo.com/ajax/side/hotSearch（热榜）和https://weibo.com/ajax/search/all（综合搜索），但本项目用的是更稳定的https://s.weibo.com/weibo—— 它是前端真实调用的地址，参数明文可见。

# src/fetcher/search_fetcher.py def fetch_page(self, keyword: str, page: int) -> str: params = { "q": keyword, "typeall": "1", "suball": "1", "timescope": f"custom:{self.start_date}:{self.end_date}", "page": str(page), "Refer": "https://s.weibo.com/", "tw": "1" } headers = { "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36", "Cookie": self.auth.get_cookie_string(), # 注入有效 Cookie "X-Requested-With": "XMLHttpRequest" } url = "https://s.weibo.com/weibo" response = self.session.get(url, params=params, headers=headers, timeout=15) response.raise_for_status() return response.text

注意三个关键点：

• timescope参数必须是custom:YYYY-MM-DD:YYYY-MM-DD格式，不能用week或month（微博后端已废弃）；
• ReferHeader 必须存在且指向s.weibo.com，否则返回空数据；
• self.session是requests.Session()实例，已预置了 Cookie 和连接池，避免重复握手开销。

4.2SearchParser.parse_search_page()：XPath 解析的健壮写法

微博搜索页 DOM 结构复杂，且经常插入广告位、推荐位干扰。本项目采用「锚点定位 + 相对路径」策略，而非绝对路径：

# src/parser/search_parser.py def parse_search_page(self, html: str) -> List[Dict]: tree = etree.HTML(html) items = [] # 定位所有微博卡片：找 class="card-wrap" 且># src/fetcher/comment_fetcher.py def fetch_comments(self, weibo_id: str, weibo_html: str) -> List[Dict]: # 从详情页 HTML 中提取真实 ID tree = etree.HTML(weibo_html) real_id = tree.xpath('//div[@class="WB_cardwrap WB_feed_type S_bg2"]/@data-sudaref') if not real_id: raise ValueError(f"No>self.session = requests.Session() adapter = requests.adapters.HTTPAdapter( pool_connections=10, pool_maxsize=20, max_retries=3 ) self.session.mount("http://", adapter) self.session.mount("https://", adapter)

5.4 现象：JSONL 输出文件中出现乱码（如\u5927\u6a21\u578b）

原因：json.dump()默认ensure_ascii=True，将中文转为 Unicode 转义。虽然合法，但不利于人工查看和grep检索。
解决：修改storage/jsonl_writer.py中的write_batch()方法，显式传入ensure_ascii=False：

for item in batch: json_line = json.dumps(item, ensure_ascii=False) + "\n" f.write(json_line)

5.5 现象：多关键词配置下，部分关键词数据量极少甚至为 0

原因：微博搜索接口对冷门关键词返回结果数有限制（如低于 10 条时不再分页），max_pages_per_keyword设置过大反而导致无效请求堆积。
解决：在SearchFetcher.fetch_all_pages()中增加动态页数控制：每页返回结果数 < 5 时，立即终止该关键词的后续请求，并记录logs/warning.log。此优化使整体采集效率提升 40%。

6. 二次开发实战：如何快速接入企业微信通知与数据质量校验

当你把基础采集跑通后，下一步往往是「让数据真正可用」。我最近在一个客户项目中，基于本项目快速扩展了两个刚需功能：采集完成自动推送企业微信消息和对输出数据做字段完整性校验。下面直接给出可复制的代码和配置。

6.1 接入企业微信机器人：三步完成告警推送

企业微信机器人 Webhook 地址可在「工作台 → 应用管理 → 创建自定义机器人」中获取。本项目通过storage/notify.py模块封装：

# src/storage/notify.py import requests import json from config.settings import load_settings class WeComNotifier: def __init__(self): self.settings = load_settings() self.webhook = self.settings.get("wecom_webhook", "") def send_alert(self, title: str, content: str): if not self.webhook: return data = { "msgtype": "markdown", "markdown": { "content": f"## {title}\n> {content}\n\n- 采集时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}\n- 数据量：{len(content.split('条'))} 条" } } requests.post(self.webhook, json=data, timeout=5) # 在 crawler_manager.py 的 finish() 方法末尾调用： # notifier = WeComNotifier() # notifier.send_alert("微博采集完成", f"关键词「{keyword}」共采集 1247 条")

只需在settings.yaml中添加：

wecom_webhook: "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxx"

6.2 数据质量校验：用 Pydantic 定义微博 Schema 并自动过滤脏数据

原始爬取数据常有字段缺失（如user_id为空）、类型错误（如reposts_count是字符串"12"而非整数）。我们在parser/validator.py中定义强约束 Schema：

# src/parser/validator.py from pydantic import BaseModel, validator from typing import Optional, List class WeiboItem(BaseModel): id: str text: str created_at: str user_id: str user_nickname: str reposts_count: int comments_count: int attitudes_count: int @validator("text") def text_must_not_be_empty(cls, v): if not v.strip(): raise ValueError("text cannot be empty") return v.strip() @validator("created_at") def created_at_must_be_iso(cls, v): try: datetime.fromisoformat(v.replace("Z", "+00:00")) except ValueError: raise ValueError("created_at must be ISO8601 format") return v # 在 JsonlWriter.write_batch() 中插入校验： valid_items = [] for item in batch: try: valid_item = WeiboItem(**item) valid_items.append(valid_item.dict()) except Exception as e: logger.warning(f"Invalid item dropped: {item.get('id', 'unknown')} - {e}")

这样，所有不符合 Schema 的数据都会被静默丢弃，并记录警告日志。某次客户数据验收时，我们靠这个校验发现了 3.2% 的user_id字段为空，及时回溯到SearchParser中修复了 XPath 表达式。

6.3 从那以后我每次新增一个采集目标，都强制走一遍「三验流程」

• 一验字段：用WeiboItemSchema 校验输出，确保所有字段类型、非空、格式合规；
• 二验时间：检查created_at是否在start_date和end_date范围内，防止微博前端时间筛选失效；
• 三验去重：对output/下所有 JSONL 文件，用awk -F'"' '{print $4}' | sort | uniq -d快速检查 ID 重复率，超过 0.1% 就触发重爬。

这套流程让我在交付 7 个不同客户的微博采集需求时，数据验收一次通过率 100%。希望帮到你。

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