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

用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

四、性能优化与工程实践

采集层性能优化。每个平台Worker独立进程运行。进程间通过Kafka Topic解耦。Worker崩溃不影响其他平台采集。请求限速通过令牌桶算法实现。突发流量时排队而非粗暴拒绝。

Kafka分区策略按平台划分。每个平台一个分区，避免数据交叉。消费者组按分析模块划分。清洗、情感分析、告警各一个消费组。并行消费提升处理吞吐量。

清洗层优化。广告过滤用预编译正则。一次编译多次匹配，避免重复开销。水军检测用滑动窗口。只保留最近1小时的发帖历史。超过窗口的历史数据自动清理。

情感分析模型优化。生产环境用BERT-based模型。词典匹配仅作为低置信度的快速通道。BERT模型用ONNX Runtime加速推理。batch_size=32的批量推理吞吐量提升5倍。GPU推理延迟<10ms/条。CPU推理用模型蒸馏版本延迟<50ms/条。

讽刺检测是精度瓶颈。词典匹配的讽刺检测覆盖率仅60%。生产环境需要专项训练讽刺数据集。中文讽刺语料稀缺是最大障碍。自建讽刺标注数据集至少5000条。标注质量比数量更重要。

告警引擎优化。负面比例告警用滑动窗口计算。窗口大小5分钟，步长1分钟。避免单条数据触发误告警。话题聚类用TF-IDF+余弦相似度。相似度>0.7的内容归入同一话题。

五、总结与技术提炼