xiaoxia-saas/scripts/ci_notify.py

#!/usr/bin/env python3
"""
CI 通知脚本 - 发送飞书消息通知
用法:
  NOTIFY_MODE=start JOB_NAME="Build API" python3 scripts/ci_notify.py
  NOTIFY_MODE=success JOB_NAME="Deploy Staging" python3 scripts/ci_notify.py
  NOTIFY_MODE=failure JOB_NAME="Unit Tests" python3 scripts/ci_notify.py

环境变量:
  NOTIFY_MODE     - 通知类型: start/success/failure
  JOB_NAME        - Job名称
  CI_NOTIFY_WEBHOOK - 飞书Webhook地址
  GITHUB_SHA      - Commit SHA (可选)
  GITHUB_REF_NAME - 分支名 (可选)
  GITHUB_RUN_ID   - Run ID (可选)
  GITHUB_REPOSITORY - 仓库名 (可选)
  GITHUB_SERVER_URL - Gitea地址 (可选)

设计原则:
  通知失败永远不阻断主流程（永远返回0）
"""

import json
import os
import sys
import urllib.request
from datetime import datetime


def get_env(name, default=""):
    return os.environ.get(name, default)


def send_feishu_notify(webhook_url, title, content, color="blue"):
    """发送飞书通知（简单卡片格式）"""
    if not webhook_url:
        print("[INFO] 未配置 CI_NOTIFY_WEBHOOK，跳过通知")
        return True

    # 状态颜色映射
    color_map = {
        "green": "green",
        "red": "red",
        "blue": "blue",
        "yellow": "yellow",
    }
    header_color = color_map.get(color, "blue")

    # 构造卡片
    card = {
        "config": {"wide_screen_mode": True},
        "header": {
            "title": {"tag": "plain_text", "content": title},
            "template": header_color,
        },
        "elements": [
            {"tag": "div", "text": {"tag": "lark_md", "content": content}},
        ],
    }

    payload = {"msg_type": "interactive", "card": card}
    data = json.dumps(payload).encode("utf-8")

    req = urllib.request.Request(
        webhook_url,
        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")
        result = json.loads(resp_body)
        if result.get("code", 0) != 0:
            print(f"[WARN] 飞书通知返回错误: {result.get('msg', resp_body)}", file=sys.stderr)
            return False
        print("[INFO] 飞书通知发送成功")
        return True
    except Exception as e:
        print(f"[WARN] 飞书通知发送失败: {e}", file=sys.stderr)
        return False


def build_message():
    """根据环境变量构造通知消息"""
    notify_mode = get_env("NOTIFY_MODE", "info").lower()
    job_name = get_env("JOB_NAME", "未知Job")
    branch = get_env("GITHUB_REF_NAME", "未知分支")
    sha = get_env("GITHUB_SHA", "")[:8]
    run_id = get_env("GITHUB_RUN_ID", "")
    repo = get_env("GITHUB_REPOSITORY", "")
    server_url = get_env("GITHUB_SERVER_URL", "https://git.xiaoxiajianji.com")

    # 状态映射
    status_map = {
        "start": ("🔔 CI 任务开始", "blue", "开始执行"),
        "success": ("✅ CI 任务成功", "green", "执行成功"),
        "failure": ("❌ CI 任务失败", "red", "执行失败"),
        "info": ("ℹ️ CI 通知", "blue", "通知"),
    }
    title, color, status_text = status_map.get(notify_mode, status_map["info"])

    # 构造内容
    content_lines = [
        f"**任务**: {job_name}",
        f"**状态**: {status_text}",
        f"**分支**: {branch}",
    ]
    if sha:
        content_lines.append(f"**Commit**: `{sha}`")
    if run_id and repo and server_url:
        run_url = f"{server_url}/{repo}/actions/runs/{run_id}"
        content_lines.append(f"**详情**: [点击查看]({run_url})")

    content_lines.append(f"**时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

    content = "\n".join(content_lines)
    return title, content, color


def main():
    webhook = get_env("CI_NOTIFY_WEBHOOK", "")
    title, content, color = build_message()

    print(f"[CI Notify] 模式: {get_env('NOTIFY_MODE')}")
    print(f"[CI Notify] 任务: {get_env('JOB_NAME')}")

    send_feishu_notify(webhook, title, content, color)

    # 永远返回0，不阻断主流程
    return 0


if __name__ == "__main__":
    sys.exit(main())