xiaoxia-saas/scripts/ci_notify_failure.py

#!/usr/bin/env python3
"""发送 CI 失败通知到飞书/项目群 webhook（增强版：带失败诊断）。

诊断功能：自动分析失败原因，给出分类和修复建议。
"""

import json
import os
import subprocess
import sys
import urllib.request


def run_diagnosis() -> dict:
    """运行失败诊断脚本，返回诊断结果"""
    diag_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ci/ci_failure_diagnosis.py")
    if not os.path.exists(diag_script):
        diag_script = "scripts/ci/ci_failure_diagnosis.py"

    result = {
        "category": "unknown",
        "category_cn": "未知",
        "severity": "medium",
        "summary": "",
        "error_lines": [],
        "suggestions": [],
        "auto_fixable": False,
    }

    try:
        # 运行诊断脚本
        env = os.environ.copy()
        env["DIAGNOSIS_OUTPUT"] = "/tmp/ci_diagnosis_result.json"

        proc = subprocess.run([sys.executable, diag_script], capture_output=True, text=True, timeout=30, env=env)

        # 尝试读取结果文件
        output_file = "/tmp/ci_diagnosis_result.json"
        if os.path.exists(output_file):
            with open(output_file) as f:
                result = json.load(f)
        elif proc.stdout:
            # 从stdout解析
            pass
    except Exception as e:
        print(f"诊断脚本执行失败: {e}", file=sys.stderr)

    return result


def main() -> int:
    webhook = os.environ.get("CI_NOTIFY_WEBHOOK", "")
    if not webhook:
        print("未配置 CI_NOTIFY_WEBHOOK，跳过通知")
        return 0

    failed_job = os.environ.get("FAILED_JOB", "Unknown Job")
    branch = os.environ.get("GITHUB_REF_NAME", "unknown")
    commit = os.environ.get("GITHUB_SHA", "unknown")[:8]
    actor = os.environ.get("GITHUB_ACTOR", "unknown")
    run_id = os.environ.get("GITHUB_RUN_ID", "unknown")
    repo = os.environ.get("GITHUB_REPOSITORY", "unknown")
    run_url = f"https://git.xiaoxiajianji.com/{repo}/actions/runs/{run_id}"
    pr_number = os.environ.get("PR_NUMBER", "")

    # 运行诊断
    diagnosis = run_diagnosis()

    # 构建卡片内容
    severity_color = {"high": "red", "medium": "orange", "low": "blue"}
    card_status = severity_color.get(diagnosis.get("severity", "medium"), "red")

    # 标题
    title = f"❌ CI失败 - {diagnosis.get('category_cn', '未知')}"

    # 诊断部分
    diag_lines = []
    diag_lines.append(f"**任务**: {failed_job}")
    diag_lines.append(f"**分类**: {diagnosis.get('category_cn', '未知')}")
    if diagnosis.get("summary"):
        diag_lines.append(f"**问题**: {diagnosis['summary']}")

    # 错误行
    error_lines = diagnosis.get("error_lines", [])
    if error_lines:
        diag_lines.append("")
        diag_lines.append("**关键错误**:")
        for err in error_lines[:3]:
            if len(err) > 100:
                err = err[:97] + "..."
            diag_lines.append(f"`{err}`")

    # 修复建议
    suggestions = diagnosis.get("suggestions", [])
    if suggestions:
        diag_lines.append("")
        diag_lines.append("**修复建议**:")
        for i, s in enumerate(suggestions[:3], 1):
            diag_lines.append(f"{i}. {s}")

    if diagnosis.get("auto_fixable"):
        diag_lines.append("")
        diag_lines.append("💡 *可自动修复的问题，试试Rerun*")

    # 基本信息
    info_lines = [
        f"**分支**: {branch}",
        f"**提交**: `{commit}`",
        f"**提交者**: {actor}",
    ]
    if pr_number:
        info_lines.append(f"**PR**: #{pr_number}")

    elements = [
        {
            "tag": "div",
            "text": {
                "tag": "lark_md",
                "content": "\n".join(diag_lines),
            },
        },
        {
            "tag": "hr",
        },
        {
            "tag": "div",
            "text": {
                "tag": "lark_md",
                "content": "\n".join(info_lines),
            },
        },
        {
            "tag": "action",
            "actions": [
                {
                    "tag": "button",
                    "text": {"tag": "plain_text", "content": "查看失败日志"},
                    "url": run_url,
                    "type": "danger",
                },
            ],
        },
    ]

    payload = {
        "msg_type": "interactive",
        "card": {
            "header": {
                "title": {
                    "tag": "plain_text",
                    "content": title,
                },
                "status": card_status,
            },
            "elements": elements,
        },
    }

    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.read()
        print("通知已发送（带诊断信息）")
    except Exception as e:
        print(f"通知发送失败: {e}", file=sys.stderr)
        return 1

    return 0


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

# trigger CI - bypass [ci skip] bug