本系列博客基于一个真实可运行的电商评论舆情分析项目。上一篇《第6篇FastAPI 长任务异步化实践从HTTP 阻塞到进程内 TaskManager》一、痛点开场舆情系统的价值在于第一时间发现风险差评率突然飙升、出现假货/曝光/监管部门等敏感词运营需要马上知道而不是等用户自己打开页面才发现。所以预警不能只是写进数据库等人看而是要规则命中时自动落库留痕、可追溯主动推送通知给负责人飞书/钉钉支持人工确认闭环处理完标记 closed这篇讲预警系统的实现三类规则引擎 Webhook 通知 通知器抽象。二、三类规则引擎backend/agents/alerting/nodes.py定义了默认规则DEFAULT_RULES{high_negative_rate:{enabled:True,threshold:0.20},# 负面率 20%spike_detection:{enabled:True,z_score_threshold:2.0,lookback_days:7},keyword_alert:{enabled:True,keywords:[投诉,举报,监管部门,曝光,骗,假货]},}# 命中即 critical 的高危词CRITICAL_KEYWORDS[监管部门,曝光,骗,假货,传销,非法]规则 1高负面率def_apply_high_negative_rate(rate,threshold):ifratethreshold:levelcriticalifrate0.40elsewarning# 40% 升级 criticalreturn{rule:high_negative_rate,level:level,title:f负面率异常{rate:.1%}超过阈值{threshold:.0%},description:f最近 24 小时内负面率{rate:.1%}建议关注并介入处理。,metric_value:round(rate,4),threshold:threshold,}returnNone规则 2突增检测def_apply_spike_detection(current_rate,historical_rate,threshold):ifhistorical_rateisNoneorhistorical_rate0:returnNone# 注意这里其实是相对变化比例不是严格 z-score见第 9 篇复盘change_ratio(current_rate-historical_rate)/historical_rateifchange_ratiothreshold:return{rule:spike_detection,level:warning,title:负面率突增,description:f负面率从{historical_rate:.1%}上升至{current_rate:.1%}变化{change_ratio:.1%}。,...}returnNone规则 3敏感词命中asyncdef_query_keyword_alerts(tenant_id,window_hours):# SELECT r.content, ra.sentiment_label FROM reviews r# JOIN review_analyses ra ON ra.review_id r.id# WHERE tenant_id:t AND review_time :cutoff AND (content ILIKE %投诉% OR ...)forcontent,review_time,sentimentinrows:matched_kw[kwforkwinkeywordsifkwincontent]is_criticalany(kwinCRITICAL_KEYWORDSforkwinmatched_kw)alerts.append({rule:keyword_alert,level:criticalifis_criticalelseinfo,title:f敏感词命中{/.join(matched_kw)},description:f评论包含敏感词{content[:100]}...,...})三、落库alert_eventssave_alerts_node把预警写入数据库asyncdefsave_alerts_node(state):alertsstate.get(alerts,[])ifnotalerts:return{structured_output:_build_stats(state)}asyncwithAsyncSessionLocal()assession:foralertinalerts:awaitsession.execute(text(INSERT INTO alert_events (tenant_id, level, rule, title, description, metric_value, threshold) VALUES (:tenant_id, :level, :rule, :title, :description, :metric, :threshold)),{...})awaitsession.commit()# 落库后再通知关键顺序notify_resultsawait_notify_channels(alerts)stats_build_stats(state)stats[notify]notify_resultsreturn{structured_output:stats}四、主动通知飞书/钉钉 Webhook通知器抽象backend/notifications/base.pydataclassclassNotificationResult:ok:boolchannel:strmessage:strerror:Optional[str]Nonehttp_status:Optional[int]NoneclassBaseNotifier:name:strbaseasyncdefsend(self,payload:dict)-NotificationResult:raiseNotImplementedErrorstaticmethoddefformat_alert_card(alert:dict)-dict:把预警 dict 格式化为通用卡片。level_emoji{critical:,warning:,info:}return{title:f{level_emoji.get(alert.get(level),)}{alert.get(title,舆情预警)},level:alert.get(level,info),rule:alert.get(rule,),description:alert.get(description,),metric_value:alert.get(metric_value),threshold:alert.get(threshold),}飞书通知器backend/notifications/feishu.pyclassFeishuNotifier(BaseNotifier):namefeishudef__init__(self,webhook_urlNone,timeout10.0):self.webhook_urlwebhook_urlorget_settings().feishu_webhook_url self.timeouttimeoutasyncdefsend(self,payload)-NotificationResult:ifnotself.webhook_url:returnNotificationResult(okFalse,channelself.name,errorFEISHU_WEBHOOK_URL 未配置)# 飞书交互式卡片header 颜色随级别变化card{msg_type:interactive,card:{header:{title:{tag:plain_text,content:payload.get(title,舆情预警)},template:(redifpayload.get(level)criticalelseorangeifpayload.get(level)warningelseblue),},elements:[{tag:div,text:{tag:lark_md,content:payload.get(description,)}},{tag:note,elements:[{tag:plain_text,content:f规则{payload.get(rule,-)}| 指标{payload.get(metric_value,-)}| 阈值{payload.get(threshold,-)}}]},],},}returnawaitself._post(card)钉钉通知器backend/notifications/dingtalk.pyclassDingTalkNotifier(BaseNotifier):namedingtalkasyncdefsend(self,payload)-NotificationResult:ifnotself.webhook_url:returnNotificationResult(okFalse,channelself.name,errorDINGTALK_WEBHOOK_URL 未配置)body{msgtype:markdown,markdown:{title:payload.get(title,舆情预警),text:(f##{payload.get(title)}\n\nf**级别**{payload.get(level)}\n\nf**规则**{payload.get(rule)}\n\nf**说明**{payload.get(description)}\n\nf 指标{payload.get(metric_value)}| 阈值{payload.get(threshold)}),},}returnawaitself._post(body)分渠道发送 失败不阻断asyncdef_notify_channels(alerts)-dict:results{feishu:[],dingtalk:[]}ifnotalerts:returnresultstry:feishuFeishuNotifier()dingDingTalkNotifier()foralertinalerts:cardBaseNotifier.format_alert_card(alert)# 只推 critical warninginfo 不打扰ifalert.get(level)in(critical,warning):iffeishu.webhook_url:rawaitfeishu.send(card)results[feishu].append({rule:alert.get(rule),ok:r.ok,error:r.error})ifding.webhook_url:rawaitding.send(card)results[dingtalk].append({rule:alert.get(rule),ok:r.ok,error:r.error})exceptExceptionase:logger.warning(alerting.notify_failed,errorstr(e)[:200])returnresults设计要点失败不阻断主流程通知挂了不影响落库用 try/except 包住分级推送只有 critical warning 走 webhookinfo 不打扰结果回传stats[notify]记录每个渠道的发送结果可观测五、调度与 API定时扫描APScheduler 每 6 小时执行一次_run_alert_scan手动扫描POST /api/v1/alerts/scan前端立即扫描按钮列表GET /api/v1/alerts?statusactive|closed确认POST /api/v1/alerts/confirm?alert_idxxx更新 statusclosed confirmed_by confirmed_at六、诚实边界它还不是完整通知系统无去重/抑制同一问题每 6 小时扫描都可能重复推送运营会被轰炸无通知记录表没有落库发给谁、是否成功无法追溯无签名校验钉钉 webhook 未做 secret 签名安全边界弱z-score名不副实突增检测实际是相对变化比例不是统计 z-score生产化方案预警业务键(tenant_id, rule, 资源维度, 时间窗口)唯一命中已有 active 预警则跳过抑制窗口同一预警 60 分钟内只推一次超时未解决自动升级重推新增notifications表记录每次发送的渠道/状态/错误钉钉加timestamp sign签名七、总结三类规则负面率、突增、敏感词覆盖主要风险信号落库 通知先留痕再推送失败不阻断通知器抽象BaseNotifier 飞书/钉钉两个实现可扩展诚实边界去重、抑制、记录表、签名是下一步下一篇预告《Vue3 TS 舆情看板ECharts 图表与前后端契约管理》
企业数字化 ERP 产品动态
相关推荐
胸肌上部怎么练饱满源码解析：面试突击避坑指南 胸肌上部怎么练饱满源码解析：面试突击避坑指南 报错一堆看不懂 StackTrace？别慌，这就像你练胸肌只练了中缝，上部空得能塞进拳头，看着就不专业。今天咱们不聊虚的，直接拆解【胸肌上部怎么练饱满】背后的逻辑，用【源码解析】的思路，把那些让… · 2026/9/22 22:14:19
韩语零基础入门实战：搞定高频面试题里的性能瓶颈 韩语零基础入门实战：搞定高频面试题里的性能瓶颈 你是不是也遇到过这种情况：从网上复制了一段韩语发音合成或文本处理的代码，本地跑起来报错一片，或者速度慢得像蜗牛，完全不知道该怎么调？别急，这不仅仅是韩语学习的问题，更是编程性能优化的经典场景。… · 2026/9/22 22:14:13
手写实现读书口诀避坑指南：3个血泪教训救你项目 手写实现读书口诀避坑指南：3个血泪教训救你项目 看了一堆教程还是不会写项目？别怪自己笨，是你没掌握“读书口诀”背后的手写实现逻辑。很多后端开发在重构业务代码时，习惯照抄文档里的示例，结果上线后数据错乱、接口超时，排查三天三夜才发现是核心算法… · 2026/9/22 22:14:00
烽火机顶盒开发避坑：从零搭建到最佳实践，彻底告别环境卡死 烽火机顶盒开发避坑：从零搭建到最佳实践，彻底告别环境卡死 配置环境就卡半天，是不是让你想砸键盘？别急，这是绝大多数开发者在接触【烽火机顶盒】定制开发时的真实痛点。很多人以为只要会写代码就能搞定，结果在交叉编译、驱动适配、系统裁剪上耗了半个月… · 2026/9/22 23:00:56
3步读懂压缩器源码解析 搞定项目搭建难题 3步读懂压缩器源码解析 搞定项目搭建难题 很多开发者卡在“语法会背，项目不会搭”的瓶颈期。你盯着文档里的 compress() 方法发呆，心里想：这底层到底是怎么把数据变小了？ 别急，今天咱们不整虚的，直接拆解【压缩器】的【源码解析】。… · 2026/9/22 23:00:50
袁辉实战：3步搞定源码解析，新手避坑指南 袁辉实战：3步搞定源码解析，新手避坑指南 刚学完 Python 或 Java 的语法，打开编辑器却像无头苍蝇？很多初学者都卡在“学会语法却不知怎么搭项目”这一步。别急，这不是你的错，而是缺少一个从理论到落地的桥梁。今天我们就通过袁辉这个实战… · 2026/9/22 23:00:43
3个步骤搞定丫丫项目搭建与源码解析 3个步骤搞定丫丫项目搭建与源码解析 刚毕业拿到 offer，或者准备跳槽面试，你是不是也卡在这个坎上？ 书上的语法都背熟了，LeetCode 刷题也顺手，但一让你从 0 到 1 搭个项目，脑子就一片空白。… · 2026/9/22 23:00:30
图解原理：3步搞定如何设置电脑开机密码防黑客 图解原理：3步搞定如何设置电脑开机密码防黑客 版本升级后 API 全变了？别慌，这次我们拆解最底层的逻辑。很多开发者习惯用 sudo 一把梭，却忘了 如何设置电脑开机密码… · 2026/9/22 23:00:17
5个新手避坑指南：搞定ps学习软件，告别API变更焦虑 5个新手避坑指南：搞定ps学习软件，告别API变更焦虑 版本升级后 API 全变了，这是无数开发者在接触 ps学习软件 相关前端交互时最真实的噩梦。刚写好的代码，换个版本直接报错，断点调试半天发现接口签名都换了。对于刚入行的新人来说，这种“… · 2026/9/22 23:00:10
5个电影海报图片处理坑，新手避坑指南 5个电影海报图片处理坑，新手避坑指南 刚写完代码，一运行屏幕直接炸了。满屏红色的 StackTrace 滚得比弹幕还快，什么 NullPointerException 、 ImageIO.read() returned null 、… · 2026/9/22 0:00:07
注册微信公众账号：一文搞懂从0到1全流程 注册微信公众账号：一文搞懂从0到1全流程 复制来的代码跑不通，报错信息满屏飞，到底卡在哪？别急，咱们先停下手里的调试。很多开发者觉得注册微信公众账号只是填个表单、传个身份证那么简单，真上手才发现坑深不见底。今天这篇 一文搞懂… · 2026/9/22 0:00:07
