赛事运营团队在复盘一场社区杯赛时经常要面对两类数据一类是比分、排名、击杀数这类结构化赛果另一类是弹幕、标题、群聊速报里的观众文本。比如“【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 jieba0.42.1 requests2.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:0008: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 23.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(rhttps?://\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_allFalse)这里过滤掉单字但保留数字因为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, keylen, reverseTrue)) EVENT_RE re.compile( rf(?Pquantity{NUMBER}{QUANT_UNIT})? rf(?Pvalue{NUMBER})? rf(?Paction{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, encodingutf-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(parentsTrue, exist_okTrue) OUTPUT_PATH.write_text(report, encodingutf-8) # TODO: 推送逻辑接入后在这里调用 send_webhook(report) if __name__ __main__: main()如果数据文件有 6 条示例消息其中两条包含“走脸”两条包含“暴击”两条包含“翻盘”输出会看到行走脸2, 暴击2, 翻盘2同时热门原句里能看到“喜欢我五个1850走脸吗”出现两次。5.3 验证检查点跑完脚本后除了看有没有报错还要做以下检查1850是否出现在高频词里。“走脸”是否被合并成一个词而不是拆成“走”和“脸”。事件列表里是否同时出现quantity五个、value1850、action走脸。事件原句是否保留了可读上下文。如果事件计数为 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, jsonpayload, timeout5) 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 21在 Windows 开发机上可以使用任务计划程序设定触发周期。如果团队已经有 Celery、APScheduler 等调度系统则把hotword_monitor.py改造成一个可调用任务由调度中心统一控制。6.3 环境变量配置示例在.env文件中可以写EVENT_WEBHOOK_URLhttps://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour_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 赛事热词监控：用Python从弹幕提取高频词与梗句
赛事运营团队在复盘一场社区杯赛时经常要面对两类数据一类是比分、排名、击杀数这类结构化赛果另一类是弹幕、标题、群聊速报里的观众文本。比如“【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 jieba0.42.1 requests2.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:0008: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 23.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(rhttps?://\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_allFalse)这里过滤掉单字但保留数字因为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, keylen, reverseTrue)) EVENT_RE re.compile( rf(?Pquantity{NUMBER}{QUANT_UNIT})? rf(?Pvalue{NUMBER})? rf(?Paction{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, encodingutf-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(parentsTrue, exist_okTrue) OUTPUT_PATH.write_text(report, encodingutf-8) # TODO: 推送逻辑接入后在这里调用 send_webhook(report) if __name__ __main__: main()如果数据文件有 6 条示例消息其中两条包含“走脸”两条包含“暴击”两条包含“翻盘”输出会看到行走脸2, 暴击2, 翻盘2同时热门原句里能看到“喜欢我五个1850走脸吗”出现两次。5.3 验证检查点跑完脚本后除了看有没有报错还要做以下检查1850是否出现在高频词里。“走脸”是否被合并成一个词而不是拆成“走”和“脸”。事件列表里是否同时出现quantity五个、value1850、action走脸。事件原句是否保留了可读上下文。如果事件计数为 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, jsonpayload, timeout5) 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 21在 Windows 开发机上可以使用任务计划程序设定触发周期。如果团队已经有 Celery、APScheduler 等调度系统则把hotword_monitor.py改造成一个可调用任务由调度中心统一控制。6.3 环境变量配置示例在.env文件中可以写EVENT_WEBHOOK_URLhttps://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour_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