python实现发送钉钉消息

需求：企业部门本地部署的缺陷管理工具，企业内部使用钉钉进行工作交流，老板想让每天汇报项目的测试情况；

设计思路：

1、创建钉钉群机器人，每天发送项目测试信息（缺陷数量、关闭数量、修复数量、项目名称、新增缺陷数量、新增缺陷负责人）；

2、使用python读取mysql数据库的表构造成钉钉的消息内容

实现方式：python3

一、安装python开发环境以及相关依赖库

安装python3开发环境可以百度；安装好后按下windowds+R，打开cmd，使用 pip install命令安装python库（需要库如下：pymysql，requests, json，nnlog)

二、创建本地文件夹DingMsg，并在文件夹创建mysql.py和DingMsg.py

三、编写读取mysql的代码

# coding: utf-8
#!/usr/bin/python
import requests, json
from pymysql import connect
import nnloglog = nnlog.Logger('DingMsg.log',level='debug',backCount=5,when='D')
class Op_mysql:def __int__(self):print('aaaa')
#用于获取返回结果为单个的@classmethoddef con_mysql(self,sql):conn = connect(host='',port=,user='',password='',db='',charset='utf8')cursor=conn.cursor()
#根据传入sql语句执行sql并将返回结果存入resultcursor.execute(sql)results = cursor.fetchone()for result in results:passlog.info(result)conn.close()return result
#用于获取返回结果为多个的def con_mysql_list(self, list):conn = connect(host='',port=,user='',password='',db='',charset='utf8')cursor = conn.cursor()cursor.execute(list)
#根据传入sql语句执行sql并将返回结果存入listnamelistname=[]resultsList = cursor.fetchall()for r_List in resultsList:listname.append(r_List)log.info(listname)return listname@classmethoddef getTodayNewbugs(self):# 获取今日新增缺陷数o = Op_mysql()news=o.con_mysql(sql = "SELECT COUNT(*) FROM  WHERE DATEDIFF(,NOW())=0")return news@classmethoddef getAllbugs(self):# 获取总缺陷数o=Op_mysql()all_bugs=o.con_mysql(sql="")return all_bugs@classmethoddef getProjectname(self):#获取项目名称o=Op_mysql()projectname = o.con_mysql_list(sql="")return projectname@classmethod# 获取今天新增bug的负责人def getassigned(self):o = Op_mysql()assigned = o.con_mysql_list(sql="")return assigned@classmethoddef getWaitingBugs(self):o = Op_mysql()WaitingBugs=o.con_mysql(sql="")return WaitingBugs@classmethod#获取今日关闭的缺陷状态def getClosebugs(self):o = Op_mysql()CloseBugs = o.con_mysql(sql="")return CloseBugs@classmethod#今日修复缺陷数def getFixbugs(self):o = Op_mysql()fixbugs = o.con_mysql(sql="")return fixbugsif __name__ == "__main__":op_mysql=Op_mysqlop_mysql.getProjectname()op_mysql.getAllbugs()op_mysql.getWaitingBugs()op_mysql.getClosebugs()op_mysql.getassigned()

四、编写DingMsg.py的代码

# coding: utf-8
#!/usr/bin/python
import requests, json
from pymysql import connect
import mysql
from mysql import Op_mysql
import nnloglog = nnlog.Logger('DingMsg.log',level='info',backCount=5,when='D')
# 发送钉钉消息
def send_dingtalk_message(url,content,mobile_list):headers = {'Content-Type': 'application/json'}data = {"msgtype": "text","text": {"content":content},"at": {# 要@的人"atMobiles": mobile_list,# 是否@所有人"isAtAll": False}}r = requests.post(url, headers=headers, data=json.dumps(data))log.info(r.text)return r.textif __name__ == "__main__":# 获取dingtalk token urlaccess_token = ''op_mysql = Op_mysql# 获取总缺陷数all_bugs = op_mysql.getAllbugs()# 获取今日提交的缺陷数new_bugs = op_mysql.getTodayNewbugs()# 获取项目名称projectname=op_mysql.getProjectname()namelist=[]for name in projectname:for n in name:namelist.append(n)if len(namelist)<=0:namelist.append("中台")log.info(namelist)# 获取待解决缺陷数waitingbugs = op_mysql.getWaitingBugs()#今日修复缺陷数fixbugs=op_mysql.getFixbugs()#今日关闭缺陷数closebugs=op_mysql.getClosebugs()#获取手机号列表phonenumber_list = op_mysql.getassigned()# 钉钉消息内容，注意{测试日报}是自定义的关键字，需要在钉钉机器人设置中添加，这样才能接收到消息content = F"----测试日报----\n新增缺陷项目名称：{namelist}\n今日新增缺陷数：{new_bugs}\n今日关闭缺陷数：{closebugs}\n今日修复缺陷数：{fixbugs}\n待解决缺陷数：{waitingbugs}\n总缺陷数：{all_bugs}\n"# 要@的人的手机号，可以是多个，注意：钉钉机器人设置中需要添加这些人，否则不会接收到消息mobile_list = []#从列表中取出手机号for a in phonenumber_list:for b in a:mobile_list.append(b)log.info(mobile_list)# 发送钉钉消息try:if new_bugs >= 0:send_dingtalk_message(access_token, content, mobile_list)except Exception as e:log.error(e)

四、获取钉钉机器人的url

打开钉钉群，点击【群设置】-【智能群助手】添加机器人，选择自定义的机器人，选择自定义关键词，关键词填写测试日报。完成后将url填写到DingMsg.py的main方法的access_token值上