Python微信/钉钉消息推送脚本
通过企业微信机器人或钉钉机器人发送消息通知，支持文本、Markdown、图片，可集成到各种脚本中。
详细内容
#!/usr/bin/env python3
# Python微信/钉钉消息推送脚本
# 使用方法: 修改webhook后直接运行，或作为模块导入
import requests
import json
import hashlib
import base64
import time
import hmac
import urllib.parse
# ========== 配置区 ==========
# 企业微信机器人Webhook (在群里添加机器人获取)
WECHAT_WEBHOOK = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your_key"
# 钉钉机器人Webhook
DINGTALK_WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token=your_token"
DINGTALK_SECRET = "your_secret" # 加签密钥(可选)
# ============================
class WeChatNotifier:
"""企业微信消息推送"""
def __init__(self, webhook=WECHAT_WEBHOOK):
self.webhook = webhook
def send_text(self, content, mentioned_list=None):
"""发送文本消息"""
data = {
"msgtype": "text",
"text": {
"content": content,
"mentioned_list": mentioned_list or []
}
}
return self._send(data)
def send_markdown(self, title, content):
"""发送Markdown消息"""
data = {
"msgtype": "markdown",
"markdown": {
"content": f"## {title}\n\n{content}"
}
}
return self._send(data)
def _send(self, data):
try:
response = requests.post(self.webhook, json=data, timeout=10)
result = response.json()
if result.get('errcode') == 0:
print('✅ 企业微信消息发送成功')
return True
else:
print(f'❌ 发送失败: {result}')
return False
except Exception as e:
print(f'❌ 请求异常: {e}')
return False
class DingTalkNotifier:
"""钉钉消息推送"""
def __init__(self, webhook=DINGTALK_WEBHOOK, secret=DINGTALK_SECRET):
self.webhook = webhook
self.secret = secret
def _get_sign_url(self):
"""生成加签URL"""
if not self.secret:
return self.webhook
timestamp = str(round(time.time() * 1000))
string_to_sign = f'{timestamp}\n{self.secret}'
hmac_code = hmac.new(
self.secret.encode('utf-8'),
string_to_sign.encode('utf-8'),
digestmod=hashlib.sha256
).digest()
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
return f'{self.webhook}×tamp={timestamp}&sign={sign}'
def send_text(self, content):
"""发送文本消息"""
data = {
"msgtype": "text",
"text": {"content": content}
}
return self._send(data)
def send_markdown(self, title, content):
"""发送Markdown消息"""
data = {
"msgtype": "markdown",
"markdown": {"title": title, "text": content}
}
return self._send(data)
def _send(self, data):
try:
url = self._get_sign_url()
response = requests.post(url, json=data, timeout=10)
result = response.json()
if result.get('errcode') == 0:
print('✅ 钉钉消息发送成功')
return True
else:
print(f'❌ 发送失败: {result}')
return False
except Exception as e:
print(f'❌ 请求异常: {e}')
return False
# 使用示例
if __name__ == '__main__':
# 企业微信推送
wechat = WeChatNotifier()
wechat.send_text('【系统通知】服务器备份完成\n时间: 2026-08-30 03:00\n状态: 成功')
wechat.send_markdown(
'每日运维报告',
'> CPU使用率: 25%\n'
'> 内存使用率: 45%\n'
'> 磁盘使用率: 60%\n'
'> 服务状态: 全部正常'
)
# 钉钉推送
# dingtalk = DingTalkNotifier()
# dingtalk.send_text('Hello from Python!')