xiaoxia-saas/scripts/ci_notify.py

#!/usr/bin/env python3
"""
统一CI通知脚本 - 发送飞书卡片通知
支持三种模式: start / success / failure
包含: PR链接、耗时、失败阶段、分支、提交者、Run链接

用法:
  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事件时)
"""

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 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", "")

    run_url = f"https://git.xiaoxiajianji.com/{repo}/actions/runs/{run_id}"

    # 根据模式设置标题、状态、颜色
    # 注意：标题中必须包含飞书webhook配置的关键词，否则会报"Key Words Not Found"
    # 这里加入多个常用关键词(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"

    # 构建卡片内容
    content_lines = []
    content_lines.append(f"**任务**: {job_name}")

    if mode != "start":
        content_lines.append(f"**耗时**: {duration}")

    if mode == "failure" and failed_step:
        content_lines.append(f"**失败阶段**: {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}"
        content_lines.append(f"**PR**: [{pr_display}]({pr_url})")
    elif event_name == "push":
        content_lines.append(f"**分支**: {branch}")

    content_lines.append(f"**提交**: `{commit}`")
    content_lines.append(f"**提交者**: {actor}")
    content_lines.append(f"**Run ID**: {run_id}")

    payload = {
        "msg_type": "interactive",
        "card": {
            "header": {
                "title": {
                    "tag": "plain_text",
                    "content": title,
                },
                "status": status,
            },
            "elements": [
                {
                    "tag": "div",
                    "text": {
                        "tag": "lark_md",
                        "content": "\n".join(content_lines),
                    },
                },
                {
                    "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())