资讯动态

生产级微博爬虫:Requests+Selenium混合架构实战

发布时间:2026/10/25 11:57:01 来源:尧图企业网站定制
简介这是一份面向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 方案快 35 倍比纯 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 请求调度策略如何让爬虫既快又不被封关键不在并发数而在「请求节奏控制」。项目采用三级节流域名级限频对weibo.com主域设置min_interval1.2s防触发服务端 QPS 熔断接口级权重搜索接口权重为 1用户主页为 2评论接口为 3因后者更重需更多资源动态退避当 HTTP 状态码为 418Im 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.md3.1settings.yaml配置详解改这 7 个字段就能跑通这是项目唯一需要人工编辑的文件。以下是关键字段说明其他字段保持默认即可字段类型必填示例值说明base_urlstring是https://weibo.com微博主站地址勿加/search_keywordslist是[人工智能, 大模型]搜索关键词列表支持中文、英文、话题#start_datestring是2024-01-01搜索起始日期格式 YYYY-MM-DDend_datestring是2024-01-31搜索截止日期含max_pages_per_keywordint否50每个关键词最多抓取页数默认 100output_formatstring否jsonl可选jsonl,sqlite,csvenable_headlessbool否true是否启用无头模式false 用于调试滑块注意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:3308: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, linesTrue) 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: fcustom:{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, paramsparams, headersheaders, timeout15) 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 [] # 定位所有微博卡片找 classcard-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[classWB_cardwrap WB_feed_type S_bg2]/data-sudaref) if not real_id: raise ValueError(fNo>self.session requests.Session() adapter requests.adapters.HTTPAdapter( pool_connections10, pool_maxsize20, max_retries3 ) self.session.mount(http://, adapter) self.session.mount(https://, adapter)5.4 现象JSONL 输出文件中出现乱码如\u5927\u6a21\u578b原因json.dump()默认ensure_asciiTrue将中文转为 Unicode 转义。虽然合法但不利于人工查看和grep检索。解决修改storage/jsonl_writer.py中的write_batch()方法显式传入ensure_asciiFalsefor item in batch: json_line json.dumps(item, ensure_asciiFalse) \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, jsondata, timeout5) # 在 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?keyxxxxx6.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(fInvalid 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%。希望帮到你。本文还有配套的精品资源点击获取

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑