简介：这份PDF收录何佳、周长胜、石显锋发表于《郑州大学学报（理学版）》的论文，聚焦网络舆情监控系统的实现路径，适合信息检索、中文信息处理与文本挖掘方向的学习者和研发人员参考。文章从2009年CNNIC网民规模数据切入，讨论网络舆情监控的现实需求，并按功能模块梳理系统结构：网页信息采集、信息预处理、文本挖掘、热点发现与跟踪及结果展示；其中对主题网络爬虫、页面清洗、分词与特征提取、文本分类聚类、观点倾向性识别等关键环节均有展开，可帮助读者理解从海量非结构化网页到舆情分析报告的完整技术链路。资源包仅含1个PDF文件，约312KB，轻量便于查阅和归档。目前已有67人学习下载，适合作为课程作业、论文选题或系统设计阶段的技术框架参考。

1. 网络舆情监控系统在监控什么：一条差评背后的四层链路

做网络舆情监控系统的实现方法，起点常让人措手不及。某天早上，运营在群里贴出一条截图，说这个词昨晚在几个平台被反复提起，问能不能把前后三天的讨论完整拉出来。手工翻页，翻到凌晨也拼不出全貌。这条需求拆开是四层：采集层把散落在多源的内容按时间收回，存储层用倒排索引把它们变成可检索的刻度，分析层给出情感倾向、主题归属和热度曲线，预警层在热度越过阈值时通知真正相关的人。四层里任何一层偷工，后面全歪。适合正在做第一版舆情系统的人，也适合接手一套跑了大半年、越跑越慢的系统的从业者。接下来按这条链路往下走，把每一层的选型理由、可执行代码和参数讲清楚。

2. 采集层怎么搭：从多源抓取到增量水位与去重

采集层的目标只有一个：在可接受的成本下，把目标源的公开内容按时间顺序完整拿回来，且不漏、不重、不炸。听起来简单，做起来是整条链路里最容易翻车的地方。常见做法是把源分成三类分别处理：有稳定接口的走接口，只有页面的走浏览器渲染或轻量解析，更新频繁、需要实时性的走推送或轮询。三类源的抓取节奏、失败模型和数据结构都不一样，硬塞进一个采集器里，最后一定变成一团乱麻。

2.1 三类数据源与采集通道的选型

先做一次源盘点，把每个源归到表里，后面所有参数都从这张表推。不做盘点直接写代码，常见的后果是：某个源凌晨频控收紧，整批任务全挂，第二天才发现水位停在十几个小时前。

选型上有一条经验：能拿到结构化时间字段的源，一律走时间游标，不要用页码游标。页码游标在上游有新内容插入时会整体位移，翻到后面必然重复或漏抓，这是最典型的血泪经验。时间游标的前提是上游按时间升序返回，做不到就退化成“水位减重叠窗口”的近似方案。

2.2 用时间游标写一个可断点续抓的增量采集器

下面这段是整个采集层的地基，逻辑不复杂，但每个参数都值得抠。

import time import httpx def pull_source(client, source, cursor, max_pages=200): """ source: 源配置字典 cursor: 上次抓到的最大发布时间戳（秒），首次为 0 返回: (新增记录列表, 新水位) """ records, new_cursor = [], cursor page_size = source.get("page_size", 50) for page in range(1, max_pages + 1): params = { source.get("page_param", "page"): page, source.get("size_param", "size"): page_size, source.get("sort_param", "sort"): "time_asc", } try: resp = client.get(source["url"], params=params, timeout=15) resp.raise_for_status() items = source["parser"](resp.json()) except Exception as exc: # 单页失败不推进水位，交给上层退避重试 raise RuntimeError(f"{source['name']} page={page} err={exc}") from exc if not items: break fresh = [it for it in items if it["ts"] > cursor] records.extend(fresh) page_max = max(it["ts"] for it in items) new_cursor = max(new_cursor, page_max) # 不足一页 或 本页已无新数据，视为追平 if len(items) < page_size or page_max <= cursor: break time.sleep(source.get("sleep", 0.8)) return records, new_cursor

逐条说。cursor是水位，只在上游按时间升序返回时成立；如果上游是乱序或按热度排序，就要把游标改成cursor - overlap，overlap一般取 120～300 秒，用少量重复换取不漏。max_pages是防死循环的闸门，没有它，一旦上游对超出范围的分页返回首页内容，采集器会一直翻到天亮。sleep控制请求间隔，接口型源 0.5～1 秒，页面型源 1～3 秒，具体看频控反馈，别一上来就调到 0.1 秒再被封。

调用侧要配连接复用和指数退避，这两点决定了凌晨批量任务能不能活下来：

def run_source(source, state, backoff=(30, 120, 600)): cursor = state.get(source["name"], 0) # 连接池复用，避免每个请求重新建连与解析 with httpx.Client(headers=source.get("headers", {}), limits=httpx.Limits(max_connections=8), http2=True) as client: for attempt, wait in enumerate((0,) + tuple(backoff)): if wait: time.sleep(wait) try: rows, new_cursor = pull_source(client, source, cursor) return rows, new_cursor except RuntimeError: continue raise RuntimeError(f"{source['name']} 连续重试失败，保留旧水位 {cursor}")

max_connections=8是常见起点，源多的时候按总并发不超过 40 分配。退避序列从 30 秒起，是因为多数频控的冷却窗口在分钟级；直接 1 秒一次硬顶，只会把临时限制变成持久限制。水位只在整轮成功后才落盘，中途失败保留旧值，宁可重抓一轮也不能丢一段。

提示：水位落盘要带版本号或更新时间，采集器启动时先读水位再看有没有超过 2 小时未推进的源，超了就告警。这套检查是整条链路的黑匣子，出问题第一眼看它。

2.3 去重与正文清洗：SimHash 阈值和短文本策略

同一内容被多个账号搬运、被同一源重复返回，是常态。去重分两层：一层是硬主键去重，用源标识 + 源内 ID做唯一键，写库时忽略冲突；另一层是近似去重，处理改标题、改标点、加前缀的搬运。后者用 SimHash 就够，成本低、可增量。

import hashlib import re TOKEN_RE = re.compile(r"[\u4e00-\u9fa5]|[a-zA-Z0-9]+") def shingles(text, n=3): toks = TOKEN_RE.findall(text.lower()) if len(toks) <= n: return toks return ["".join(toks[i:i + n]) for i in range(len(toks) - n + 1)] def simhash(text, bits=64): if not text or len(text) < 20: return None # 太短的正文不参与近似去重 vec = [0] * bits for tok in shingles(text): h = int(hashlib.md5(tok.encode("utf-8")).hexdigest(), 16) for i in range(bits): vec[i] += 1 if (h >> i) & 1 else -1 out = 0 for i in range(bits): if vec[i] > 0: out |= 1 << i return out def hamming(a, b): return bin(a ^ b).count("1")

参数上，bits=64配合海明距离不超过 3，是中文短中文本的经验值；距离放宽到 5 以上，同话题不同观点会被误杀，收紧到 1 又会漏掉改几个字的搬运。shingles用 3-gram 而不是单字，是为了保留词序信息，单字哈希对“不满意”和“满意不”这种差别不敏感。

近邻检索不要全表比对。把 64 位切成 4 段 16 位，任一相同即入候选集，比较次数能压下去一到两个数量级，这个结构塞进缓存服务里就能扛住日常增量。

正文清洗的坑比去重更多。列表页拿到的摘要往往截断在句子中间，直接入库会让后续情感判断偏负。常见做法是保留原始摘要字段，另存一个content_full，抽取失败时不要用摘要顶替，而是标记extract_status=partial，让分析层决定是否降权。

注意：正文长度小于 40 字的内容，情感模型的输出基本是噪音，建议单独走一条轻量通道，只用规则判定并在结果里标注置信度低，不要混进主指标。

3. 存储与检索：中文分词和倒排索引决定查询上限

采集回来的东西，最终要回答三类问题：某关键词在某时间段怎么讨论的、某个话题下有哪些内容、某条内容的传播轨迹是什么。这三类查询天然适合倒排索引，而中文场景下，分词器的选择直接决定了这套系统能不能用。

3.1 为什么舆情系统几乎绕不开倒排索引加时间切片

用关系库做关键词检索，like '%词%'在百万级数据上就开始明显拖慢，而且无法处理同义词、短语、权重。倒排索引把“词到文档”的映射预先建好，查询从扫表变成查词典，量级差得非常明显。中文还有一个额外问题：没有天然空格分隔，必须先分词。分词粒度不对，会出现“搜‘舆情’搜不到‘网络舆情’里的那一段”的尴尬。

时间切片是第二个关键决定。按天建索引（形如opinion-20240601），带来三个好处：过期数据整索引删除，不用一条条删；冷热分离简单，最近 7 天放热节点，更早的转冷；单索引规模可控，分片恢复时间短。代价是跨天查询要合并多个索引，但这个代价远小于一个巨型索引带来的运维麻烦。

量级到百万以上，按天切几乎成了默认选择。再往上，会在天索引内部再加路由字段，把同一话题的内容尽量落到同一分片，减少查询时的扇出。

3.2 索引模板：字段映射、分词器与时间切片参数

模板要一次定好，后面改映射的成本极高。下面这份是一个可用起点，字段按业务再增删。

PUT _index_template/opinion_daily { "index_patterns": ["opinion-*"], "template": { "settings": { "number_of_shards": 3, "number_of_replicas": 1, "refresh_interval": "30s", "mapping.total_fields.limit": 2000 }, "mappings": { "dynamic": "strict", "properties": { "doc_id": { "type": "keyword" }, "source": { "type": "keyword" }, "channel": { "type": "keyword" }, "publish_ts": { "type": "date", "format": "epoch_second" }, "title": { "type": "text", "analyzer": "ik_max_word", "search_analyzer": "ik_smart" }, "content": { "type": "text", "analyzer": "ik_max_word", "search_analyzer": "ik_smart" }, "keywords": { "type": "keyword" }, "sentiment": { "type": "float" }, "topics": { "type": "keyword" }, "simhash": { "type": "keyword" }, "metrics": { "type": "object", "enabled": false } } } } }

几个参数值得单独解释。analyzer用ik_max_word建索引，是为了把“网络舆情监控系统”切成最细粒度的多个词，提高召回；search_analyzer用ik_smart，是为了查询时不再过度切分，保证精度。这套粗细搭配是中文检索里最常见的组合。refresh_interval从默认 1 秒调到 30 秒，写入吞吐能提升数倍，代价是数据延迟半分钟可见，对舆情场景完全可接受。

dynamic: strict是防字段爆炸的关键。不设它，上游多返回一个嵌套对象，映射就自动扩一层，几个月后字段数上千，集群状态变红。metrics显式设enabled: false，表示这个对象只存不索引，需要时取回原文自己解析。mapping.total_fields.limit设 2000 是上限保护，真撞到说明字段设计该重构了。

3.3 写入链路：批次大小与背压控制

批量写入的批次大小不是越大越好。太小，请求数过多，集群协调开销占比高；太大，单请求内存压力大，失败重试代价也大。经验区间是每批 500～2000 条文档、单批体积控制在 5～15 MB，先按条数再按体积做双限。

def bulk_write(client, index, docs, batch=1000, max_bytes=10 * 1024 * 1024): for i in range(0, len(docs), batch): chunk, size, ops = [], 0, [] for d in docs[i:i + batch]: body = d.get("_raw", "") size += len(body.encode("utf-8")) if size > max_bytes and chunk: ops.extend(flush(client, index, chunk)) chunk, size = [], 0 chunk.append(d) if chunk: ops.extend(flush(client, index, chunk)) # 有失败项时按错误类型分流：映射冲突直接丢弃并记录，超时进重试队列 return ops def flush(client, index, chunk): body = [] for d in chunk: body.append({"index": {"_index": index, "_id": d["doc_id"]}}) body.append({k: v for k, v in d.items() if k != "_raw"}) resp = client.bulk(operations=body, refresh=False) if resp.get("errors"): return [it for it in resp["items"] if it["index"].get("error")] return []

背压来自两头：采集侧生产速度和写入侧消费速度不匹配时，队列会持续堆积。做法是给队列设高水位，超过就把采集端的sleep动态放大，而不是让队列无限涨到内存溢出。失败项要分类处理：映射冲突属于数据问题，重试多少次都失败，应直接落死信；超时和连接错误属于临时问题，进重试队列并带退避。把这两类混在一起重试，是常见的隐性故障源。

4. 舆情分析怎么落地：情感判定、主题聚类与热点识别

分析层最容易被写成演示代码：调个模型，输出一个标签，看起来很完整，上线后发现标签分布严重偏移、聚类结果每天都不一样、热点永远是那几个大词。问题不在模型，在于没有为业务设计分层策略和稳定输出。

4.1 情感判定：规则先筛、模型兜底

纯模型的问题是可解释性差、短文本不准、版本一换结果就漂。纯规则的问题是召回低、遇到反讽就失灵。可行的组合是：高置信区间用规则快速判定并直接出结果，中间地带交给模型，模型也不确定的归入“待人工抽检”，不强行给标签。

NEG_WORDS = {"失望", "投诉", "退货", "崩溃", "卡顿", "虚假"} POS_WORDS = {"满意", "推荐", "好用", "及时", "解决"} INTENSIFIERS = {"非常": 1.5, "特别": 1.4, "有点": 0.6, "稍微": 0.5} NEGATORS = {"不", "没", "未", "别"} def rule_score(text): score = 0.0 hit_neg = hit_pos = 0 for w, k in INTENSIFIERS.items(): if w in text: text = text.replace(w, "") for w in NEG_WORDS: if w in text: hit_neg += 1 for w in POS_WORDS: if w in text: hit_pos += 1 score = (hit_pos - hit_neg) / max(hit_pos + hit_neg, 1) if any(n in text for n in NEGATORS): score = -score * 0.8 # 否定词整体翻转并降权 return round(score, 3) def judge(text, model, rule_th=0.6, model_th=0.75): s = rule_score(text) if abs(s) >= rule_th: return {"label": "neg" if s < 0 else "pos", "score": abs(s), "via": "rule"} prob = model.predict_proba([text])[0] conf = float(max(prob)) if conf >= model_th: return {"label": "neg" if prob[0] > prob[1] else "pos", "score": conf, "via": "model"} return {"label": "unknown", "score": conf, "via": "fallback"}

rule_th=0.6意味着词面证据占比超过六成才直接采信，低于这个值说明句子结构复杂或含混。model_th=0.75是模型置信门槛，低于它宁可标unknown，因为一个错误的情感标签进入趋势曲线，比缺一条数据危害大得多。实际运行里unknown的比例通常落在 8%～15%，这部分每周抽 200 条人工标注，反过来评估规则词表和模型版本是否需要更新。

注意：情感值不要只存最终标签，把via字段一起存进去。趋势图按via分层看，能立刻发现是规则失效还是模型漂移。

4.2 主题聚类：向量化加阈值聚类，簇数不要预先定

主题聚类的目标是“同一件事的讨论聚在一起”，而不是“把内容分成 K 类”。预先定 K 是演示代码的思路，真实数据里每天的新话题数根本不受控。做法是先向量化，再按相似度阈值增量归并。

import numpy as np def assign_topic(vec, centroids, threshold=0.72): """vec: 归一化后的句向量; centroids: {topic_id: 向量}""" if not centroids: return None, 0.0 ids = list(centroids.keys()) mat = np.stack([centroids[i] for i in ids]) sims = mat @ vec # 已归一化，点积即余弦 idx = int(np.argmax(sims)) best = float(sims[idx]) if best >= threshold: return ids[idx], best return None, best def update_centroid(centroid, vec, alpha=0.1): new = (1 - alpha) * centroid + alpha * vec return new / np.linalg.norm(new)

threshold=0.72是余弦相似度，对应夹角约 44 度，是个偏宽松的起点。调高到 0.8 以上，一个话题会被拆成多个，热点统计被稀释；调低到 0.6 以下，不同话题会粘在一起，看起来像是“全网在讨论同一件事”。alpha=0.1控制中心漂移速度，设太大，话题会被最新几条内容带跑偏；设太小，话题演化跟不上。这个参数最好用一周的历史数据跑一遍，看簇的合并与分裂频率再定。

新增内容如果和所有中心都不够近，就新建话题，不要强行归到最近的簇。这条规则能让长尾话题保留下来，代价是话题数会缓慢增长，需要定期做一次合并清理，把相似度超过 0.85 且成员数很少的簇并掉。

4.3 热点识别：用增长率和跨源扩散，而不是绝对量

按绝对数量排序，结果永远是那几个常青大词。真正的热点是“短时间内相对自身基线明显抬升、并且跨多个源同时出现”的话题。用两个因子：增长率和扩散度，再叠加对数压缩防止爆量压过一切。

import math def hot_score(c_now, c_prev, source_cnt, base_source=3, w=(1.0, 0.5)): growth = (c_now - c_prev) / max(c_prev, 1) # 相对自身基线 spread = max(source_cnt - base_source, 0) / base_source score = w[0] * math.log1p(growth * 10) + w[1] * math.log1p(spread) return round(score, 3)

log1p(growth * 10)让增长率从 1 倍涨到 10 倍带来的分数变化逐渐变缓，避免一个小众话题因为基数小、增长率虚高冲到榜首。base_source=3表示至少三个源同时出现才算扩散，这个值按接入源总数调整，源少的时候设 2。两个权重之和不必为 1，因为分数只用于排序，不做概率解释。

窗口选择上，短窗口（15 分钟）抓突发，长窗口（6 小时）抓持续发酵。两套同时跑，短窗口报警、长窗口定级，比单一窗口稳得多。

5. 上线后才暴露的坑：五类故障的排查顺序

演示环境和生产环境的差别，基本都在这五条上。按现象、原因、解决三段来写，遇到类似症状可以直接对号。

5.1 采集与清洗环节的三次翻车

第一条：越抓越少，重复率突然升到四成以上。现象是某个源当天入库量正常，但去重后剩下的不足一半，翻页日志显示后面几页反复出现前面的内容。原因是用了页码游标，上游在两次翻页之间有新内容插入，整个列表后移，第二页其实覆盖了第一页的尾部。解决办法是把页码游标换成时间游标，取不到时间字段就用水位减 120 秒重叠窗口，重复由去重层消化。改完之后重复率回到 5% 以内，代价是每轮多抓一到两页。

第二条：凌晨定时任务集体超时，白天手动跑却正常。现象是批量任务在固定时间点全部失败，错误集中在连接超时。原因是采集器每个请求都新建连接，凌晨上游响应变慢，建连时间累积把整个任务拖垮。解决办法是长生命周期客户端加连接池复用，max_connections按源数量分配，再加一层指数退避。改完之后单轮耗时从 40 分钟压到 9 分钟，凌晨也不再整批失败。

第三条：正文抽取出来的内容半截断，情感判断整体偏负。现象是某几个源的负面占比明显高于其他源，人工抽查发现很多内容只有列表页摘要。原因是详情页抽取规则失效，回退逻辑用摘要顶替了正文，摘要天然偏负面词。解决办法是抽取失败不顶替，写入extract_status=partial，分析层对这类记录降权或者跳过。这条不修，整个情感趋势图都是歪的。

第四条：去重把同话题不同观点误杀了。现象是某次事件中，支持与反对的两种表述只留下了一条。原因是 SimHash 海明距离设到了 6，改几个字的搬运和观点相反的短评都落进了阈值内。解决办法是把距离收回 3，同时给正文长度低于 40 字的内容单独走一条不参与近似去重的通道。长短分离之后，误杀率肉眼可见地降下来。

5.2 存储与分析环节的两个隐蔽陷阱

第五条：集群状态变黄甚至变红，节点日志里全是字段数超限。现象是运行两三个月后，写入开始报映射异常，重启也不恢复。原因是动态映射开着，上游多返回一层嵌套结构，每个键都被建成独立字段，几个月后字段数破万。解决办法是模板里设dynamic: strict，不确定结构的对象用enabled: false只存不索引，同时把mapping.total_fields.limit设成 2000 作为兜底。改模板需要重建索引，这一步要提前规划好滚动切换，否则只能停写。

第六条：情感模型上线一个月后准确率悄悄下滑。现象是没有改代码，但人工抽检的准确率从 0.82 掉到 0.7 左右。原因是话题分布发生漂移，模型训练时的语料和新出现的表达方式不匹配，而且没人监控这个指标。解决办法是建一个固定的抽检集，每周抽 200 条人工标注，同时监控unknown占比和via=model的分布变化，两个指标任一异常就触发规则词表更新或模型重训。这套监控比模型本身更重要。

这六条的共同点：都能在演示阶段被忽略，在真实流量下几周内必然暴露。排查顺序建议按“采集水位 → 入库量 → 去重率 → 索引健康 → 情感分布”往下走，越靠前的层出问题，后面的指标全不可信。

6. 阈值校准与时间衰减：让预警分数经得起回放

预警阈值最忌讳拍脑袋定。我一般会先用历史数据做回放：把过去 30 天的内容按时间重放一遍，用不同阈值跑出命中集合，再和人工标注的“值得预警事件”对比，算总体的 F1。

def replay(events, thresholds, label_hit): best = None for th in thresholds: tp = fp = fn = 0 for e in events: fired = e["score"] >= th real = label_hit(e) if fired and real: tp += 1 elif fired: fp += 1 elif real: fn += 1 f1 = 2 * tp / max(2 * tp + fp + fn, 1) if best is None or f1 > best[1]: best = (th, f1) return best

thresholds用 0.1 的步长从 0.3 扫到 1.2 就够，再细没有意义，因为分数本身的方差比 0.05 大。回放里要特别注意一点：同一个事件如果持续三天，会被反复计为命中，导致 F1 虚高。做法是按事件 ID 去重，一个事件在一个窗口内只算一次。

另一个常被忽略的是时间衰减。热度分数不该只看当前窗口，历史累积要按半衰期打折，否则一条老热点会长期占着榜首。

0.5 ** (age_hours / half_life)这个式子直接乘到分数上即可。我试过用 6 小时做起点，突发话题的榜首停留时间大约 8 到 12 小时，和人工判断基本对得上。

说个教训：早期我图省事，把预警阈值直接写成常量放在配置里，结果一次大范围活动导致全量分数抬升，预警连响三天，值班的人直接把告警静音了。后来改成阈值按源和话题分别设基线，再加冷却窗口，才把这个口子堵上。阈值这东西没有后悔药，只能靠回放和持续校准。希望帮到你。

本文还有配套的精品资源，点击获取