赛事热词监控:用Python从弹幕提取高频词与梗句
2026/9/8 20:30:54 网站建设 项目流程

赛事运营团队在复盘一场社区杯赛时,经常要面对两类数据:一类是比分、排名、击杀数这类结构化赛果,另一类是弹幕、标题、群聊速报里的观众文本。比如“【MD融合杯】喜欢我五个1850走脸吗?”这类文本,前半句是活动品牌,后半句是观众对某一局画风的直观概括。运营希望从大量这样的文本里自动提炼高频动作词、关键数值和热门句式,用来快速生成次日战报标题,而不是比赛结束后半夜人工翻聊天记录。

这篇文章要解决的,就是一套“赛事热词与梗句监控”的最小可运行工具。它以本地导出的授权聊记录作为输入,经过清洗、分词、用户词典、事件抽取、热度统计和模板输出,形成一份可读的战报速览,再通过机器人 Webhook 推送到运营群。示例输入会直接使用标题里的“喜欢我五个1850走脸吗”,用它说明一个短文本里能抽出多少可用于战报的信息。

整套工具不需要 GPU,不需要大数据平台,在普通开发机上用 Python 就能跑通。关键是先把“要识别什么”想清楚,再写规则和代码。

1. 先把需求拆清楚:从“五个1850走脸”里到底要识别什么

1.1 观众文本至少可以拆成四个字段

“喜欢我五个1850走脸吗?”这句话如果直接拿去做词频,只能得到“喜欢”“走脸”“1850”这些孤立词。对于战报标题准备,运营通常希望得到更接近业务语义的表达:

字段示例值业务含义
数量上下文五个说明动作规模,可能是五局、五张牌、五次操作
核心数值1850可以代表攻击力、单局得分、淘汰分或卡牌数值
动作/战术词走脸观众对某一局策略风格的高度概括
原始句子喜欢我五个1850走脸吗?保留语境,供人工判断是否适合写成战报标题

这里不需要先理解“走脸”在具体游戏里到底是快攻策略还是操作名称,只需要在词典里告诉程序:这个组合词属于动作词,并且后续遇到类似语境时可以复用到同一个事件类型里。

1.2 只做词频表为什么不够

词频表能回答“哪些词出现得多”,但回答不了“这些词是如何组合在一起的”。同样是“1850”出现三次,可能来自“1850分取胜”“1850输出”“五个1850走脸”,三种语境运营价值完全不同。

更关键的是,电竞观众文本里存在大量自定义缩写、社区黑话和新造词。通用分词工具如果没有用户词典,可能把“走脸”切成“走”和“脸”,把“翻盘”切成“翻”和“盘”,最后输出词的粒度太碎,没法形成战报素材。

所以这里的实现思路是先补词典,再分词,再抽事件。先让机器认识“这个游戏里有哪些固定动作词”,然后让正则负责挑出动作词前后的数量和数值,最后按原句频次做热度过滤。

2. 搭建目录结构并准备样例数据

2.1 技术栈和版本确认

下面这套示例使用 Python 3.10 及以上版本。核心依赖只有一个分词库jieba,如果需要发送 Webhook,再安装requests

python --version pip install jieba==0.42.1 requests==2.32.3

如果原始材料没有给出明确版本,落地前要按当前运行环境确认依赖版本。jieba的接口已经很稳定,但项目进入生产环境前,还是要固定锁版本。

2.2 目录结构

建议用一个小目录把数据、词典、配置和主脚本分开,避免后续文件越来越多时无从下手。

md_insight/ ├── config.py ├── hotword_monitor.py ├── requirements.txt ├── data/ │ └── battle_texts.json ├── dict/ │ └── custom_dict.txt └── output/ └── report.txt

各文件作用:

文件作用
config.py路径、阈值、窗口参数集中管理
hotword_monitor.py主程序,负责读取、分析、生成报告
data/battle_texts.json比赛聊天文本的本地导出样本
dict/custom_dict.txt赛事黑话和选手名等自定义词典
output/report.txt生成的文本速报,先人工检查再用

2.3 准备一份模拟聊天数据

这里准备的数据文件是简化示例,字段只保留解析需要的核心信息。实际系统里还可以加入频道 ID、消息 ID、发送者身份、消息类型等,但最小版本不需要这些。

{ "version": "1", "source": "demo_export", "title": "【MD融合杯】喜欢我五个1850走脸吗?", "exported_at": "2025-07-20T21:30:00+08:00", "messages": [ { "time": "21:01:01", "text": "喜欢我五个1850走脸吗?" }, { "time": "21:01:03", "text": "喜欢我五个1850走脸吗?" }, { "time": "21:01:05", "text": "这波五连暴击直接结束" }, { "time": "21:01:08", "text": "五连暴击太离谱了" }, { "time": "21:01:10", "text": "对面又是三局翻盘" }, { "time": "21:01:12", "text": "连续三局翻盘合理吗" } ] }

需要强调一点:这里是演示使用本地导出的授权公开聊天记录,不代表鼓励收集私人聊天内容。接入真实平台时,必须确认数据来源符合平台规则和授权范围,避免未经同意抓取、存储和展示用户消息。

2.4 自定义词典

jieba默认词典处理日常文本足够,但遇到“走脸”“融合杯”这类社区词时,分词结果不一定符合预期。自定义词典采用三列格式:

词语 词频 词性

例如dict/custom_dict.txt

融合杯 50 n 走脸 30 v 翻盘 40 v 斩杀 40 v 暴击 40 v 连击 30 v 反杀 30 v 速攻 30 v 1850 60 m 1850 伤害 5 n

注意:同一词语不要重复只保留高优先级配置。第三列词性可以留空,但最少要有两列。

3. 主程序实现:清洗、分词和自定义词典

3.1 配置中心:把可变参数集中起来

新建config.py,用于放路径和业务参数。把阈值写在代码里虽然能跑,但后续比赛项目变大时会很难找到“为什么某条句子没有进报告”。

from pathlib import Path BASE_DIR = Path(__file__).resolve().parent DATA_PATH = BASE_DIR / "data" / "battle_texts.json" DICT_PATH = BASE_DIR / "dict" / "custom_dict.txt" OUTPUT_PATH = BASE_DIR / "output" / "report.txt" MIN_TOKEN_LEN = 1 MIN_EVENT_COUNT = 2

3.2 清洗函数要处理什么内容

弹幕和聊天文本通常带有 URL、@消息、HTML 标签、控制字符,甚至偶尔出现粘贴错乱的内容。直接进入分词,会把无意义字符也算进词频。清洗函数做的事情是:

  • 去掉<...>HTML 标签。
  • 去掉#话题#
  • 去掉@昵称
  • 去掉 URL。
  • 去掉不可见控制字符。
  • 把不适合分词的符号替换成空格,避免把句子挤在一起。
import re def clean_text(text: str) -> str: if not text: return "" text = re.sub(r"<[^>]+>", "", text) text = re.sub(r"#\S+", "", text) text = re.sub(r"@[\w\u4e00-\u9fa5_-]+", "", text) text = re.sub(r"https?://\S+", "", text) text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", "", text) # 保留中英文、数字和常用标点,其余先替换为空格 text = re.sub( r"[^\w\u4e00-\u9fa5!?。!??,,、::”“\"']", " ", text ) return text.strip()

这段代码里最重要的一行是最后的替换规则。它不会删除中文,只会把表情符号、特殊符号等转换为空格。这样后面正则匹配“数量 + 数值 + 动作”时,不会被大段表情干扰。

3.3 降噪和分词

聊天文本中“的、了、吗、我、你”这类词出现频率很高,但不可能成为战报标题的核心词。因此分词后需要一个过滤列表。

from collections import Counter from pathlib import Path import jieba STOPWORDS = { "的", "了", "吗", "呢", "吧", "啊", "我", "你", "他", "她", "它", "这", "那", "就是", "真的", "感觉", "怎么", "什么", "为什么", "还是", "已经", "一个", "我们", "你们", "他们", } def load_user_dict(dict_path: Path): if dict_path.exists(): jieba.load_userdict(str(dict_path)) def filter_tokens(words): result = [] for token in words: token = token.strip() if not token: continue if token in STOPWORDS: continue if len(token) == 1 and not token.isdigit(): continue result.append(token) return result def tokenize_sentence(sentence: str): return jieba.lcut(sentence, cut_all=False)

这里过滤掉单字但保留数字,因为1850是核心数值。如果赛事里还会出现“中单”“打野”这类双字及以上术语,自定义词典要优先覆盖这些词。

3.4 句子拆分

一条消息可能包含多个完整句子,例如:“这波五连暴击直接结束。下一把稳住。”如果当作一整句分析,事件抽取会把不同动作强行绑定在同一句话里。先按句号、感叹号、问号拆分。

def split_sentences(text: str): parts = re.split(r"[。!?!?\n]+", text) return [p.strip() for p in parts if p.strip()]

4. 事件抽取:用正则捕获“数量 + 数值 + 动作”

4.1 为什么用规则而不是直接训模型

示例文本规模很小,用深度学习模型反而会面临标注数据不足、训练周期长、解释成本高等问题。在早期验证阶段,最有效的方法是配置一张动作词表,再配合正则把动作周边的数量、数值找出来。

规则方案的优势是:

  • 可解释:输出结果可以反查命中的原句。
  • 可快速调整:运营说“连击也要算”,加一个词即可。
  • 改动成本低:不需要重新训练。

缺点是词表依赖人工维护。它适合“社区赛事热词监控”的第一版,不适合没有任何领域词表的通用舆情系统。

4.2 事件正则设计

首先定义需要识别的动作词。示例动作词放在列表里,后续可以在配置文件中扩展。

ACTION_TERMS = [ "走脸", "斩杀", "翻盘", "连击", "暴击", "反杀", "连败", "速攻", ] AR_NUM = r"(?:0|[1-9]\d*)" CN_NUM = r"(?:[零一二两三四五六七八九十百千万]+)" NUMBER = rf"(?:{AR_NUM}|{CN_NUM})" QUANT_UNIT = r"(?:个|局|次|种|张|只|名|位|轮|连|场)?" ACTION = "|".join(sorted(ACTION_TERMS, key=len, reverse=True)) EVENT_RE = re.compile( rf"(?P<quantity>{NUMBER}{QUANT_UNIT})?" rf"(?P<value>{NUMBER})?" rf"(?P<action>{ACTION})" )

正则解释:

  • quantity:解析如“五个”“1850”“三局”“五次”这类数量信息。
  • value:解析动作前紧邻的核心数值,例如“五个1850走脸”中的1850
  • action:解析动作词,如“走脸”“斩杀”。

这一版正则会偏宽,需要结合真实语料反复调整,但不能因此放弃规则。先跑通一条链路,比一开始追求完美模型更重要。

4.3 事件提取函数

def extract_events(sentence: str): events = [] for match in EVENT_RE.finditer(sentence): events.append( { "quantity": match.group("quantity") or "", "value": match.group("value") or "", "action": match.group("action"), "sentence": sentence, } ) return events

如果把五个1850走脸放入匹配器,预期输出类似:

{ "quantity": "五个", "value": "1850", "action": "走脸" }

但也会出现“昨天三次翻盘”这种句子被匹配出quantity=三次, action=翻盘的情况,这正是我们需要的。如果出现误匹配,比如“不要连败”会被识别出动作词“连败”,就需要在STOPWORDS或否定前缀层做二次过滤。

4.4 完整分析函数

把清洗、分词、事件抽取整合到一个analyze_messages函数中:

def analyze_messages(messages, dict_path: Path): load_user_dict(dict_path) token_counter = Counter() event_sentence_counter = Counter() events = [] for msg in messages: text = clean_text(msg.get("text", "")) if not text: continue for sentence in split_sentences(text): words = tokenize_sentence(sentence) for token in filter_tokens(words): token_counter[token] += 1 for event in extract_events(sentence): event["time"] = msg.get("time", "") events.append(event) event_sentence_counter[sentence] += 1 return token_counter, events, event_sentence_counter

这里把分词和事件抽取分开:分词用来做高频词统计,事件抽取用来识别结构化短语。两者各自独立,避免后续改动事件规则时把词频统计流程弄坏。

5. 生成速报并验证结果

5.1 模板输出函数

速报不要直接作为正式战报发出去,它只是运营的候选素材。输出内容要包含:

  • 时间段内的事件数量。
  • Top 3 动作词。
  • Top 5 高频词。
  • 出现次数最多的三句原始文本。
  • 用于人工核对的可疑数值。
def build_report(token_counter, events, event_sentence_counter): lines = [] lines.append("赛事热词速报") lines.append("=" * 30) lines.append(f"本轮共提取事件:{len(events)} 条") lines.append("") action_counter = Counter(e["action"] for e in events) lines.append("Top 5 动作词:") for action, count in action_counter.most_common(5): lines.append(f" {action}: {count}") lines.append("") lines.append("Top 8 高频词:") for token, count in token_counter.most_common(8): lines.append(f" {token}: {count}") lines.append("") lines.append("热门原句:") for sentence, count in event_sentence_counter.most_common(3): lines.append(f" {count} 次:{sentence}") return "\n".join(lines)

注意:不要在模板里自动断言“本次比赛热门操作是走脸”。如果只出现一条“走脸”句子,字数超过阈值也可能被标题化,误导读者。最简单的办法是把出现次数小于MIN_EVENT_COUNT的事件放在“待观察”区域,不进入 Top 列表。

5.2 主流程

import json from config import DATA_PATH, DICT_PATH, OUTPUT_PATH, MIN_EVENT_COUNT def load_messages(data_path: Path): with open(data_path, "r", encoding="utf-8") as f: data = json.load(f) return data.get("messages", []) def main(): messages = load_messages(DATA_PATH) token_counter, events, event_sentence_counter = analyze_messages( messages, DICT_PATH ) report = build_report(token_counter, events, event_sentence_counter) print(report) OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True) OUTPUT_PATH.write_text(report, encoding="utf-8") # TODO: 推送逻辑接入后,在这里调用 send_webhook(report) if __name__ == "__main__": main()

如果数据文件有 6 条示例消息,其中两条包含“走脸”,两条包含“暴击”,两条包含“翻盘”,输出会看到行走脸=2, 暴击=2, 翻盘=2,同时热门原句里能看到“喜欢我五个1850走脸吗”出现两次。

5.3 验证检查点

跑完脚本后,除了看有没有报错,还要做以下检查:

  1. 1850是否出现在高频词里。
  2. “走脸”是否被合并成一个词而不是拆成“走”和“脸”。
  3. 事件列表里是否同时出现quantity=五个value=1850action=走脸
  4. 事件原句是否保留了可读上下文。
  5. 如果事件计数为 0,优先检查词典路径和正则是否被转义错误影响。

6. 把接入机器人 Webhook,定时跑起来

6.1 Webhook 推送函数

在企业微信、钉钉、飞书里创建一个群机器人后,会得到一个 Webhook 地址。使用 Webhook 时不要把 URL 写在仓库里,而是通过环境变量注入。

import os import requests def send_webhook(text: str, webhook_url: str | None = None): url = webhook_url or os.getenv("EVENT_WEBHOOK_URL", "") if not url: print("Webhook URL 未配置,跳过推送") return 0 payload = { "msgtype": "text", "text": { "content": text } } resp = requests.post(url, json=payload, timeout=5) resp.raise_for_status() return resp.status_code

第一次接入时,建议先往自己的测试群推送,不要直接推到正式运营群。Webhook 一旦泄露,可能被外部调用刷消息,所以生产环境还要配合群机器人安全设置和 IP 白名单。

6.2 手动运行和定时运行

命令行直接运行:

cd md_insight python hotword_monitor.py

比赛期间需要每隔一段时间自动执行一次,最简单的做法是使用系统cron

*/10 * * * * cd /path/to/md_insight && /usr/bin/python3 hotword_monitor.py >> logs/cron.log 2>&1

在 Windows 开发机上可以使用任务计划程序设定触发周期。如果团队已经有 Celery、APScheduler 等调度系统,则把hotword_monitor.py改造成一个可调用任务,由调度中心统一控制。

6.3 环境变量配置示例

.env文件中可以写:

EVENT_WEBHOOK_URL=https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your_key

然后在程序中用python-dotenv加载,或者直接由部署平台注入环境变量。不要把真实 key 提交到 Git。

7. 常见问题与排查链路

7.1 现象速查表

下面的表格整理了这套系统最容易出现的问题:

问题现象常见原因检查方式处理建议
分词后“走脸”不见了用户词典未加载或词典路径错误打印jieba.lcut输出,检查自定义词典是否生效确认DICT_PATH存在,检查词典文件编码为 UTF-8
事件抽取总是匹配不到动作词列表不完整或正则表达式被业务符号干扰打印清洗后的句子,确认句子有空格或标点产生切割先跑一条最简句子,再逐步加入原始文本
高频词全是“喜欢”“真的”之类停用词列表覆盖不全查看filter_tokens结果补充停用词,但不要把所有情感词都停用
同一条内容重复推送没有做消息去重查看日志中相同 sentence 出现次数引入消息 ID 或句子哈希去重
数字被错误组合到动作“1850走脸”“1850 斩杀”解析过宽打印事件字典里的字段对 action 前文本长度设置上限,增加否定词处理

7.2 第一个大坑:词典没加载却以为规则写错

很多人改了custom_dict.txt,脚本没重启,或者jieba.load_userdict晚于第一次jieba.lcut执行,就会发现自定义词不生效。jieba在加载缓存后可能不会重新读取词典,改动词典后最好清理缓存,或在部署脚本里固定输出日志:

jieba.lcut("走脸")

如果返回["走脸"],说明词典生效。如果返回["走", "脸"],先检查加载路径和编码。

7.3 第二个大坑:把正则能否匹配当作交付标准

正则命中原句不等于业务事件正确。比如“他是连败还是连击”这句话,可能误匹配为动作=连败。规则只能解决常见模式,不能解决句法歧义。

处理方式是给事件增加“来源句子”字段,同时把低频事件单独归类。输出报告只给人工复核提供线索,不能直接让机器发布正式定论。

7.4 第三个大坑:忽略文本中的时间窗口直接做全量统计

聊天内容有明显时间波动,比赛开场、高潮、结束时的关键词分布完全不同。如果脚本每天只跑一次,统计的是全天累计词频,运营难以快速感知当前“正在发生什么”。

更合理的做法是先按消息时间过滤最近 5 分钟或 10 分钟的数据,再做词频统计。示例数据里的time字段就是为这个目的预留的。正式接入时,建议转成标准的 ISO 时间字符串,便于窗口过滤。

8. 从临时脚本到生产工具的补强清单

学习环境跑通之后,生产环境需要的不是把脚本做得更炫,而是把数据、日志、权限、可回溯性补齐。

8.1 数据结构化补强

弹幕原始文本只适合做初筛,不适合长期做报表。生产级系统建议增加一张事件表:

CREATE TABLE event_log ( id BIGINT PRIMARY KEY AUTO_INCREMENT, source

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询