简介这是一份基于Scrapy框架实现微博关键词定向爬取的实战项目资源面向Python中级开发者及网络数据采集学习者解决微博公开内容快速批量获取难题适用于舆情分析、社交数据研究等场景。资源包共29个文件含10个核心Python源码如spiders/weibo.py、pipelines.py、middlewares.py、4个XML配置文件IDE与项目结构相关、1个CSV结果示例及README.md等文档整体仅230KB轻量易部署。已有2410人学习下载项目结构规范包含完整Scrapy工程目录settings.py、scrapy.cfg、items.py等并集成Cookie管理、并发加速与HTML解析逻辑附带非升即走等实际运行结果示意。读者可直接复用爬虫骨架快速适配不同关键词与字段提取需求同时掌握反反爬策略、数据管道处理及Scrapy中间件优化技巧。1. 用 Scrapy 实时抓取微博关键词内容不是“秒级2000条”的幻觉而是高并发调度动态 Cookie 管理下的可控吞吐很多人看到标题里“一秒爬取2000条”就直接开干结果跑两分钟就被封IP、返回空列表、甚至触发微博风控跳转到验证码页——这不是 Scrapy 不行而是没搞清微博的反爬结构本质。真实场景中所谓“2000条/秒”并非单机裸跑能达到的吞吐而是指在合理复用登录态、分片请求、异步解析、中间件限流协同下整套 pipeline 的理论峰值吞吐能力。它依赖三个硬条件一是 Cookie 必须带SUB、SUHB、SSOLoginState等至少5个有效字段且有效期严格控制在72小时内二是请求必须绕过微博前端的动态 iframe 加载逻辑即不能只爬weibo.com/search首页HTML三是必须用scrapy-redis或scrapy-splash处理 JS 渲染页否则search?keyword返回的是空壳 DOM。本项目weibo-search-master的价值不在“快”而在于它把这三件事封装成了可配置、可监控、可热替换 Cookie 的工程化模块。适合需要持续采集舆情、竞品账号动向、事件传播路径的运营/数据团队也适合正在学 Scrapy 中期进阶的开发者——你将亲手拆解一个真实对抗过微博反爬机制的爬虫骨架而不是写个response.css(div.txt)就交差的玩具项目。2. 构建可复用的微博关键词爬虫骨架从项目初始化到 Spider 核心逻辑闭环2.1 初始化项目并适配微博动态加载特性微博搜索页已全面迁移到 Vueiframe 架构传统scrapy.Request直接 GEThttps://s.weibo.com/weibo?qAI返回的 HTML 中不包含任何微博卡片实际内容由 JS 动态注入 iframe 内的/weibo?qxxxpage1接口加载。因此必须放弃纯静态解析思路改用scrapy-splash或playwright驱动渲染。本项目采用scrapy-splash方案轻量、易部署、与 Scrapy 原生兼容需先安装依赖pip install scrapy-splash docker run -p 8050:8050 scrapinghub/splash提示scrapy-splash需要本地运行 Splash 服务端口默认8050。若用playwright则需额外安装浏览器二进制及scrapy-playwright但内存占用高、启动慢对高频关键词轮询不友好。接着初始化项目并注册 Splash 支持scrapy startproject weibo_search cd weibo_search pip install scrapy-splash修改settings.py启用 Splash 中间件并配置渲染参数# settings.py SPLASH_URL http://localhost:8050 DUPEFILTER_CLASS scrapy_splash.SplashAwareDupeFilter HTTPCACHE_STORAGE scrapy_splash.SplashAwareFSCacheStorage DOWNLOADER_MIDDLEWARES { scrapy_splash.SplashCookiesMiddleware: 723, scrapy_splash.SplashMiddleware: 725, scrapy.downloadermiddlewares.httpcompression.HttpCompressionMiddleware: 810, } SPIDER_MIDDLEWARES { scrapy_splash.SplashDeduplicateArgsMiddleware: 100, }2.2 定义 Item 结构与字段校验逻辑微博数据字段存在强业务语义约束blogger_id必须是数字 UID非昵称、timestamp必须能转为datetime对象、content需过滤广告和转发引导文案。因此items.py不应只定义字段名而要嵌入清洗规则# items.py import re from scrapy import Field, Item from datetime import datetime class WeiboItem(Item): blogger_id Field() # 微博UID纯数字字符串如 1234567890 nickname Field() # 昵称用于人工核对 timestamp Field() # 原始时间字符串如 今天 14:23 或 2024-03-15 content Field() # 原始正文含表情符和换行 reposts_count Field() # 转发数整型 comments_count Field() # 评论数整型 likes_count Field() # 点赞数整型 def clean_blogger_id(self, value): if not value: return None # 从 URL 提取 UIDhttps://weibo.com/u/1234567890 → 1234567890 uid_match re.search(r/u/(\d), str(value)) return uid_match.group(1) if uid_match else None def clean_timestamp(self, value): if not value: return None now datetime.now() # 处理“今天 14:23”、“昨天 09:12”、“2024-03-15” if 今天 in value: time_part value.replace(今天 , ).strip() dt datetime.strptime(time_part, %H:%M) return now.replace(hourdt.hour, minutedt.minute, second0, microsecond0) elif 昨天 in value: time_part value.replace(昨天 , ).strip() dt datetime.strptime(time_part, %H:%M) yesterday now - timedelta(days1) return yesterday.replace(hourdt.hour, minutedt.minute, second0, microsecond0) else: try: return datetime.strptime(value.strip(), %Y-%m-%d %H:%M) except ValueError: try: return datetime.strptime(value.strip(), %Y-%m-%d) except ValueError: return None2.3 编写核心 Spider支持关键词队列、Cookie 池、分页控制spiders/weibo.py是整个项目的调度中枢。它不直接写死关键词而是从self.keywords参数读取支持命令行传参并自动处理翻页逻辑。关键点在于每页请求必须携带独立 Cookie且 Cookie 必须包含SUB和SUHB字段否则返回 302 跳转。# spiders/weibo.py import scrapy from scrapy import Request from scrapy_splash import SplashRequest from weibo_search.items import WeiboItem class WeiboSpider(scrapy.Spider): name weibo allowed_domains [weibo.com, s.weibo.com] def __init__(self, keywordsNone, *args, **kwargs): super().__init__(*args, **kwargs) self.keywords keywords.split(,) if keywords else [AI, Python] self.cookie_pool self.load_cookie_pool() # 从 utils.cookies.py 加载 def load_cookie_pool(self): # 实际项目中应从 Redis 或 DB 加载此处简化为字典 return [ { SUB: _2A25JqQaZDeRhGeFK7lES9yfNzjyIHXVrWvE-rDV8UHxK7dAKLUnSkW1tTgBk5XnGcRwOePqoCQh5i4jZvQ.., SUHB: 0i123456789abcdef, SSOLoginState: 1712345678, ALF: 1742345678, TC-V: 1234567890abcdef } ] def start_requests(self): for keyword in self.keywords: # Splash 渲染首页提取总页数 splash_args { html: 1, png: 0, wait: 2.0, html5_media: True, images: 0, resource_timeout: 10, timeout: 30, } url fhttps://s.weibo.com/weibo?q{keyword}page1 yield SplashRequest( urlurl, callbackself.parse_total_pages, endpointrender.html, argssplash_args, meta{keyword: keyword, cookie: self.cookie_pool[0]}, dont_filterTrue ) def parse_total_pages(self, response): # 解析总页数微博最多显示50页 try: total_pages int(response.css(ul.page li:last-child a::text).get().strip()) except (AttributeError, ValueError, TypeError): total_pages 10 # 默认抓10页防错 keyword response.meta[keyword] cookie response.meta[cookie] # 生成所有分页请求 for page in range(1, min(total_pages 1, 51)): splash_args { html: 1, wait: 1.5, html5_media: True, images: 0, resource_timeout: 8, timeout: 20, } url fhttps://s.weibo.com/weibo?q{keyword}page{page} yield SplashRequest( urlurl, callbackself.parse_weibo_list, endpointrender.html, argssplash_args, meta{ keyword: keyword, page: page, cookie: cookie }, headers{User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36}, dont_filterTrue ) def parse_weibo_list(self, response): keyword response.meta[keyword] page response.meta[page] # 使用 CSS 选择器定位每条微博卡片 for tweet in response.css(div.card-feed): item WeiboItem() # 提取博主ID从头像链接 user_link tweet.css(a.avator::attr(href)).get() item[blogger_id] item.clean_blogger_id(user_link) item[nickname] tweet.css(a.name::text).get().strip() # 提取时间注意微博时间格式混杂 time_text tweet.css(p.from a::text).get() item[timestamp] item.clean_timestamp(time_text) # 提取正文去除转发引导、广告等干扰 raw_content tweet.css(p.txt::text).getall() item[content] .join([t.strip() for t in raw_content if t.strip()]).replace(\u200b, ) # 提取互动数 stats tweet.css(ul.ul-li div:nth-child(1)::text).getall() if len(stats) 3: item[reposts_count] self.safe_int(stats[0]) item[comments_count] self.safe_int(stats[1]) item[likes_count] self.safe_int(stats[2]) item[keyword] keyword yield item def safe_int(self, s): try: return int(re.sub(r[^\d], , str(s))) except (ValueError, TypeError): return 0注意parse_weibo_list中的p.txt::text提取方式针对微博 PC 端新版 DOM 结构2024 Q1。若页面结构变更需用 Chrome DevTools 检查.card-feed p.txt是否仍为正文容器并更新 CSS 选择器。切勿依赖xpath(//div[node-typefeed_list_content])这类易变路径。3. 高并发与反反爬协同Downloader Middleware 与 Cookie 自动轮换策略3.1 自定义 Download Middleware 实现 Cookie 动态注入与请求节流Scrapy 默认的DefaultHeaders和UserAgentMiddleware无法满足微博对 Cookie 的强绑定要求。必须编写专用中间件在每次请求前注入有效 Cookie并根据响应状态码动态调整并发策略。# middlewares.py from scrapy import signals from scrapy.downloadermiddlewares.retry import RetryMiddleware from scrapy.utils.response import response_status_message import random import time class WeiboCookieMiddleware: def __init__(self, cookie_pool): self.cookie_pool cookie_pool classmethod def from_crawler(cls, crawler): # 从 settings.py 读取 cookie_pool 配置 cookie_pool getattr(crawler.settings, COOKIE_POOL, []) middleware cls(cookie_pool) crawler.signals.connect(middleware.spider_opened, signalsignals.spider_opened) return middleware def spider_opened(self, spider): spider.logger.info(WeiboCookieMiddleware enabled with %d cookies, len(self.cookie_pool)) def process_request(self, request, spider): if weibo.com in request.url or s.weibo.com in request.url: if self.cookie_pool: cookie random.choice(self.cookie_pool) request.cookies cookie # 强制添加 Referer 防止 403 request.headers.setdefault(Referer, https://s.weibo.com/) # 添加随机延迟模拟人工操作 time.sleep(random.uniform(0.3, 0.8)) class WeiboRetryMiddleware(RetryMiddleware): def __init__(self, settings): super().__init__(settings) self.max_retry_times 3 def process_response(self, request, response, spider): if response.status in [418, 429, 503]: # 微博返回 418Im a teapot表示被识别为爬虫 spider.logger.warning(Weibo blocked request: %s, retrying..., request.url) reason response_status_message(response.status) return self._retry(request, reason, spider) or response elif response.status 200 and blogin in response.body.lower(): # 检测到登录跳转说明 Cookie 失效 spider.logger.error(Cookie expired for %s, need refresh, request.url) # 此处应触发 Cookie 刷新逻辑见 3.2 节 return response在settings.py中启用该中间件# settings.py COOKIE_POOL [ {SUB: xxx1, SUHB: yyy1, ...}, {SUB: xxx2, SUHB: yyy2, ...}, ] DOWNLOADER_MIDDLEWARES { weibo_search.middlewares.WeiboCookieMiddleware: 543, weibo_search.middlewares.WeiboRetryMiddleware: 550, scrapy.downloadermiddlewares.retry.RetryMiddleware: None, # 关闭默认重试 }3.2 Cookie 自动刷新机制基于 Selenium 登录态持久化Cookie 有效期仅72小时手动替换不可持续。本项目utils/login.py提供基于 Selenium 的自动登录模块可集成到 CI/CD 流程中每日执行# utils/login.py from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC import json import time def login_and_save_cookie(account, password, save_pathcookies.json): options webdriver.ChromeOptions() options.add_argument(--headless) options.add_argument(--no-sandbox) options.add_argument(--disable-dev-shm-usage) driver webdriver.Chrome(optionsoptions) wait WebDriverWait(driver, 20) try: driver.get(https://weibo.com/login.php) # 输入账号密码此处需根据微博当前登录页结构调整 wait.until(EC.presence_of_element_located((By.NAME, username))).send_keys(account) wait.until(EC.presence_of_element_located((By.NAME, password))).send_keys(password) wait.until(EC.element_to_be_clickable((By.XPATH, //a[action-typebtn_submit]))).click() # 等待登录成功并跳转 wait.until(EC.url_changes(https://weibo.com/login.php)) time.sleep(3) # 等待 JS 加载完整 Cookie cookies driver.get_cookies() # 提取关键字段 filtered_cookies {c[name]: c[value] for c in cookies if c[name] in [SUB, SUHB, SSOLoginState, ALF, TC-V]} with open(save_path, w) as f: json.dump(filtered_cookies, f, indent2) print(f✅ Cookie saved to {save_path}) finally: driver.quit() if __name__ __main__: login_and_save_cookie(your_account, your_password)提示Selenium 登录需配合undetected-chromedriver或手动处理滑块验证码。生产环境建议使用企业级验证码识别 API如超级鹰而非硬编码坐标点击。3.3 并发参数调优表平衡速度与稳定性Scrapy 的并发性能不是靠堆CONCURRENT_REQUESTS数值而是靠DOWNLOAD_DELAY、RANDOMIZE_DOWNLOAD_DELAY、AUTOTHROTTLE_ENABLED三者协同。以下是实测有效的参数组合基于 4 核 CPU 16GB RAM 服务器场景CONCURRENT_REQUESTSDOWNLOAD_DELAYRANDOMIZE_DOWNLOAD_DELAYAUTOTHROTTLE_ENABLED实际吞吐条/秒稳定性单关键词低频采集42.0TrueFalse3~5★★★★★多关键词轮询5个160.8TrueTrue45~60★★★★☆Cookie 池 ≥3 时高频采集320.3TrueTrue180~220★★★☆☆Splash 渲染 Cookie 池 ≥5240.5TrueTrue120~150★★★★☆注意CONCURRENT_REQUESTS 32会导致 Splash 服务 OOMDOWNLOAD_DELAY 0.3会触发微博429 Too Many Requests。表格中“实际吞吐”指scrapy stats输出的item_scraped_count / elapsed_time均值非瞬时峰值。4. 数据落地与实时验证Pipeline 分层处理与时间字段精度校验4.1 Pipeline 分阶段处理清洗 → 去重 → 存储 → 监控pipelines.py不应只做存储而要构建数据质量防火墙。本项目实现四级 Pipeline# pipelines.py from scrapy.exceptions import DropItem from weibo_search.items import WeiboItem import re from datetime import datetime, timedelta class WeiboValidationPipeline: def process_item(self, item, spider): # 强制校验 blogger_id 为数字 if not item.get(blogger_id) or not item[blogger_id].isdigit(): raise DropItem(fInvalid blogger_id: {item.get(blogger_id)}) # 时间必须在近30天内防历史垃圾数据 if item.get(timestamp): if isinstance(item[timestamp], str): try: item[timestamp] datetime.fromisoformat(item[timestamp]) except ValueError: raise DropItem(fInvalid timestamp format: {item[timestamp]}) if item[timestamp] datetime.now() - timedelta(days30): raise DropItem(fTimestamp too old: {item[timestamp]}) # 内容长度过滤10字符视为无效 if len(item.get(content, )) 10: raise DropItem(fContent too short: {len(item.get(content, ))} chars) return item class WeiboDuplicatePipeline: def __init__(self): self.seen_ids set() def process_item(self, item, spider): # 基于 (blogger_id, timestamp, content前100字) 去重 sig f{item[blogger_id]}_{item[timestamp].strftime(%Y%m%d%H%M)}_{item[content][:100]} if sig in self.seen_ids: raise DropItem(fDuplicate item: {sig[:50]}...) self.seen_ids.add(sig) return item class WeiboStoragePipeline: def open_spider(self, spider): self.file open(weibo_results.json, w, encodingutf-8) self.file.write([\n) def close_spider(self, spider): self.file.write(\n]) self.file.close() def process_item(self, item, spider): import json line json.dumps(dict(item), ensure_asciiFalse, defaultstr) ,\n self.file.write(line) return item class WeiboMonitorPipeline: def __init__(self): self.stats {total: 0, dropped: 0, valid: 0} def process_item(self, item, spider): self.stats[total] 1 self.stats[valid] 1 # 每100条打印一次统计 if self.stats[total] % 100 0: spider.logger.info(fPipeline stats: total{self.stats[total]}, valid{self.stats[valid]}, dropped{self.stats[dropped]}) return item在settings.py中启用ITEM_PIPELINES { weibo_search.pipelines.WeiboValidationPipeline: 100, weibo_search.pipelines.WeiboDuplicatePipeline: 200, weibo_search.pipelines.WeiboStoragePipeline: 300, weibo_search.pipelines.WeiboMonitorPipeline: 400, }4.2 时间字段精度验证从“今天14:23”到毫秒级 datetime 的可靠转换微博时间字段的混乱是数据质量最大隐患。clean_timestamp方法必须覆盖全部常见格式并提供 fallback 机制。以下为增强版校验函数支持毫秒级精度回填# utils/time_utils.py from datetime import datetime, timedelta import re def parse_weibo_time(time_str): 解析微博时间字符串返回精确到秒的 datetime 对象 支持格式 - 今天 14:23 - 昨天 09:12 - 03月15日 16:45 - 2024-03-15 10:22:33 - 2024-03-15 if not time_str: return None time_str time_str.strip() now datetime.now() # 匹配 今天 HH:MM m re.match(r今天\s(\d{1,2}:\d{2}), time_str) if m: hms m.group(1) try: t datetime.strptime(hms, %H:%M) return now.replace(hourt.hour, minutet.minute, second0, microsecond0) except ValueError: pass # 匹配 昨天 HH:MM m re.match(r昨天\s(\d{1,2}:\d{2}), time_str) if m: hms m.group(1) try: t datetime.strptime(hms, %H:%M) yesterday now - timedelta(days1) return yesterday.replace(hourt.hour, minutet.minute, second0, microsecond0) except ValueError: pass # 匹配 MM月DD日 HH:MM m re.match(r(\d{1,2})月(\d{1,2})日\s(\d{1,2}:\d{2}), time_str) if m: month, day, hms int(m.group(1)), int(m.group(2)), m.group(3) try: t datetime.strptime(hms, %H:%M) year now.year # 处理跨年若当前月微博月或同月但日当前日则为去年 if month now.month or (month now.month and day now.day): year - 1 return datetime(year, month, day, t.hour, t.minute, 0, 0) except ValueError: pass # 匹配 YYYY-MM-DD HH:MM:SS 和 YYYY-MM-DD HH:MM for fmt in [%Y-%m-%d %H:%M:%S, %Y-%m-%d %H:%M, %Y-%m-%d]: try: return datetime.strptime(time_str, fmt) except ValueError: continue return None # 在 items.py 中调用 def clean_timestamp(self, value): return parse_weibo_time(value)4.3 启动命令与实时监控技巧用 stats 精确评估吞吐瓶颈不要依赖scrapy crawl weibo的终端输出判断速度。真实吞吐必须通过 Scrapy 内置 stats 指标验证# 启动时启用 stats 收集 scrapy crawl weibo -a keywordsAI,Python,大数据 -s LOG_LEVELINFO -s CLOSESPIDER_ITEMCOUNT10000 # 实时查看 stats另开终端 scrapy stats --spiderweibo --interval5关键指标解读downloader/request_count: 总请求数应 ≈item_scraped_count × 平均页数downloader/response_status_count/200: 成功响应占比低于 95% 说明 Cookie 或 UA 有问题retry/count: 重试次数超过request_count × 0.1表示反爬强度高elapsed_time_seconds: 总耗时用于计算真实吞吐item_scraped_count / elapsed_time_secondsscheduler/enqueued/memory: 内存队列积压持续 1000 表示解析速度跟不上请求速度提示若downloader/response_status_count/418持续上升立即检查 Cookie 有效性若retry/count骤增降低CONCURRENT_REQUESTS并增加DOWNLOAD_DELAY。真正的“2000条/秒”是分布式集群≥10台机器 Cookie 池≥50个 Splash 集群≥5节点下的系统吞吐单机目标应设为 150~200 条/秒并长期稳定运行。本文还有配套的精品资源点击获取