本系列博客基于一个真实可运行的电商评论舆情分析项目。上一篇《第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 图表与前后端契约管理》 第7篇：舆情预警系统：三类规则引擎 + 飞书/钉钉 Webhook 通知
本系列博客基于一个真实可运行的电商评论舆情分析项目。上一篇《第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 图表与前后端契约管理》