专栏首页Debian中国钉钉发送报警消息的Python实现

钉钉发送报警消息的Python实现

我司使用钉钉作为主要的内部通讯工具,基本上大家在电脑和手机上都开着,消息可以第一时间查看,报警消息的即时性要求比较高,我们计划使用钉钉来发送报警通知。本文将简要介绍发送消息的API使用样例。钉钉的API接口文档,请查阅钉钉开放平台

介绍如何用 Python 实现钉钉自动发送消息,你可以用来和报警平台整合,或者做ChatOps。

获取 access token

要使用钉钉发送消息,首先需要获取access token,代码如下:

def get_access_token():
    url = 'https://oapi.dingtalk.com/gettoken?corpid=%s&corpsecret=%s' % (corp_id, corp_secret)
    request = urllib2.Request(url)
    response = urllib2.urlopen(request)
    response_str = response.read()
    response_dict = json.loads(response_str)
    error_code_key = "errcode"
    access_token_key = "access_token"
    if response_dict.has_key(error_code_key) and response_dict[error_code_key] == 0 and response_dict.has_key(access_token_key):
        return response_dict[access_token_key]
    else:
        return ''

access token 在2小时内有效,有效期内重复获取返回相同结果,重复获取会使有效期会自动顺延。corp_id 和corp_secret 是企业的 id 和 secret,在钉钉的管理后台可以找到。另外,上面用到了 urllib2 和 json,需要在脚本中引用。

消息类型

钉钉的消息类型分为:text,image,voice,file,link和OA。下面以发送文本,链接和文件消息为例进行说明。

给用户发送消息

发送文本

def send_text_to_users(access_token, users, text):
    msg_type, msg = _gen_text_msg(text)
    return _send_msg_to_users(access_token, users, msg_type, msg)

def _gen_text_msg(text):
    msg_type = 'text'
    msg = { "content": text }
    return msg_type, msg

def _send_msg_to_users(access_token, users, msg_type, msg):
    to_users = '|'.join(users)
    body_dict = {
        "touser": to_users,
        "agentid": agent_id,
        "msgtype": msg_type
    }
    body_dict[msg_type] = msg
    body = json.dumps(body_dict)
    return _send_msg("https://oapi.dingtalk.com/message/send?access_token=", access_token, body)

其中agent_id是一个钉钉应用的id,以钉钉应用的名义给用户发送消息。users是用户id列表,每个用户id是一个字符串。

发送链接

def send_link_to_users(access_token, users, url, title, text):
    msg_type, msg = _gen_link_msg(url, title, text)
    return _send_msg_to_users(access_token, users, msg_type, msg)

def _gen_link_msg(url, title, text):
    msg_type = 'link'
    msg = { 
        "messageUrl": url,
        "picUrl": "https://gw.alicdn.com/tps/TB1FN16LFXXXXXJXpXXXXXXXXXX-256-130.png",
        "title": title,
        "text": text 
    }
    return msg_type, msg

其中_send_msg_to_users方法参见前面的代码,picUrl字段设置的是钉钉官方的图片,这里用于测试。

发送文件

def send_file_to_users(access_token, users, file_name):
    media_id = upload_file(access_token, file_name)
    if media_id == '':
        return
    msg_type, msg = _gen_file_msg(media_id)
    return _send_msg_to_users(access_token, users, msg_type, msg)

def upload_file(access_token, file_name):
    register_openers()
    datagen, headers = multipart_encode({'media': open(file_name, 'rb')})
    requst_url = 'https://oapi.dingtalk.com/media/upload?access_token=' + access_token + '&type=file'
    request = urllib2.Request(requst_url, datagen, headers)
    response = urllib2.urlopen(request)
    response_str = response.read()
    response_dict = json.loads(response_str)
    media_id_key = 'media_id'
    error_code_key = 'errcode'
    if response_dict.has_key(error_code_key) and response_dict[error_code_key] == 0 and response_dict.has_key(media_id_key):
        return response_dict[media_id_key]
    else:
        return ''

需要先上传文件获得media_id,然后使用media_id将文件发送给用户。另外,这里用到了poster,可使用pip安装:

pip install poster

之后引入multipart_encode和register_openers函数:

from poster.encode import multipart_encode
from poster.streaminghttp import register_openers

给群会话发送消息

与给用户发送信息类似,区别是需要群会话id,而不是用户列表,以发送文本消息为例,代码如下:

def send_text_to_chat(access_token, chat_id, text):
    msg_type, msg = _gen_text_msg(text)
    return _send_msg_to_chat(access_token, chat_id, msg_type, msg)

def _send_msg_to_chat(access_token, chat_id, msg_type, msg):
    body_dict = {
        "chatid": chat_id,
        "msgtype": msg_type
    }
    body_dict[msg_type] = msg
    body = json.dumps(body_dict)
    return _send_msg("https://oapi.dingtalk.com/chat/send?access_token=", access_token, body)

其中_gen_text_msg 方法参见前面的代码。

群会话可以自行创建,详情请参考这里的文档

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • gRPC 1.23.0 发布,Google高性能 RPC 框架

    gRPC 1.23.0 发布了。gRPC 是 Google 开源的高性能、通用 RPC 框架,面向移动和 HTTP/2 设计,是由谷歌发布的首款基于 Proto...

    Debian社区
  • 使用 cURL 对Web请求进行性能测试

    在做 Web 开发的时候,经常需要对 Web Page 或者 REST-ful API 做简单的 Benchmark。本文将介绍如何使用 cURL 进行简单快速...

    Debian社区
  • Debian 10 Buster 图文安装指引

    本文是当前全网最全的,Debian 10 图文安装教程。本教程共有近四十张图片,按照安装步骤,一步步给予操作指引。如遇到问题,欢迎关注微信号 Debian社区 ...

    Debian社区
  • 实时日志:腾讯云 Serverless Python 运行时支持日志实时输出

    开发前言 作为一名 Serverless 架构的重度使用者,我一直对调试感到恐慌:经常在测试接口的时候,会通过网页/PostMan 触发函数,然后没得到预期的...

    腾讯云serverless团队
  • oracle数据库查询open_cursors值的sql语句,达梦数据库查询MAX_SESSION_STATEMENT值方法,MAX_SESSION_STATEMENT的最大值、上限是多少。

    windows 下需要用 cmd 登录 oracle 数据库,用 show parameter open_cursors 来查询。 可以看到 open_cur...

    小蓝枣
  • 高等数学——简单直观地了解定积分

    对于很多人来说定积分的内容其实早在高中就已经接触过了,比如在高中物理当中,我们经常使用一种叫做”微元法“的方法来解决一些物理问题。但实际上所谓的”微元法“本质上...

    TechFlow-承志
  • 前端每日实战:85# 视频演示如何用纯 CSS 创作一个小球反弹的动画

    https://github.com/comehope/front-end-daily-challenges

    前端博客 : alili.tech
  • 合理使用const,慎用自运算

        项目最的出了几次运营事故,都是因为使用自乘、自加、自減运算,错改了非局部变量,导致将用户数据写溢出,最终只能进行回档处理。先给大家展示一下,漏出bug的...

    王亚昌
  • MyBatis学习总结(一)——ORM概要与MyBatis快速起步

    程序员应该将核心关注点放在业务上,而不应该将时间过多的浪费在CRUD中,多数的ORM框架都把增加、修改与删除做得非常不错了,然后数据库中查询无疑是使用频次最高、...

    张果
  • WCF采用 netTcpBinding 发生的Socket errors

    最近在项目中采用 Windows service 作为WCF services的宿主, 在服务和客户端的调用上没有发生如何的异常和错误,但是经常发生下述错误: ...

    张善友

扫码关注云+社区

领取腾讯云代金券