#!/usr/bin/env python3 """ 统一CI通知脚本 - 发送飞书卡片通知 支持三种模式: start / success / failure 包含: PR链接、耗时、失败阶段、分支、提交者、Run链接、Runner信息 用法: NOTIFY_MODE=start JOB_NAME="xxx" python3 scripts/ci_notify.py NOTIFY_MODE=success JOB_NAME="xxx" JOB_DURATION="2m30s" python3 scripts/ci_notify.py NOTIFY_MODE=failure JOB_NAME="xxx" FAILED_STEP="xxx" JOB_DURATION="2m30s" python3 scripts/ci_notify.py 环境变量: CI_NOTIFY_WEBHOOK - 飞书webhook地址 (必填) NOTIFY_MODE - 通知模式: start / success / failure (必填) JOB_NAME - Job名称 (必填) JOB_DURATION - 耗时，如"2m30s" (成功/失败时建议传) FAILED_STEP - 失败的步骤名 (失败时建议传) GITHUB_REF_NAME - 分支名 GITHUB_SHA - commit SHA GITHUB_ACTOR - 提交者 GITHUB_RUN_ID - Run ID GITHUB_REPOSITORY - 仓库路径 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 import os import sys import urllib.request def get_env(name, default=""): """读取环境变量""" return os.environ.get(name, default) def format_duration(seconds_str): """将秒数格式化为易读形式""" try: seconds = int(float(seconds_str)) mins = seconds // 60 secs = seconds % 60 if mins > 0: return f"{mins}m{secs}s" return f"{secs}s" except (ValueError, TypeError): 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: print("未配置 CI_NOTIFY_WEBHOOK，跳过通知") print("如需启用，请在仓库 Settings -> Secrets and variables -> Actions 中添加 CI_NOTIFY_WEBHOOK") return 0 mode = get_env("NOTIFY_MODE", "failure").lower() job_name = get_env("JOB_NAME", "Unknown Job") duration = get_env("JOB_DURATION") if not duration: duration_sec = get_env("JOB_DURATION_SECONDS") duration = format_duration(duration_sec) if duration_sec else "计算中..." failed_step = get_env("FAILED_STEP", "") branch = get_env("GITHUB_REF_NAME", "unknown") commit = get_env("GITHUB_SHA", "unknown")[:8] actor = get_env("GITHUB_ACTOR", "unknown") run_id = get_env("GITHUB_RUN_ID", "unknown") repo = get_env("GITHUB_REPOSITORY", "unknown") 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告警"关键词提高命中率 if mode == "start": title = f"🔄 CI通知：{job_name} 开始构建" status = "blue" button_text = "查看进度" button_type = "primary" elif mode == "success": title = f"✅ CI通知：{job_name} 构建成功" status = "green" button_text = "查看详情" button_type = "primary" else: # failure title = f"❌ CI告警：{job_name} 构建失败" status = "red" button_text = "查看失败日志" button_type = "danger" # 构建卡片内容 - 左侧标签+右侧值的结构化展示 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": fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**耗时**\n{duration}"}}) else: fields.append({"is_short": True, "text": {"tag": "lark_md", "content": "**状态**\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: fields.append({"is_short": False, "text": {"tag": "lark_md", "content": f"**失败步骤**\n{failed_step}"}}) # 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[:30]}" fields.append({"is_short": False, "text": {"tag": "lark_md", "content": f"**PR**\n[{pr_display}]({pr_url})"}}) elif event_name == "push": fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**分支**\n{branch}"}}) 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", "card": { "header": { "title": { "tag": "plain_text", "content": title, }, "status": status, }, "elements": [ { "tag": "div", "fields": fields, }, { "tag": "action", "actions": [ { "tag": "button", "text": {"tag": "plain_text", "content": button_text}, "url": run_url, "type": button_type, } ], }, ], }, } data = json.dumps(payload).encode("utf-8") req = urllib.request.Request( webhook, data=data, headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=10) as resp: resp_body = resp.read().decode("utf-8") # 飞书返回code=0表示成功 try: result = json.loads(resp_body) if result.get("code", 0) != 0: print(f"通知发送告警: 飞书返回错误 - {result.get('msg', resp_body)}", file=sys.stderr) print(f"通知已发送 ({mode}) - 飞书返回非0，但不阻断CI流程") else: print(f"通知已发送 ({mode})") except json.JSONDecodeError: print(f"通知已发送 ({mode})") except Exception as e: print(f"通知发送告警: {e}", file=sys.stderr) # 通知无论成功失败都不阻断CI主流程，统一返回0 return 0 if __name__ == "__main__": sys.exit(main())