Skip to main content

Ding Ding Alert Bot

Project description

DingTalkAlert

Python 3.10+ License: GPL v3 PyPI

一个简单易用的钉钉机器人报警 Python 库,支持多种消息类型和灵活的配置方式。

✨ 功能特性

  • 多种消息类型:文本、Markdown、链接、交互式卡片
  • 灵活的配置:支持环境变量和代码配置
  • 安全认证:支持加签密钥(Secret)验证
  • @功能:支持@指定人员和@所有人
  • 错误处理:完善的错误处理和日志记录
  • 类型提示:完整的类型注解,IDE 友好
  • 轻量级:仅依赖 requests
  • 开箱即用:提供完整的使用示例和测试用例

📦 安装

使用 pip 安装

pip install dingDingAlert

从源码安装

git clone https://github.com/tfkxsx/dingDingAlert.git
cd dingDingAlert
pip install -e .

🚀 快速开始

1. 获取钉钉机器人配置

  1. 在钉钉群中添加机器人
  2. 获取 Webhook URL 中的 access_token
  3. 如有安全设置,获取加签密钥(Secret)

2. 基本使用

from DingTalkAlert import DingTalkAlert

# 初始化机器人(不带 Secret)
dingtalk = DingTalkAlert(token="your_access_token")

# 发送文本消息
dingtalk.send_text("Hello, 钉钉!")

# 发送 Markdown 消息
dingtalk.send_markdown(
    title="系统通知",
    text="### 系统状态\n- CPU: **25%**\n- 内存: **68%**"
)

# 发送链接消息
dingtalk.send_link(
    title="查看详情",
    text="点击查看系统监控",
    message_url="https://example.com"
)

3. 使用环境变量配置

import os
from DingTalkAlert import DingTalkAlert, Config

# 设置环境变量
os.environ['DINGTALK_TOKEN'] = 'your_access_token'
os.environ['DINGTALK_SECRET'] = 'your_secret'  # 可选
os.environ['APP_NAME'] = '我的应用'            # 可选

# 使用 Config 类获取配置
dingtalk = DingTalkAlert(
    token=Config.DINGTALK_TOKEN,
    secret=Config.DINGTALK_SECRET
)

# 发送消息
dingtalk.send_text(f"{Config.APP_NAME} 启动成功")

📖 详细使用说明

初始化

from DingTalkAlert import DingTalkAlert

# 不带 Secret 的初始化
dingtalk = DingTalkAlert(token="your_access_token")

# 带 Secret 的初始化(推荐)
dingtalk = DingTalkAlert(
    token="your_access_token",
    secret="your_secret_key"
)

发送文本消息

# 普通文本消息
dingtalk.send_text("这是一条普通消息")

# @指定人员
dingtalk.send_text(
    content="请处理这个问题",
    at_mobiles=["13800138000", "13900139000"]
)

# @所有人
dingtalk.send_text(
    content="紧急通知,请所有人注意!",
    is_at_all=True
)

发送 Markdown 消息

markdown_content = """
### 系统监控报告
**时间**: 2024-01-01 10:00:00

#### 监控指标
- ✅ CPU 使用率: **25%**
- ⚠️ 内存使用率: **85%**
- ❌ 磁盘使用率: **95%**

#### 建议操作
1. 清理临时文件
2. 扩容磁盘空间
"""

dingtalk.send_markdown(
    title="系统告警",
    text=markdown_content,
    at_mobiles=["13800138000"]
)

发送链接消息

dingtalk.send_link(
    title="查看监控面板",
    text="实时系统监控数据和历史趋势",
    message_url="https://monitor.example.com",
    pic_url="https://example.com/thumbnail.png"  # 可选
)

发送交互式卡片

buttons = [
    {
        "title": "查看详情",
        "actionURL": "https://example.com/detail"
    },
    {
        "title": "立即处理",
        "actionURL": "https://example.com/action"
    },
    {
        "title": "忽略",
        "actionURL": "https://example.com/dismiss"
    }
]

dingtalk.send_action_card(
    title="处理告警",
    text="检测到系统异常,请选择处理方式",
    btns=buttons,
    btn_orientation="0"  # "0": 垂直排列, "1": 水平排列
)

⚙️ 配置说明

环境变量配置

| 变量名 | 描述 | 是否必填 | 默认值 | |--------|------|----------|--------| | DINGTALK_TOKEN | 钉钉机器人 access_token | 是 | 无 | | DINGTALK_SECRET | 加签密钥(Secret) | 否 | None | | APP_NAME | 应用名称 | 否 | 报警机器人 |

代码配置

from DingTalkAlert import Config

# 修改配置
Config.DINGTALK_TOKEN = "new_token"
Config.DINGTALK_SECRET = "new_secret"
Config.APP_NAME = "新应用名称"

🔧 API 参考

DingTalkAlert

__init__(token: str, secret: Optional[str] = None)

初始化钉钉机器人。

参数:

  • token: 钉钉机器人的 access_token
  • secret: 加签密钥(可选)

send_text(content: str, at_mobiles: List[str] = None, is_at_all: bool = False) -> Dict

发送文本消息。

参数:

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

send_markdown(title: str, text: str, at_mobiles: List[str] = None, is_at_all: bool = False) -> Dict

发送 Markdown 消息。

参数:

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

send_link(title: str, text: str, message_url: str, pic_url: str = None) -> Dict

发送链接消息。

参数:

  • title: 消息标题
  • text: 消息内容
  • message_url: 点击跳转的URL
  • pic_url: 图片URL(可选)

send_action_card(title: str, text: str, btns: List[Dict], btn_orientation: str = "0") -> Dict

发送交互式卡片。

参数:

  • title: 卡片标题
  • text: 卡片内容
  • btns: 按钮列表,格式为 [{"title": "...", "actionURL": "..."}]
  • btn_orientation: 按钮排列方向("0": 垂直, "1": 水平)

Config

配置管理类,支持环境变量和代码配置。

📚 示例

完整示例

#!/usr/bin/env python3
"""
完整的钉钉机器人使用示例
"""

import os
from datetime import datetime
from DingTalkAlert import DingTalkAlert, Config

# 配置环境变量
os.environ['DINGTALK_TOKEN'] = 'your_token_here'
os.environ['DINGTALK_SECRET'] = 'your_secret_here'
os.environ['APP_NAME'] = '生产监控系统'

# 初始化机器人
dingtalk = DingTalkAlert(
    token=Config.DINGTALK_TOKEN,
    secret=Config.DINGTALK_SECRET
)

def send_daily_report():
    """发送日报"""
    current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    
    report = f"""
### 📊 每日系统报告
**生成时间**: {current_time}
**应用名称**: {Config.APP_NAME}

#### 今日统计
- 总请求数: **1,234,567**
- 成功率: **99.8%**
- 平均响应时间: **245ms**

#### 异常情况
- API 超时: **23次**
- 数据库连接失败: **5次**
- 缓存命中率下降: **-2.3%**

#### 建议
1. 优化数据库连接池
2. 增加缓存服务器
3. 监控 API 响应时间
"""
    
    # 发送 Markdown 报告
    result = dingtalk.send_markdown(
        title="每日系统报告",
        text=report,
        at_mobiles=["13800138000"]  # @运维人员
    )
    
    return result

def send_alert(level: str, message: str):
    """发送告警"""
    alert_levels = {
        "info": "ℹ️",
        "warning": "⚠️", 
        "error": "❌",
        "critical": "🚨"
    }
    
    emoji = alert_levels.get(level, "ℹ️")
    
    alert_content = f"""
{emoji} **{level.upper()} 告警**
**时间**: {datetime.now().strftime('%H:%M:%S')}
**应用**: {Config.APP_NAME}

**内容**: {message}

**建议操作**:
1. 立即检查相关服务
2. 查看详细日志
3. 如有需要联系技术支持
"""
    
    # 根据级别决定是否@所有人
    is_at_all = level in ["critical", "error"]
    
    result = dingtalk.send_markdown(
        title=f"{level.upper()} 告警通知",
        text=alert_content,
        is_at_all=is_at_all
    )
    
    return result

# 使用示例
if __name__ == "__main__":
    # 发送日报
    send_daily_report()
    
    # 发送告警
    send_alert("warning", "数据库连接数接近阈值")
    
    # 发送成功通知
    dingtalk.send_text(f"{Config.APP_NAME} 报告发送完成")

运行测试

项目包含完整的测试用例,可以通过以下方式运行:

# 设置环境变量
export DINGTALK_TOKEN="your_test_token"
export DINGTALK_SECRET="your_test_secret"

# 运行测试
python DingTalkAlert/demo.py

🔍 调试和错误处理

查看响应结果

所有发送方法都会返回钉钉 API 的响应结果:

result = dingtalk.send_text("测试消息")
print(f"响应: {result}")

if result.get('errcode') == 0:
    print("消息发送成功")
else:
    print(f"发送失败: {result.get('errmsg')}")

常见错误

| 错误码 | 说明 | 解决方法 | |--------|------|----------| | 0 | 成功 | - | | 130101 | 参数错误 | 检查参数格式 | | 310000 | 机器人不存在 | 检查 token 是否正确 | | 310001 | 签名不匹配 | 检查 secret 是否正确 | | 310002 | IP 不在白名单 | 将服务器 IP 添加到白名单 |

🤝 贡献

欢迎贡献代码、报告问题或提出改进建议!

开发环境设置

  1. 克隆仓库
git clone https://github.com/tfkxsx/dingDingAlert.git
cd dingDingAlert
  1. 安装开发依赖
pip install -e ".[dev]"
  1. 运行测试
python -m pytest

提交代码

  1. Fork 仓库
  2. 创建功能分支
  3. 提交更改
  4. 创建 Pull Request

📄 许可证

本项目基于 GPL-3.0 许可证开源。详见 LICENSE 文件。

📞 支持

🙏 致谢

感谢以下开源项目:


提示: 在生产环境中使用前,请确保:

  1. 正确配置机器人安全设置(IP 白名单、关键词等)
  2. 处理好异常情况,避免因网络问题导致程序崩溃
  3. 合理控制消息发送频率,避免被钉钉限流 EOF

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

dingdingalert-1.0.1.tar.gz (22.9 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

dingdingalert-1.0.1-py3-none-any.whl (23.1 kB view details)

Uploaded Python 3

File details

Details for the file dingdingalert-1.0.1.tar.gz.

File metadata

  • Download URL: dingdingalert-1.0.1.tar.gz
  • Upload date:
  • Size: 22.9 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.0

File hashes

Hashes for dingdingalert-1.0.1.tar.gz
Algorithm Hash digest
SHA256 5321ed57f1dffe5ef55c1960cd95893be9b557763eda491c603f9e92c07ac519
MD5 d8aebdc58aaa32696e7c263cc5f1f34a
BLAKE2b-256 17f9f379b89799cb53f0af55eb4cda75b1637dece65049d17837da71367dae2b

See more details on using hashes here.

File details

Details for the file dingdingalert-1.0.1-py3-none-any.whl.

File metadata

  • Download URL: dingdingalert-1.0.1-py3-none-any.whl
  • Upload date:
  • Size: 23.1 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.0

File hashes

Hashes for dingdingalert-1.0.1-py3-none-any.whl
Algorithm Hash digest
SHA256 587fcb240a5c908285c2bd5cdd621c2f0681f242ab1e86e935f70b8aed990e66
MD5 4a0f1c07142d74db195cc21018f7561e
BLAKE2b-256 537697f2760eb2029b008cf405567e8ff70730ee6c7991255d2ca7657edf8954

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page