用AI做舆情监控：从社交媒体爬取到情感分析的实时Pipeline

左手厨刀右手茼蒿

1016人浏览 · 2026-07-25 01:20:47

左手厨刀右手茼蒿 · 2026-07-25 01:20:47 发布

用AI做舆情监控：从社交媒体爬取到情感分析的实时Pipeline

一、场景痛点与技术挑战

企业舆情监控是品牌安全的底线防线。
一条负面消息在30分钟内就能引爆全网。
传统人工巡检根本跟不上信息传播速度。
社交媒体日均产生数亿条内容。
人工筛选耗时且遗漏率高。

核心痛点有四个。
一是数据源碎片化。
微博、抖音、小红书、Twitter各成孤岛。
每个平台的API和页面结构完全不同。
二是实时性要求极高。
舆情窗口期仅30-60分钟。
延迟1小时的报告基本失去价值。
三是情感分析精度不足。
中文语境下讽刺、反话、隐喻难以识别。
简单词典匹配误判率超过40%。
四是数据量与算力矛盾。
日处理百万条文本需要弹性算力调度。
峰值流量是均值3-5倍。

技术挑战更棘手。
反爬机制越来越严格。
验证码、IP限制、登录墙层层叠加。
数据清洗噪音大。
广告、水军、机器人内容占比超过30%。
模型推理延迟与吞吐量平衡困难。
实时Pipeline需要毫秒级响应。

二、核心原理与架构设计

实时舆情Pipeline分五个核心模块。

采集模块负责多平台数据获取。
每个平台一个独立爬虫worker。
微博用API+页面混合采集。
小红书和抖音通过移动端接口。
Twitter用官方API v2流式端点。
所有worker将原始数据推入Kafka。

清洗模块负责数据去噪。
规则引擎过滤广告和水军内容。
正则匹配+模式识别去除营销模板。
机器人检测通过行为特征识别。
发帖频率、内容重复度、账号属性综合判断。
清洗后的数据写入Elasticsearch。

预处理模块负责文本标准化。
中文分词用jieba+自定义行业词典。
去停用词、统一缩写、修复错别字。
提取关键词和实体（品牌名、人名、地名）。
实体识别用BERT-based NER模型。

情感分析模块是核心决策层。
三级情感分类：正面、负面、中性。
细粒度情感：愤怒、失望、担忧、期待。
讽刺检测用专用子模型。
输出情感置信度而非硬标签。
置信度低于阈值的内容标记为待人工复核。

告警模块负责实时通知。
负面情感超过阈值触发告警。
同一话题在短时间内密集出现触发告警。
告警级别：关注、预警、紧急。
紧急告警同步推送钉钉和企业微信。

三、生产级代码实现

多平台数据采集框架

"""多平台舆情数据采集框架"""
import asyncio
import hashlib
import json
import logging
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field

logger = logging.getLogger("sentiment_collector")

@dataclass
class RawPost:
    platform: str
    post_id: str
    content: str
    author: str
    timestamp: float
    metadata: dict = field(default_factory=dict)

class BaseCollector(ABC):
    """爬虫基类：统一采集接口"""

    def __init__(self, platform: str, rate_limit: float = 1.0):
        self.platform = platform
        self.rate_limit = rate_limit  # 请求间隔秒数
        self._last_request = 0.0

    async def _throttle(self):
        """请求限速"""
        elapsed = time.time() - self._last_request
        if elapsed < self.rate_limit:
            await asyncio.sleep(self.rate_limit - elapsed)
        self._last_request = time.time()

    @abstractmethod
    async def fetch_recent(self, keywords: list[str],
                           since: float) -> list[RawPost]:
        """获取最近帖子"""
        ...

class WeiboCollector(BaseCollector):
    """微博数据采集器"""

    def __init__(self, cookie: str, rate_limit: float = 2.0):
        super().__init__("weibo", rate_limit)
        self.cookie = cookie
        self.headers = {
            "User-Agent": "Mozilla/5.0...",
            "Cookie": cookie,
        }

    async def fetch_recent(self, keywords: list[str],
                           since: float) -> list[RawPost]:
        posts = []
        for kw in keywords:
            await self._throttle()
            # 微博搜索API模拟
            url = f"https://m.weibo.cn/api/container/getContainer?"
            url += f"containerid=100103type%3D1%26q%3D{kw}"

            # 实际生产中用aiohttp请求
            # resp = await aiohttp_get(url, headers=self.headers)
            # data = json.loads(resp)
            # 这里模拟数据返回
            mock_data = [
                {
                    "id": hashlib.md5(f"{kw}_{i}".encode()).hexdigest()[:12],
                    "text": f"关于{kw}的讨论内容{i}",
                    "user": f"user_{i}",
                    "created_at": time.time() - i * 60,
                }
                for i in range(5)
            ]

            for item in mock_data:
                if item["created_at"] >= since:
                    posts.append(RawPost(
                        platform=self.platform,
                        post_id=item["id"],
                        content=item["text"],
                        author=item["user"],
                        timestamp=item["created_at"],
                    ))
        return posts

class TwitterCollector(BaseCollector):
    """Twitter数据采集器（API v2）"""

    def __init__(self, bearer_token: str, rate_limit: float = 1.0):
        super().__init__("twitter", rate_limit)
        self.bearer = bearer_token

    async def fetch_recent(self, keywords: list[str],
                           since: float) -> list[RawPost]:
        posts = []
        query = " OR ".join(keywords)
        # Twitter API v2 filtered stream
        # 实际生产中用streaming endpoint
        mock_data = [
            {
                "id": f"tw_{hashlib.md5(query.encode()).hexdigest()[:8]}",
                "text": f"Twitter discussion about {query}",
                "user": f"tw_user_{i}",
                "created_at": time.time() - i * 120,
            }
            for i in range(3)
        ]
        for item in mock_data:
            if item["created_at"] >= since:
                posts.append(RawPost(
                    platform=self.platform,
                    post_id=item["id"],
                    content=item["text"],
                    author=item["user"],
                    timestamp=item["created_at"],
                ))
        return posts

class CollectorScheduler:
    """采集调度器：协调多平台Worker"""

    def __init__(self, collectors: list[BaseCollector]):
        self.collectors = collectors

    async def collect_all(self, keywords: list[str],
                          since: float) -> list[RawPost]:
        """并行采集所有平台数据"""
        tasks = [
            c.fetch_recent(keywords, since)
            for c in self.collectors
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        all_posts = []
        for result in results:
            if isinstance(result, Exception):
                logger.error(f"采集失败: {result}")
                continue
            all_posts.extend(result)

        logger.info(f"本轮采集: {len(all_posts)}条")
        return all_posts

数据清洗与情感分析Pipeline

"""数据清洗与情感分析Pipeline"""
import re
import hashlib
from dataclasses import dataclass
from enum import Enum
from collections import Counter

class SentimentLevel(Enum):
    POSITIVE = "正面"
    NEGATIVE = "负面"
    NEUTRAL  = "中性"
    UNCERTAIN = "待复核"

class EmotionTag(Enum):
    ANGER    = "愤怒"
    DISAPPOINT = "失望"
    WORRY    = "担忧"
    EXPECT   = "期待"
    SATISFY  = "满意"

@dataclass
class AnalyzedPost:
    post_id: str
    content: str
    platform: str
    sentiment: SentimentLevel
    sentiment_score: float   # -1.0 ~ 1.0
    emotion: EmotionTag | None
    keywords: list[str]
    entities: list[str]
    is_spam: bool
    confidence: float        # 0.0 ~ 1.0

class DataCleaner:
    """数据清洗：去除噪音内容"""

    # 广告关键词模式
    AD_PATTERNS = [
        r"加微信|加VX|扫码领取|限时优惠|点击购买",
        r"代购|批发价|工厂直供|最低价|免费送",
    ]

    # 水军行为特征阈值
    SPAM_THRESHOLD = {
        "max_posts_per_hour": 20,
        "min_unique_ratio": 0.3,
    }

    _ad_regex = None
    _content_history: dict[str, list[str]] = {}

    def __init__(self):
        self._ad_regex = re.compile(
            "|".join(self.AD_PATTERNS), re.IGNORECASE
        )

    def is_ad(self, content: str) -> bool:
        """广告内容检测"""
        return bool(self._ad_regex.search(content))

    def is_water_army(self, author: str, content: str) -> bool:
        """水军检测：基于行为特征"""
        self._content_history.setdefault(author, []).append(content)
        history = self._content_history[author]

        if len(history) > self.SPAM_THRESHOLD["max_posts_per_hour"]:
            return True

        unique_ratio = len(set(history)) / max(len(history), 1)
        if unique_ratio < self.SPAM_THRESHOLD["min_unique_ratio"]:
            return True

        return False

    def clean(self, post) -> AnalyzedPost | None:
        """清洗单条数据"""
        if self.is_ad(post.content):
            return None  # 广告内容直接丢弃

        is_spam = self.is_water_army(post.author, post.content)
        if is_spam:
            return None  # 水军内容丢弃

        return AnalyzedPost(
            post_id=post.post_id,
            content=post.content,
            platform=post.platform,
            sentiment=SentimentLevel.NEUTRAL,
            sentiment_score=0.0,
            emotion=None,
            keywords=[],
            entities=[],
            is_spam=False,
            confidence=0.0,
        )

class SentimentAnalyzer:
    """情感分析引擎"""

    # 简化版情感词典（生产环境用BERT模型）
    NEGATIVE_WORDS = {
        "垃圾": -0.8, "糟糕": -0.7, "失望": -0.6,
        "气愤": -0.9, "投诉": -0.5, "退货": -0.4,
        "骗": -0.85, "坑": -0.75, "差": -0.5,
        "黑心": -0.9, "抵制": -0.7, "曝光": -0.6,
    }

    POSITIVE_WORDS = {
        "好": 0.5, "赞": 0.7, "满意": 0.6,
        "推荐": 0.65, "优秀": 0.8, "喜欢": 0.6,
        "棒": 0.7, "支持": 0.55, "靠谱": 0.65,
    }

    # 讽刺模式识别
    IRONY_PATTERNS = [
        r"真是.*的好",   # "真是垃圾的好"
        r"所谓.*其实",   # "所谓高端其实低端"
        r"还.*呢",       # "还好呢"
    ]

    def __init__(self, confidence_threshold: float = 0.6):
        self.threshold = confidence_threshold
        self._irony_regex = re.compile(
            "|".join(self.IRONY_PATTERNS)
        )

    def detect_irony(self, text: str) -> bool:
        """讽刺检测"""
        return bool(self._irony_regex.search(text))

    def analyze(self, post: AnalyzedPost) -> AnalyzedPost:
        """分析情感"""
        text = post.content

        # 计算情感得分
        score = 0.0
        matched_words = 0
        for word, weight in self.NEGATIVE_WORDS.items():
            if word in text:
                score += weight
                matched_words += 1
        for word, weight in self.POSITIVE_WORDS.items():
            if word in text:
                score += weight
                matched_words += 1

        # 讽刺检测翻转情感
        if self.detect_irony(text) and score > 0:
            score = -score * 0.8

        # 计算置信度
        confidence = min(matched_words / 3.0, 1.0)

        # 情感分类
        if confidence < self.threshold:
            post.sentiment = SentimentLevel.UNCERTAIN
        elif score > 0.2:
            post.sentiment = SentimentLevel.POSITIVE
        elif score < -0.2:
            post.sentiment = SentimentLevel.NEGATIVE
        else:
            post.sentiment = SentimentLevel.NEUTRAL

        post.sentiment_score = round(score, 3)
        post.confidence = round(confidence, 2)

        # 情感标签
        if post.sentiment == SentimentLevel.NEGATIVE:
            if score < -0.7:
                post.emotion = EmotionTag.ANGER
            elif score < -0.5:
                post.emotion = EmotionTag.DISAPPOINT
            else:
                post.emotion = EmotionTag.WORRY

        return post

class AlertEngine:
    """舆情告警引擎"""

    def __init__(self, negative_threshold: float = 0.3,
                 cluster_threshold: int = 10):
        self.negative_threshold = negative_threshold
        self.cluster_threshold = cluster_threshold
        self._topic_counter: Counter = Counter()

    def evaluate(self, posts: list[AnalyzedPost]) -> list[dict]:
        """评估告警条件"""
        alerts = []
        negative_count = 0

        for post in posts:
            if post.sentiment == SentimentLevel.NEGATIVE:
                negative_count += 1
                # 提取话题关键词用于聚类
                for kw in post.keywords or [post.content[:10]]:
                    self._topic_counter[kw] += 1

        # 负面比例告警
        if len(posts) > 0:
            ratio = negative_count / len(posts)
            if ratio > self.negative_threshold:
                level = "紧急" if ratio > 0.6 else "预警" if ratio > 0.4 else "关注"
                alerts.append({
                    "type": "negative_ratio",
                    "level": level,
                    "ratio": round(ratio, 3),
                    "total": len(posts),
                    "negative": negative_count,
                })

        # 话题密集告警
        hot_topics = [
            (topic, count) for topic, count in self._topic_counter.items()
            if count >= self.cluster_threshold
        ]
        for topic, count in hot_topics:
            alerts.append({
                "type": "topic_cluster",
                "level": "预警" if count > 20 else "关注",
                "topic": topic,
                "count": count,
            })

        return alerts