©2020-2024 ioDraw All rights reserved,
Privacy Policy
# coding: utf-8 # subject: accumulate total bug-count and bug-count in a week.
import re import json import time from functools import reduce from datetime
import date, timedelta from jira import JIRA class AnalysisJira(object):
"""summary the amount of bug in jira""" def __init__(self, server, user, pwd):
# connect jira system try: self.jira = JIRA(server=server, basic_auth=(user,
pwd)) except Exception as e: raise def jql_query(self, project=None, text='',
reporter=None, order='', max_result=10): """use JQL to query issue of current
user""" try: issues = self.jira.search_issues("project = '{}' AND text ~ '{}'
AND reporter in {} ORDER BY {}"\ .format(project, text, reporter, order),
maxResults=max_result) except Exception as e: raise e else: return issues
@staticmethod def merge_dict(original={'a':{'a':1,'c':3},'b':{'b':2,'c':2}},
allow_keys=['a','b']): """merge a few dict to one dict by summing its values of
that are same key""" if isinstance(original, dict) and original is not dict()
and isinstance(allow_keys, (tuple,list)): dict_list = [original[k] for k in
allow_keys if k in original.keys() and isinstance(original.get(k), dict)]
#print(dict_list) if len(dict_list) >=1: wanted = dict_list[0] for item in
dict_list[1:]: for key in set(list(wanted.keys())+list(item.keys())):
wanted.setdefault(key, 0) wanted[key] = wanted.get(key, 0) + item.get(key, 0)
return wanted else: pass else: return None def base_datas(self, **kargs): """
get semi-finished datas, return such as {'New':
[{'subject':'***',priority':'Major','created':'2019-08-01','updated':'2019-09-01'}]}
""" try: base_datas = {'priorities': set()} for issue in
self.jql_query(**kargs): id = issue.key #ID project = issue.fields.project #项目
issue_type = issue.fields.issuetype #问题类型 reporter = issue.fields.reporter #报告人
assignee = issue.fields.assignee #经办人 subject = issue.fields.summary #主题
created = re.search('(.*)(?=T)', issue.fields.created).group() #创建 updated =
re.search('(.*)(?=T)', issue.fields.updated).group() #更新 priority =
str(issue.fields.priority) #优先级 status = re.search(r'[a-zA-Z]+',
str(issue.fields.status)).group() #状态 detail = {'subject': subject,'priority':
priority,'created': created,'updated': updated} if status in base_datas.keys():
base_datas[status].append(detail) else: base_datas[status] = [detail] #
记录所有的优先级 base_datas['priorities'].add(priority) except Exception as e: raise
else: return base_datas def current_data(self, **kargs):
"""对条件查询的结果，进行加工得到总数据""" try: data = self.base_datas(**kargs) #
删除data中的priorities，并利用它 priority = data.pop('priorities') # 精简每个状态中的数据 for k,v
in data.items(): all_priorities = [y['priority'] for y in data[k]] data[k] =
{x:all_priorities.count(x) for x in set(all_priorities)} # 得出总数据summary的详情
data['Summary'] = dict().fromkeys(priority, 0) # 以状态种类作为第一循环 for value in
list(data.values())[0:len(data)-1]: for k in priority: data['Summary'][k] +=
value.get(k, 0) # 得出各项的total for k,v in data.items(): data[k]['total'] =
sum(data[k].values()) except Exception as e: raise else: data_str =
json.dumps(data).replace('}, ', '}\n').replace('{', '').replace('}',
'').replace("'", '').replace('"', '') return data, data_str; def
one_week_data(self, end_date=date.today().isoformat(), past_days = 7, **kargs):
"""统计距某个日期一周内的bug数据""" datas = {'created_in_a_week':{}, 'closed_in_a_week':{},
'unsolved_at_now':{}, 'to_be_verified':{}} #
由end_date得到过去6天的日期，加上end_date，组成新的日期列表 today = date.today().isoformat()
past_dates = [(date(*map(int, end_date.split('-'))) -
timedelta(days=days)).isoformat() for days in range(1,past_days)]
past_dates.insert(0, end_date) #print(past_dates) try: if end_date > today:
print("请输入小于等于今天的日期!") else: raw_data = self.base_datas(**kargs) priority =
raw_data.pop('priorities') for k,v in raw_data.items(): for item in
raw_data[k]: if item['created'] in past_dates and item['priority'] in
datas['created_in_a_week'].keys(): datas['created_in_a_week'][item['priority']]
+= 1 elif item['created'] in past_dates and item['priority'] not in
datas['created_in_a_week'].keys(): datas['created_in_a_week'][item['priority']]
= 1 else: pass for item in raw_data.get('Closed', []): if item['updated'] in
past_dates and item['priority'] in datas['closed_in_a_week'].keys():
datas['closed_in_a_week'][item['priority']] += 1 elif item['updated'] in
past_dates and item['priority'] not in datas['closed_in_a_week'].keys():
datas['closed_in_a_week'][item['priority']] = 1 else: pass except Exception as
e: raise else: for item in datas.keys(): datas[item]['total'] =
sum(datas[item].values()) if datas[item] != {} else 0 current =
self.current_data(**kargs)[0] # 未解决包括新建、处理中、重新打开(unresolved = New + InProcess +
ReOpened) datas['unsolved_at_now'] = self.merge_dict(current, ['New',
'Inprocess', 'Reopened']) datas['to_be_verified'] = current['Resolved']
datas['end_date'] = end_date datas_str = json.dumps(datas).replace('}, ',
'}\n').replace('{', '').replace('}', '').replace('"', '') return datas,
datas_str if __name__ == '__main__': server = '***' user='***' pwd='***'
condition = {'project':'***', 'text':'***', 'reporter':'****',
'order':'createdDate DESC', 'max_result':1000} # 当前的bug情况 #total_data =
AnalysisJira(server, user, pwd).current_data(**condition) #print(total_data[1])