【Python】通过钉钉机器人定时发送JIRA的缺陷

一、背景：由于目前项目交付周期比较紧张，部门要求每日测试bug要做到日清，所以写了如下脚本，每天17点通过机器人发送每日BUG修复情况。

二、效果如下：

三、代码详设

a. 调用JIRA，JQL语法进行搜索

import requests
import time
import hmac
import hashlib
import base64
import urllib.parse
import argparse
import pandas as pd
from jira import JIRA
from datetime import datetime
import urllib3
urllib3.disable_warnings()

class MyJira:
    def __init__(self, url, username, password) -> None:
        self.url = url
        self.username = username
        self.password = password
        self.jira = None
        self.login_jira(self.url, self.username, self.password)

    def login_jira(self, url, username, password, verify=False):
        jira = JIRA(url, basic_auth=(username, password), options={'verify': verify})
        self.jira = jira
        return jira
    # fields: 问题的描述，创建时间等所有的配置域，返回fields
    def get_all_fields(self):
        return self.jira.fields()

    # 获取问题的工作流
    def get_transitions(self, issue):
        return self.jira.transitions(issue)
    # 或者活动日志
    def get_worklogs(self, issue):
        return self.jira.worklogs(issue)

    # JQL(Jira Query Language)，Jira的Python库便是基于JQL语法进行搜索的，返回的是搜索到的问题列表
    def search_issue(self, jql, max_result=-1, **kw):
        search_result = self.jira.search_issues(jql, maxResults=max_result, **kw)
        return search_result
    
    def search_all_issue(self, jql, count=1000, **kw):
        startAt = 0
        page = 1
        sr = []
        sr_set = self.search_issue(jql, startAt=startAt, max_result=count, **kw)
        sr.extend(sr_set)
        print(f">>>>>>>>page count: {page}, total count: {len(sr)}, start at: {startAt}<<<<<<<<")
        while len(sr_set) == count:
            startAt = page * count
            sr_set = self.search_issue(jql, startAt=startAt, max_result=count, **kw)
            sr.extend(sr_set)
            page += 1
            print(f">>>>>>>>page count: {page}, total count: {len(sr)}, start at: {startAt}<<<<<<<<")
        return sr

 b.连接钉钉，创建机器人，获取token和加密ID

class DingTalk:
    def __init__(self, url) -> None:
        self.url = url

    def send_message(self, token, secret: str, msgtype: str = 'markdown', title: str = '', text: str = ''):
        timestamp, sign = self.sign(secret)
        data = {
            "msgtype": msgtype,
            "markdown": {
                "title": title,
                "text": text
            }
        }
        res = requests.post(f'{self.url}?access_token={token}×tamp={timestamp}&sign={sign}', json=data)
        print(res.status_code)

    @classmethod
    def sign(cls, secret):
        timestamp = str(round(time.time() * 1000))
        secret_enc = secret.encode('utf-8')
        string_to_sign = '{}n{}'.format(timestamp, secret)
        string_to_sign_enc = string_to_sign.encode('utf-8')
        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
        return timestamp, sign

c.主程序

def issue_to_str(issue):
    # print(issue.key)
    create_time = issue.fields.created
    delta_time = pd.to_datetime(datetime.now()) - pd.to_datetime(create_time[0:20])
    print(delta_time)
    days, hours, *_ = delta_time.components
    day_str = f'{days}天' if days > 0 else ''
    time_str = f'{day_str}{hours}小时'
    return f'- {issue.key}|{issue.fields.summary}|优先级: {issue.fields.priority.name}'
    # |经办人: {issue.fields.assignee.displayName}|{time_str}

def main(delta_hour: str):
    jira_url = 'https://jira.znlhzl.org'
    username = 'juchen'
    password = 'xxxxxxxxxxxxxxx'
    ding_url = 'https://oapi.dingtalk.com/robot/send'


    groups = [
        ('token','加密ID')
    ]
    ding = DingTalk(ding_url)

    jira = MyJira(jira_url, username, password)

    reporters = [
        'juchen', 'gaojincheng', 'hanwanyao1', 'wudijing', 'haoqi', 'yanya', 'zhumeng', 'wangkai2'
    ]
    bug_all_sql = f'project = SRD AND created >= 2022-04-17 AND issuetype = "缺陷(Bug)" AND reporter in ({",".join(reporters)}) order by created DESC'
    current_day_finished = f'status in ("[Bug]待回归", "[Bug]正常关闭", "[Bug]无效关闭") AND updated >= {delta_hour} AND reporter in ({",".join(reporters)}) order by created DESC'
    current_day_fixed = f'status in ("[Bug]待回归") AND updated >= {delta_hour} AND reporter in ({",".join(reporters)}) order by created DESC'
    current_day_left = f'status = "[BUG]待修复" AND created >= 2022-01-14  AND reporter in ({",".join(reporters)}) order by created DESC'

    # 总共bug数
    bug_all = jira.search_issue(bug_all_sql)
    # 今日已修复
    sr_finished_day = jira.search_issue(current_day_finished)
    print(sr_finished_day)
    # 今日待回归
    sr_cur_day = jira.search_issue(current_day_fixed)
    print(len(sr_cur_day))
    # 全部待修复
    sr_all = jira.search_all_issue(current_day_left, expand='changelog')

    sr_list = [issue_to_str(i) for i in sr_all]
    sr_list1 = [issue_to_str(i) for i in sr_cur_day]

    # 无未修复缺陷不发送
    if len(sr_all) == 0 and len(sr_cur_day) == 0:
        return
    elif len(sr_cur_day) == 0:
        data = [
            "## **缺陷修复情况**",
            # f"### **累计已修复缺陷数: {sr_total.total}**",
            f"### **截至目前项目总共缺陷数: {bug_all.total}**",
            f"### **今天已修复缺陷数: {sr_finished_day.total}**",
            f"### **今天待回归缺陷数: {sr_cur_day.total}**",
            f"### **剩余未修复缺陷数: {len(sr_all)}**",
            f"### **=====待研发修复详情如下=====**",
            *sr_list
        ]
    else:
        data = [
            "## **缺陷修复情况**",
            # f"### **累计已修复缺陷数: {sr_total.total}**",
            f"### **截至目前项目总共缺陷数: {bug_all.total}**",
            f"### **今天已修复缺陷数: {sr_finished_day.total}**",
            f"### **今天待回归缺陷数: {sr_cur_day.total}**",
            f"### **剩余未修复缺陷数: {len(sr_all)}**",
            f"### **=====待研发修复详情如下=====**",
            *sr_list,
            f"### **=====待测试回归问题详情如下====**",
            *sr_list1
        ]
    text = 'n'.join(data)
    # print(text)
    for token, secret in groups:
        ding.send_message(token, secret, msgtype='markdown', title='自营研发部', text=text)

if __name__ == '__main__':

    main('-17h')