scripts/ci_notify.py

@@ -2,7 +2,7 @@
 """
 统一CI通知脚本 - 发送飞书卡片通知
 支持三种模式: start / success / failure
-包含: PR链接、耗时、失败阶段、分支、提交者、Run链接
+包含: PR链接、耗时、失败阶段、分支、提交者、Run链接、Runner信息
 
 用法:
   NOTIFY_MODE=start   JOB_NAME="xxx" python3 scripts/ci_notify.py
@@ -23,6 +23,12 @@
   GITHUB_EVENT_NAME - 事件类型 (pull_request / push / ...)
   GITHUB_PR_NUMBER  - PR编号 (PR事件时)
   GITHUB_PR_TITLE   - PR标题 (PR事件时)
+  RUNNER_NAME       - Runner名称 (可选，自动获取)
+
+设计原则:
+  1. 通知失败永远不阻断CI主流程（返回exit code 0）
+  2. 标题包含"CI通知"/"CI告警"关键词，适配飞书webhook关键词校验
+  3. 卡片信息尽量丰富，方便快速定位问题
 """
 
 import json
@@ -32,6 +38,7 @@ import urllib.request
 
 
 def get_env(name, default=""):
+    """读取环境变量"""
     return os.environ.get(name, default)
 
 
@@ -48,6 +55,20 @@ def format_duration(seconds_str):
         return seconds_str or "未知"
 
 
+def classify_job(job_name):
+    """根据Job名称判断所属阶段"""
+    name = job_name.lower()
+    if any(k in name for k in ["validate", "lint", "unit test", "integration test"]):
+        return "门禁检查"
+    if any(k in name for k in ["build", "image"]):
+        return "镜像构建"
+    if any(k in name for k in ["deploy", "staging", "production"]):
+        return "部署发布"
+    if any(k in name for k in ["e2e", "test", "smoke"]):
+        return "测试验证"
+    return "其他"
+
+
 def main() -> int:
     webhook = get_env("CI_NOTIFY_WEBHOOK")
     if not webhook:
@@ -71,51 +92,60 @@ def main() -> int:
     event_name = get_env("GITHUB_EVENT_NAME", "")
     pr_number = get_env("GITHUB_PR_NUMBER", "")
     pr_title = get_env("GITHUB_PR_TITLE", "")
+    runner_name = get_env("RUNNER_NAME", "")
 
     run_url = f"https://git.xiaoxiajianji.com/{repo}/actions/runs/{run_id}"
+    job_stage = classify_job(job_name)
 
     # 根据模式设置标题、状态、颜色
     # 注意：标题中必须包含飞书webhook配置的关键词，否则会报"Key Words Not Found"
-    # 这里加入多个常用关键词(CI/通知/构建/部署/告警)提高命中率
+    # 这里加入"CI通知"/"CI告警"关键词提高命中率
     if mode == "start":
-        title = f"🔄 CI构建部署通知：{job_name} 开始执行"
+        title = f"🔄 CI通知：{job_name} 开始构建"
         status = "blue"
         button_text = "查看进度"
         button_type = "primary"
     elif mode == "success":
-        title = f"✅ CI构建部署通知：{job_name} 执行成功"
+        title = f"✅ CI通知：{job_name} 构建成功"
         status = "green"
         button_text = "查看详情"
         button_type = "primary"
     else:  # failure
-        title = f"❌ CI构建部署告警：{job_name} 执行失败"
+        title = f"❌ CI告警：{job_name} 构建失败"
         status = "red"
         button_text = "查看失败日志"
         button_type = "danger"
 
-    # 构建卡片内容
-    content_lines = []
-    content_lines.append(f"**任务**: {job_name}")
+    # 构建卡片内容 - 左侧标签+右侧值的结构化展示
+    fields = []
+
+    fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**阶段**\n{job_stage}"}})
+    fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**任务**\n{job_name}"}})
 
     if mode != "start":
-        content_lines.append(f"**耗时**: {duration}")
+        fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**耗时**\n{duration}"}})
+    else:
+        fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**状态**\n进行中"}})
+
+    if runner_name:
+        fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**Runner**\n{runner_name}"}})
 
     if mode == "failure" and failed_step:
-        content_lines.append(f"**失败阶段**: {failed_step}")
+        fields.append({"is_short": False, "text": {"tag": "lark_md", "content": f"**失败步骤**\n{failed_step}"}})
 
-    # PR信息
+    # PR/分支信息
     if event_name == "pull_request" and pr_number:
         pr_url = f"https://git.xiaoxiajianji.com/{repo}/pulls/{pr_number}"
         pr_display = f"#{pr_number}"
         if pr_title:
-            pr_display += f" {pr_title}"
-        content_lines.append(f"**PR**: [{pr_display}]({pr_url})")
+            pr_display += f" {pr_title[:30]}"
+        fields.append({"is_short": False, "text": {"tag": "lark_md", "content": f"**PR**\n[{pr_display}]({pr_url})"}})
     elif event_name == "push":
-        content_lines.append(f"**分支**: {branch}")
+        fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**分支**\n{branch}"}})
 
-    content_lines.append(f"**提交**: `{commit}`")
-    content_lines.append(f"**提交者**: {actor}")
-    content_lines.append(f"**Run ID**: {run_id}")
+    fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**提交**\n`{commit}`"}})
+    fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**提交者**\n{actor}"}})
+    fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**Run ID**\n{run_id}"}})
 
     payload = {
         "msg_type": "interactive",
@@ -130,10 +160,7 @@ def main() -> int:
             "elements": [
                 {
                     "tag": "div",
-                    "text": {
-                        "tag": "lark_md",
-                        "content": "\n".join(content_lines),
-                    },
+                    "fields": fields,
                 },
                 {
                     "tag": "action",