python 实现禅道 bug 日清通知，发送钉钉并@对应负责人

Posted 2022-12-05 七月的小尾巴

前言

在项目提测阶段，很多公司都会要求bug日清，那么对于这个指标要求，大部分公司都会将这个任务分配给测试去跟进。

那么咱们测试人员，总不能说每天去盯着禅道，然后去找对应的开发吧。那么下面就是咱们测试人员的高光时刻，通过脚本去实现统计bug情况，并且钉钉@对应的开发~

话不多说，直接开干！

数据库封装

首先，我们需要连接禅道的数据库，查询出所有项目未关闭的bug数据。

# -*- coding: utf-8 -*-
# @Time    : 2020/12/20 4:25 PM
# @Author  : 余少琪
# @FileName: mysql_db.py
# @email   : [e-mail]
# @Software: 蜂go


import pymysql
from warnings import filterwarnings
from config.setting import DevConfig


# 忽略 Mysql 告警信息
filterwarnings("ignore", category=pymysql.Warning)


class MysqlDB(object):

    def __init__(self):
        self.config = DevConfig()

        # 建立数据库连接
        self.conn = pymysql.connect(
            host='',
            user='',
            password='',
            db=''
        )

        # 使用 cursor 方法获取操作游标，得到一个可以执行sql语句，并且操作结果为字典返回的游标
        self.cur = self.conn.cursor(cursor=pymysql.cursors.DictCursor)

    def __del__(self):
        # 关闭游标
        self.cur.close()
        # 关闭连接
        self.conn.close()

    def query(self, sql, state="all"):
        """
        查询
        :param sql:
        :param state:  all 是默认查询全部
        :return:
        """

        self.cur.execute(sql)

        if state == "all":
            # 查询全部
            data = self.cur.fetchall()

        else:
            # 查询单条
            data = self.cur.fetchone()

        return data

    def excute(self, sql):
        """
        更新 、 删除、 新增
        :param sql:
        :return:
        """
        try:
            # 使用 excute 操作 sql
            rows = self.cur.execute(sql)
            # 提交事务
            self.conn.commit()
            return rows
        except Exception as e:
            print("数据库操作异常0".format(e))
            # 如果事务异常，则回滚数据
            self.conn.rollback()

封装钉钉通知

这里钉钉通知，我们需要用到钉钉机器人的 webhook、 sgin，这两个数据我是放在配置文件中读取的，大家可以直接写在代码中。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time   : 2021/11/25 13:15
# @Author : 余少琪

import base64
import hashlib
import hmac
import time
import urllib.parse
from utils.FileOperation.yamlControl import GetYamlData
from dingtalkchatbot.chatbot import DingtalkChatbot, FeedLink
from config.setting import ConfigHandler
from typing import Any


class DingTalkSendMsg(object):

    def __init__(self):
        self.timeStamp = str(round(time.time() * 1000))
        self.sign = self.get_sign()
        self.devConfig = ConfigHandler()
        # 从yaml文件中获取钉钉配置信息
        self.getDingTalk = GetYamlData(self.devConfig.config_path).get_yaml_data()['DingTalk']

        # 获取 webhook地址
        self.webhook = self.getDingTalk["webhook"] + "&timestamp=" + self.timeStamp + "&sign=" + self.sign
        self.xiaoDing = DingtalkChatbot(self.webhook)
        self.Process = CaseCount()

    def get_sign(self) -> str:
        """
        根据时间戳 + "sign" 生成密钥
        :return:
        """
        secret = GetYamlData(ConfigHandler().config_path).get_yaml_data()['DingTalk']['secret']
        string_to_sign = '\\n'.format(self.timeStamp, secret).encode('utf-8')
        hmac_code = hmac.new(secret.encode('utf-8'), string_to_sign, digestmod=hashlib.sha256).digest()
        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
        return sign

    def send_text(self, msg: str, mobiles=None) -> None:
        """
        发送文本信息
        :param msg: 文本内容
        :param mobiles: 艾特用户电话
        :return:
        """
        if not mobiles:
            self.xiaoDing.send_text(msg=msg, is_at_all=True)
        else:
            if isinstance(mobiles, list):
                self.xiaoDing.send_text(msg=msg, at_mobiles=mobiles)
            else:
                raise TypeError("mobiles类型错误 不是list类型.")

    def send_link(self, title: str, text: str, message_url: str, pic_url: str) -> None:
        """
        发送link通知
        :return:
        """
        try:
            self.xiaoDing.send_link(title=title, text=text, message_url=message_url, pic_url=pic_url)
        except Exception:
            raise

    def send_markdown(self, title: str, msg: str, mobiles=None, is_at_all=False) -> None:
        """

        :param is_at_all:
        :param mobiles:
        :param title:
        :param msg:
        markdown 格式
        """

        if mobiles is None:
            self.xiaoDing.send_markdown(title=title, text=msg, is_at_all=is_at_all)
        else:
            if isinstance(mobiles, list):
                self.xiaoDing.send_markdown(title=title, text=msg, at_mobiles=mobiles)
            else:
                raise TypeError("mobiles类型错误 不是list类型.")

    @staticmethod
    def feed_link(title: str, message_url: str, pic_url: str) -> Any:

        return FeedLink(title=title, message_url=message_url, pic_url=pic_url)

    def send_feed_link(self, *arg) -> None:
        try:
            self.xiaoDing.send_feed_card(list(arg))
        except Exception:
            raise

代码实现

实现思路：

1. 这里会涉及到查询禅道的数据库中的数据，所以采用了继承 MysqlDB类
2. 考虑到通常我们封装的数据库类，通常都会连接公司的数据库，所以我在初始化方法中，重新编写了子类的连接信息
3. 封装获取未关闭项目ID方法
4. 查询所有项目中未关闭的bug总数、对应负责人、负责人未关闭的bug总数
5. 发送钉钉通知并@对方

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time   : 2022/2/17 16:13
# @Author : 余少琪

from utils.mysql.mysqlControl import MysqlDB
import pymysql
from utils.notice.dingtalkControl import DingTalkSendMsg


class Bugs(MysqlDB):
    """ bug 日清通知 """

    def __init__(self):
        MysqlDB.__init__(self)
        try:
            # 建立数据库连接
            self.conn = pymysql.connect(
                host="",
                user="root",
                password="",
                db='',
                port=3334
            )

            # 使用 cursor 方法获取操作游标，得到一个可以执行sql语句，并且操作结果为字典返回的游标
            self.cur = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
        except Exception as e:
            print("数据库连接失败，失败原因0".format(e))

    def get_project_ids(self) -> list:
        """ 获取项目 ID """
        sql = "select * from zt_product where acl = 'open' and deleted = '0';"
        project_data = self.query(sql)
        ids = []
        for i in project_data:
            ids.append(i['id'])

        return ids

    def get_bugs(self) -> list:
        BUGS = []
        for i in self.get_project_ids():
            """查询所有项目中未关闭的bug（不包含延期处理）"""
            sql = "select count(a.id) as bugCount, b.realName , c.name, b.mobile from zt_bug as a " \\
                  "inner join zt_user as b inner join zt_product as c where a.assignedTo = b.account " \\
                  "and a.product = c.id and a.product = 0 and a.deleted = '0' and a.status != 'closed' " \\
                  "and resolution != 'postponed'GROUP BY a.assignedTo".format(i)

            # 统计出该项目中未关闭的bug对应的负责人，以及未关闭的总数
            bugsCount = self.query(sql)
            BUGS.append(bugsCount)
        return BUGS

    def send_Notice_Msg(self):
        """发送钉钉通知@对应的负责人"""
        for bugs in self.get_bugs():
            for i in bugs:
                text1 = "#### **禅道处理通知**\\n   执   行   人: 0  \\n\\n 未处理bug数量: 1 个 \\n\\n 所  属  项 目: 2 " \\
                        "\\n\\n[点我查看](http://114.55.234.41:8000/bug-browse-3-0-assigntome.html)".format(i['realName'], i['bugCount'], i['name'])
                if i['mobile'] != "":
                    # 判断如果禅道中的用户有手机号码，才会@对方
                    DingTalkSendMsg().send_markdown(title="禅道处理通知", msg=text1, mobiles=[i['mobile']])
                else:
                    DingTalkSendMsg().send_markdown(title="禅道处理通知", msg=text1)


if __name__ == '__main__':
    Bugs().send_Notice_Msg()

下面我们来看看禅道的日清通知吧~~