钉钉告警

https://open.dingtalk.com/document/orgapp/custom-bot-send-message-type

jinja 模板

## {{ bizdate }}磁盘使用情况报告

{%- for item in data %}

### {{ item.info }}

| 磁盘 | 总容量 | 已用 | 可用 | 使用率 |
|------|--------|------|------|--------|
{%- for msg in item.detail %}
| {{ msg.mountpoint }} | {{ "%.1f"|format(msg.total) }}GB | {{ "%.1f"|format(msg.used) }}GB | {{ "%.1f"|format(msg.free) }}GB | {{ '<font color="#FF0000">{:.1f}%</font>'.format(msg.percent) if msg.percent > 50 else '{:.1f}%'.format(msg.percent) }} |
{%- endfor %}
{%- endfor %}

#!/usr/bin/env python
# coding=utf-8
# -*- coding=utf-8
import requests
import json
import datetime

# 要修改的钉钉webhook
token_dd = '1bdba82cfc5c2afe204f1e99e30a4e2a49f5e2d7e6ababdfc7669287d830be12'

date_str = (datetime.datetime.now() + datetime.timedelta(days=0)).strftime("%Y-%m-%d")


def send_msg(token_dd, date_str, test_str, note_str, at_all=False):
    """
    通过钉钉机器人发送内容
    @param date_str:
    @return:
    """
    url = 'https://oapi.dingtalk.com/robot/send?access_token=' + token_dd
    headers = {'Content-Type': 'application/json;charset=utf-8'}
    content_str = '''{} 定时推送：\n'''.format(date_str)

    data = {
        "msgtype": "markdown",
        "markdown": {
            "title": "杭州天气告警",
            "text": test_str
        },
    }
    res = requests.post(url, data=json.dumps(data), headers=headers)
    print(res.text)

    return res.text

warn_text = """
**许可证号：** 330802210845  \n
**企业（字号）名称：** 衢州烟云供***************** \n
**经营地址：** 浙江省衢州**************5、47、49、51号（东方商厦百汇店内）\n
**许可证状态：** 停业 \n
"""



send_msg(token_dd, date_str, warn_text, True)

加签方式

import time
import hmac
import hashlib
import base64
import urllib.parse
import requests
import json

class DingTalkAlert:
    def __init__(self, webhook, secret):
        """
        初始化钉钉告警机器人
        :param webhook: 钉钉机器人的 Webhook 地址
        :param secret: 钉钉机器人的加签密钥 (SEC开头)
        """
        self.webhook = webhook
        self.secret = secret

    def _generate_sign(self):
        """
        生成加签签名
        规则：timestamp + "\n" + secret -> HmacSHA256 -> Base64 -> URLEncode
        """
        timestamp = str(round(time.time() * 1000))
        secret_enc = self.secret.encode('utf-8')
        string_to_sign = '{}\n{}'.format(timestamp, self.secret)
        string_to_sign_enc = string_to_sign.encode('utf-8')
        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
        return timestamp, sign

    def send_markdown_alert(self, title, text, at_mobiles=None, is_at_all=False):
        """
        发送 Markdown 格式的告警消息（推荐用于告警，排版更清晰）
        """
        timestamp, sign = self._generate_sign()
        url = f"{self.webhook}&timestamp={timestamp}&sign={sign}"

        payload = {
            "msgtype": "markdown",
            "markdown": {
                "title": title,
                "text": text
            },
            "at": {
                "atMobiles": at_mobiles or [],
                "isAtAll": is_at_all
            }
        }

        headers = {'Content-Type': 'application/json'}
        try:
            response = requests.post(url, headers=headers, data=json.dumps(payload), timeout=10)
            result = response.json()
            if result.get('errcode') == 0:
                print(f"[Success] 告警发送成功: {title}")
            else:
                print(f"[Failed] 告警发送失败: {result.get('errmsg')}")
            return result
        except Exception as e:
            print(f"[Error] 请求异常: {str(e)}")
            return None

# ================= 使用示例 =================
if __name__ == '__main__':
    # 1. 替换为你自己的 Webhook 地址和加签密钥
    WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token=你的token"
    SECRET = "SEC你的加签密钥"

    # 2. 初始化告警类
    alert = DingTalkAlert(WEBHOOK, SECRET)

    # 3. 构造告警内容并发送
    alert_title = "🚨 服务器P0级告警"
    alert_text = (
        "### 🚨 服务器P0级告警\n\n"
        "**告警级别：** P0 (严重)\n\n"
        "**主机IP：** 192.168.1.100\n\n"
        "**异常信息：** CPU使用率超过 95%，持续 5 分钟\n\n"
        "**触发时间：** 2026-09-07 15:58:00\n\n"
        "请及时登录服务器排查！"
    )

    # 发送告警，如需@指定人可传入手机号列表，如 at_mobiles=["138xxxx1234"]
    alert.send_markdown_alert(title=alert_title, text=alert_text, is_at_all=True)