115 lines
3.2 KiB
Python
115 lines
3.2 KiB
Python
import json
|
|
import requests
|
|
import time
|
|
import hmac
|
|
import hashlib
|
|
import base64
|
|
import urllib.parse
|
|
import traceback
|
|
from typing import Dict, Any, List, Optional, Union
|
|
|
|
|
|
class DingTalkBot:
|
|
"""钉钉机器人工具类，用于发送分析结果到钉钉群"""
|
|
|
|
def __init__(self, webhook_url: str, secret: Optional[str] = None):
|
|
"""
|
|
初始化钉钉机器人
|
|
|
|
Args:
|
|
webhook_url: 钉钉机器人的webhook地址
|
|
secret: 钉钉机器人的签名密钥（如果启用了安全设置）
|
|
"""
|
|
self.webhook_url = webhook_url
|
|
self.secret = secret
|
|
|
|
def _sign(self) -> str:
|
|
"""
|
|
生成钉钉机器人签名
|
|
|
|
Returns:
|
|
签名后的URL
|
|
"""
|
|
if not self.secret:
|
|
return self.webhook_url
|
|
|
|
timestamp = str(round(time.time() * 1000))
|
|
string_to_sign = f"{timestamp}\n{self.secret}"
|
|
hmac_code = hmac.new(
|
|
self.secret.encode(),
|
|
string_to_sign.encode(),
|
|
digestmod=hashlib.sha256
|
|
).digest()
|
|
|
|
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
|
|
return f"{self.webhook_url}×tamp={timestamp}&sign={sign}"
|
|
|
|
def send_text(self, content: str, at_mobiles: List[str] = None, at_all: bool = False) -> Dict[str, Any]:
|
|
"""
|
|
发送文本消息
|
|
|
|
Args:
|
|
content: 消息内容
|
|
at_mobiles: 要@的手机号列表
|
|
at_all: 是否@所有人
|
|
|
|
Returns:
|
|
接口返回结果
|
|
"""
|
|
data = {
|
|
"msgtype": "text",
|
|
"text": {"content": content},
|
|
"at": {
|
|
"atMobiles": at_mobiles if at_mobiles else [],
|
|
"isAtAll": at_all
|
|
}
|
|
}
|
|
|
|
return self._post(data)
|
|
|
|
def send_markdown(self, title: str, text: str, at_mobiles: List[str] = None, at_all: bool = False) -> Dict[str, Any]:
|
|
"""
|
|
发送Markdown消息
|
|
|
|
Args:
|
|
title: 消息标题
|
|
text: Markdown格式的消息内容
|
|
at_mobiles: 要@的手机号列表
|
|
at_all: 是否@所有人
|
|
|
|
Returns:
|
|
接口返回结果
|
|
"""
|
|
data = {
|
|
"msgtype": "markdown",
|
|
"markdown": {
|
|
"title": title,
|
|
"text": text
|
|
},
|
|
"at": {
|
|
"atMobiles": at_mobiles if at_mobiles else [],
|
|
"isAtAll": at_all
|
|
}
|
|
}
|
|
|
|
return self._post(data)
|
|
|
|
def _post(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
发送消息到钉钉
|
|
|
|
Args:
|
|
data: 消息数据
|
|
|
|
Returns:
|
|
接口返回结果
|
|
"""
|
|
try:
|
|
webhook = self._sign()
|
|
headers = {'Content-Type': 'application/json; charset=utf-8'}
|
|
response = requests.post(webhook, json=data, headers=headers)
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"发送钉钉消息时出错: {e}")
|
|
traceback.print_exc()
|
|
return {"errcode": -1, "errmsg": str(e)} |