优化:企微群聊通知从3条精简为2条 + 卡片跳转详情页

群聊通知改为:
1. image 截图(可点击放大查看)
2. markdown 告警详情 + @人员(不再重复发图)

私发卡片 card_action.url 改为跳转前端告警详情页
news 图文卡片去除默认企微官网链接

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-03-19 09:21:45 +08:00
parent a3797e7508
commit e14450edbc

View File

@@ -38,6 +38,7 @@ class WeChatService:
self._encoding_aes_key = ""
self._access_token = ""
self._token_expire_at = 0
self._service_base_url = ""
# 缓存 response_code用于更新卡片状态
# key: task_id (alarm_id), value: response_code
self._response_codes: Dict[str, str] = {}
@@ -50,6 +51,8 @@ class WeChatService:
self._secret = config.secret
self._token = config.token
self._encoding_aes_key = config.encoding_aes_key
self._service_base_url = getattr(config, "service_base_url", "").rstrip("/")
self._group_chat_id = getattr(config, "group_chat_id", "")
if self._enabled:
logger.info(f"企微通知服务已启用: corp_id={self._corp_id}")
@@ -64,6 +67,12 @@ class WeChatService:
def agent_id_int(self) -> int:
return int(self._agent_id) if self._agent_id else 0
def _alarm_detail_url(self, alarm_id: str) -> str:
"""构造告警详情页 URL"""
if self._service_base_url:
return f"{self._service_base_url}/aiot/alarm/list?alarmId={alarm_id}"
return ""
async def _get_access_token(self) -> str:
"""获取企微 access_token带缓存"""
if self._access_token and time.time() < self._token_expire_at - 60:
@@ -236,7 +245,7 @@ class WeChatService:
],
"card_action": {
"type": 1,
"url": "https://work.weixin.qq.com",
"url": self._alarm_detail_url(alarm_id) or "https://work.weixin.qq.com",
},
"button_list": [
{
@@ -548,8 +557,9 @@ class WeChatService:
article = {
"title": title,
"description": description,
"url": url or "https://work.weixin.qq.com",
}
if url:
article["url"] = url
if picurl:
article["picurl"] = picurl
@@ -585,11 +595,10 @@ class WeChatService:
mention_user_ids: Optional[List[str]] = None,
) -> bool:
"""
发送告警组合消息到群聊(3条消息)
发送告警通知到群聊(2条消息)
1. image: 告警截图
2. news: 告警详情图文卡片
3. text: @相关人员提醒处理
1. image: 告警截图(可点击放大查看)
2. markdown: 告警详情 + @人员
"""
if not self._enabled:
return False
@@ -598,7 +607,7 @@ class WeChatService:
level_name = ALARM_LEVEL_NAMES.get(alarm_level, "普通")
success = True
# ---- 1. 发送告警截图image 消息) ----
# ---- 1. 发送告警截图image 消息,可点击放大 ----
if snapshot_url:
media_id = await self.upload_media_from_url(snapshot_url)
if media_id:
@@ -609,35 +618,26 @@ class WeChatService:
else:
logger.warning(f"截图上传企微失败,跳过图片消息: alarm={alarm_id}")
# ---- 2. 发送告警详情news 图文卡片 ----
news_title = f"{level_name}{type_name}告警"
news_desc = (
f"{description}\n\n"
f"告警区域:{area_name or '未知区域'}\n"
f"摄像头:{camera_name or '未知'}\n"
f"告警时间:{event_time}\n"
f"告警ID{alarm_id}"
)
# news 卡片的 picurl 可用 COS 预签名 URL缩略图
sent = await self.send_group_news(
chat_id=chat_id,
title=news_title,
description=news_desc,
picurl=snapshot_url if snapshot_url and snapshot_url.startswith("http") else "",
# ---- 2. 告警详情 + @人员markdown 消息 ----
mention_text = ""
if mention_user_ids:
mentions = " ".join(f"<@{uid}>" for uid in mention_user_ids)
mention_text = f"\n{mentions} 请及时处理"
md_content = (
f"**【{level_name}{type_name}告警**\n"
f">告警区域:<font color=\"info\">{area_name or '未知区域'}</font>\n"
f">摄像头:<font color=\"info\">{camera_name or '未知'}</font>\n"
f">告警时间:<font color=\"warning\">{event_time}</font>\n"
f">描述:{description}"
f"{mention_text}"
)
sent = await self.send_group_markdown(chat_id, md_content)
if not sent:
success = False
# ---- 3. @相关人员markdown 消息,@渲染更可靠) ----
if mention_user_ids:
mentions = " ".join(f"<@{uid}>" for uid in mention_user_ids)
md_content = f"{mentions} 请及时处理以上**{type_name}告警**"
sent = await self.send_group_markdown(chat_id, md_content)
if not sent:
success = False
if success:
logger.info(f"群聊组合消息已发送: alarm={alarm_id}, chatid={chat_id}")
logger.info(f"群聊告警通知已发送: alarm={alarm_id}, chatid={chat_id}")
return success