a-cloud-all/.devops/scripts/dingtalk.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
钉钉 Webhook 通知模块
用于发送构建和部署通知到钉钉群
"""

import time
import hmac
import hashlib
import base64
import urllib.parse
import urllib.request
import json
from datetime import datetime


class DingTalkNotifier:
    """钉钉通知器"""

    def __init__(self, access_token, secret=None):
        """
        初始化钉钉通知器

        Args:
            access_token: 钉钉机器人的 access_token
            secret: 钉钉机器人的加签密钥（可选）
        """
        self.access_token = access_token
        self.secret = secret
        self.base_url = "https://oapi.dingtalk.com/robot/send"

    def _generate_sign(self, timestamp):
        """
        生成钉钉加签

        Args:
            timestamp: 当前时间戳（毫秒）

        Returns:
            签名字符串
        """
        if not self.secret:
            return None

        string_to_sign = f'{timestamp}\n{self.secret}'
        string_to_sign_enc = string_to_sign.encode('utf-8')
        secret_enc = self.secret.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 sign

    def _build_url(self):
        """
        构建完整的 webhook URL

        Returns:
            完整的 URL 字符串
        """
        url = f"{self.base_url}?access_token={self.access_token}"

        if self.secret:
            timestamp = str(round(time.time() * 1000))
            sign = self._generate_sign(timestamp)
            url += f"&timestamp={timestamp}&sign={sign}"

        return url

    def send_text(self, content, at_mobiles=None, at_all=False):
        """
        发送文本消息

        Args:
            content: 消息内容
            at_mobiles: @的手机号列表
            at_all: 是否@所有人

        Returns:
            发送结果 (True/False)
        """
        data = {
            "msgtype": "text",
            "text": {
                "content": content
            },
            "at": {
                "atMobiles": at_mobiles or [],
                "isAtAll": at_all
            }
        }

        return self._send_request(data)

    def send_build_success(self, repo_name, branch, commit_hash, duration):
        """
        发送构建成功通知

        Args:
            repo_name: 仓库名称
            branch: 分支名称
            commit_hash: 提交哈希
            duration: 构建耗时（秒）

        Returns:
            发送结果 (True/False)
        """
        now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        title = f"✅ 构建成功 - {repo_name}"
        text = f"""### ✅ 构建成功

**仓库**: {repo_name}
**分支**: {branch}
**提交**: {commit_hash[:8]}
**耗时**: {duration:.1f} 秒
**时间**: {now}

---
构建和部署已完成！
"""
        return self.send_markdown(title, text)

    def send_build_failure(self, repo_name, branch, commit_hash, error_msg, at_all=False):
        """
        发送构建失败通知

        Args:
            repo_name: 仓库名称
            branch: 分支名称
            commit_hash: 提交哈希
            error_msg: 错误信息
            at_all: 是否@所有人

        Returns:
            发送结果 (True/False)
        """
        now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        title = f"❌ 构建失败 - {repo_name}"
        # 限制错误信息长度
        error_display = error_msg[:500] if len(error_msg) > 500 else error_msg
        text = f"""### ❌ 构建失败

**仓库**: {repo_name}
**分支**: {branch}
**提交**: {commit_hash[:8]}
**时间**: {now}

**错误信息**:
```
{error_display}
```

---
请检查日志并修复问题！
"""
        return self.send_markdown(title, text, at_all=at_all)

    def send_build_start(self, repo_name, branch, commit_hash, commit_message=None, server_ip=None):
        """
        发送构建开始通知

        Args:
            repo_name: 仓库名称
            branch: 分支名称
            commit_hash: 提交哈希
            commit_message: 提交消息（可选）
            server_ip: 服务器IP（可选）

        Returns:
            发送结果 (True/False)
        """
        now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        title = f"🚀 构建开始 - {repo_name}"

        # 构建消息内容
        text = f"""### 🚀 构建开始

**仓库**: {repo_name}
**分支**: {branch}
**提交**: {commit_hash[:8]}"""

        if commit_message:
            text += f"\n**消息**: {commit_message}"

        if server_ip:
            text += f"\n**服务器**: {server_ip}"

        text += f"""
**时间**: {now}

---
构建任务已启动，请稍候...
"""
        return self.send_markdown(title, text)

    def _send_request(self, data):
        """
        发送 HTTP 请求到钉钉 webhook

        Args:
            data: 请求数据字典

        Returns:
            发送结果 (True/False)
        """
        try:
            url = self._build_url()
            headers = {'Content-Type': 'application/json'}
            json_data = json.dumps(data).encode('utf-8')

            req = urllib.request.Request(url, data=json_data, headers=headers)
            response = urllib.request.urlopen(req, timeout=10)
            result = json.loads(response.read().decode('utf-8'))

            if result.get('errcode') == 0:
                return True
            else:
                print(f"钉钉通知发送失败: {result.get('errmsg')}")
                return False

        except Exception as e:
            print(f"钉钉通知发送异常: {e}")
            return False

    def send_markdown(self, title, text, at_mobiles=None, at_all=False):
        """
        发送 Markdown 消息

        Args:
            title: 消息标题
            text: Markdown 格式的消息内容
            at_mobiles: @的手机号列表
            at_all: 是否@所有人

        Returns:
            发送结果 (True/False)
        """
        data = {
            "msgtype": "markdown",
            "markdown": {
                "title": title,
                "text": text
            },
            "at": {
                "atMobiles": at_mobiles or [],
                "isAtAll": at_all
            }
        }

        return self._send_request(data)