fw-builder/notifier.py at master · AutomagicCode/fw-builder · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
"""
notifier.py — 钉钉 / 企微机器人通知
"""
import json
import logging
import urllib.request
import urllib.error

log = logging.getLogger("notifier")


def send_notification(config: dict, task: dict):
    """根据配置发送通知"""
    notif_type = config.get("type", "dingtalk")

    if notif_type == "dingtalk":
        _send_dingtalk(config.get("webhook", ""), task, config)
    elif notif_type == "wechat":
        _send_wechat(config.get("webhook", ""), task, config)
    else:
        log.warning(f"Unknown notification type: {notif_type}")


def _send_dingtalk(webhook: str, task: dict, config: dict):
    """发钉钉机器人消息"""
    if not webhook:
        return

    status_emoji = "✅" if task["status"] == "success" else "❌"
    title = f"固件构建{task['status']}"

    lines = [
        f"## {status_emoji} {title}",
        f"- **工序链**: {task.get('pipeline_name', '')}",
        f"- **任务ID**: {task.get('id', '')}",
        f"- **耗时**: {task.get('duration_sec', 0)}s",
        f"- **步骤数**: {len(task.get('steps', []))}",
    ]

    for i, s in enumerate(task.get("steps", [])):
        icon = "✅" if s.get("exit_code") == 0 else "❌"
        lines.append(
            f"- **步骤{i+1} [{s.get('name','')}]**: {icon} "
            f"(rc={s.get('exit_code',-1)}, {s.get('duration_sec',0)}s)"
        )

    if task.get("output_files"):
        lines.append("- **产物**:")
        for f in task["output_files"]:
            lines.append(f"  - `{f}`")

    markdown = {
        "msgtype": "markdown",
        "markdown": {
            "title": title,
            "text": "\n".join(lines),
        },
    }

    _post_json(webhook, markdown)


def _send_wechat(webhook: str, task: dict, config: dict):
    """发企微机器人消息"""
    if not webhook:
        return

    status_text = "成功" if task["status"] == "success" else "失败"
    content = (
        f"【固件构建{status_text}】\n"
        f"工序链: {task.get('pipeline_name', '')}\n"
        f"任务ID: {task.get('id', '')}\n"
        f"耗时: {task.get('duration_sec', 0)}s\n"
    )
    for i, s in enumerate(task.get("steps", [])):
        rc = s.get("exit_code", -1)
        content += f"步骤{i+1}[{s.get('name','')}]: rc={rc}\n"

    msg = {
        "msgtype": "text",
        "text": {"content": content},
    }
    _post_json(webhook, msg)


def _post_json(url: str, payload: dict):
    """通用 POST JSON"""
    try:
        data = json.dumps(payload).encode("utf-8")
        req = urllib.request.Request(
            url,
            data=data,
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        with urllib.request.urlopen(req, timeout=10) as resp:
            body = resp.read().decode("utf-8", errors="replace")
            log.info(f"Notification sent: {body[:200]}")
    except urllib.error.URLError as e:
        log.error(f"Notification FAILED: {e}")
    except Exception as e:
        log.error(f"Notification error: {e}")