利用中间件实现任务去重与分发精细化：股吧舆情数据采集与分析实战

原创

jackcode

发布于 2025-07-23 11:48:26

1830

一、项目背景：为什么要精细化采集东财股吧？

在股票市场中，情绪驱动效应越来越明显。散户投资者对个股的情绪变化，往往先于价格的异动，而东财股吧作为国内最活跃的财经论坛之一，承载了大量关于个股的观点、讨论和预判内容。这类内容虽然非结构化，却包含极高的信号密度，是舆情监测与热点追踪的重要来源。

但在实际采集中，我们常常面临以下问题：

大量关键词重复调度，资源浪费；
分发逻辑混乱，任务调度粒度不够；
数据反复采集，结构不统一；
页面存在一定反爬机制，如频繁访问封IP、页面结构变化、UA识别等。

为解决上述问题，我们设计了一个基于中间件机制的去重与分发精细化采集方案，并围绕东财股吧构建了可用于舆情分析的数据模型与结构化流程。

二、数据目标：围绕舆情建模的三层信息结构

为满足对投资舆情的量化需求，本项目从三个维度设计了数据采集与结构化模型：

第一维度是“事件层”，聚焦于每条帖子所包含的主题、个股标识与热度信号。

例如我们关注的字段包括股票代码、对应吧名、帖子标题、正文内容、发帖时间、评论数量、互动数量（如点赞、点踩）等。这一层结构是后续进行“事件演化轨迹”分析的基础。

第二维度是“用户层”，用于识别关键发帖用户及潜在的行为操纵迹象。

我们会提取作者昵称、是否为楼主、发帖频率、单日活跃度等字段，通过行为建模技术识别“活跃舆论引导者”或“集中推送账号”，用于评估信息来源的稳定性与偏向性。

第三维度是“情绪层”，面向自然语言处理与文本分析任务设计。

我们通过分词技术提取关键词，结合情绪词典或预训练模型对正负面倾向进行打分。同时，我们还统计每条帖子中提及股票数量、关键词重复率等指标，为后续热点识别与情绪指数建模提供依据。

综合三维结构设计后，最终目标是实现：

情绪趋势可视化；
舆情高发时间点识别；
活跃用户及行为路径建模；
特定关键词（如“暴雷”“退市”等）相关个股自动预警。

三、采集方式：中间件驱动的去重与调度机制

为了保障数据获取的高效性与稳定性，我们采用了Scrapy框架 + Redis调度 + 自定义中间件的模式，构建任务调度、代理设置、身份模拟、指纹去重等模块，整体流程如下：

关键词任务投放：将需要监测的股票代码、话题关键词等，封装成URL规则，分发到任务队列。
请求调度器接入Redis中间件：保证任务唯一性，避免重复访问。
自定义下载中间件实现代理注入与身份伪装：包括亿牛云代理IP设置、用户身份模拟。
页面解析模块抽取结构化字段：对帖子作者、时间、内容、互动量等字段提取，并归一化处理。
数据写入与清洗模块：统一写入数据库或JSON文件，供后续分析或模型训练使用。

下面是核心中间件与爬虫代理配置的示意：

代理设置与身份模拟

# middlewares.py
class CustomDownloaderMiddleware:
    def __init__(self):
        #设置代理IP（参考亿牛云爬虫代理）
        self.proxy_host = "proxy.16yun.cn"
        self.proxy_port = "3100"
        self.proxy_user = "16YUN"
        self.proxy_pass = "16IP"
        self.user_agents = [
            "Mozilla/5.0 ...",  # 多个浏览器标识
        ]
        self.cookies = {
            'qgqp_b_id': 'xxxxxx',  # 示例cookie字段
        }

    def process_request(self, request, spider):
        # 注入代理
        request.meta['proxy'] = f"http://{self.proxy_user}:{self.proxy_pass}@{self.proxy_host}:{self.proxy_port}"
        # 注入UA
        request.headers['User-Agent'] = random.choice(self.user_agents)
        # 注入Cookie
        request.cookies = self.cookies

去重机制与请求指纹处理

# dupefilter.py
from scrapy.dupefilters import RFPDupeFilter
import hashlib

class CustomURLDupeFilter(RFPDupeFilter):
    def request_fingerprint(self, request):
        # 对URL进行归一处理，忽略动态参数
        url = request.url.split('&_')[0]
        return hashlib.sha1(url.encode('utf-8')).hexdigest()

采集主逻辑示例

# spider.py
class GubaSpider(RedisSpider):
    name = 'guba'
    redis_key = 'guba:start_urls'

    def make_requests_from_url(self, url):
        return scrapy.Request(url=url, dont_filter=True)

    def parse(self, response):
        posts = response.xpath('//div[@class="articleh normal_post"]')
        for post in posts:
            yield {
                'author': post.xpath('.//span[@class="l5 a5"]/text()').get(),
                'time': post.xpath('.//span[@class="l6"]/text()').get(),
                'content': post.xpath('.//a[@class="note"]/text()').get(),
                'url': response.url
            }