Jenkins 安裝的插件
插件名稱 | 作用 |
---|---|
Rebuilder | |
-
Rebuilder。
官方地址:https://plugins.jenkins.io/rebuild
安裝方式:在Jenkins插件當中直接搜索即可安裝。
功能說明:此插件可以直接重復上次構建,也可以用于查看一些參數(shù)比較復雜的構建時,上次構建所選的參數(shù)是什么。非常nice的一個插件。 -
AnsiColor。
官方地址:https://plugins.jenkins.io/ansicolor
安裝方式:在Jenkins插件當中直接搜索即可安裝。
功能說明:擴展支持我們在shell當中定義的echo -e指令,從而給一定的輸出上顏色。
使用方式:點此跳轉到使用介紹。(opens new window) -
Maven Release Plug-in。
maven風格插件。
安裝方式:在Jenkins插件當中直接搜索即可安裝。 -
user build vars。
官方地址:https://wiki.jenkins.io/display/JENKINS/Build+User+Vars+Plugin
安裝方式:在Jenkins插件當中直接搜索即可安裝。
功能說明:通過此插件,讓整個Jenkins系統(tǒng)中的用戶參數(shù)成為一個可調用的變量。
使用方式:在構建環(huán)境中選中Set Jenkins user build variables。 -
Post build task
功能說明:此功能允許您關聯(lián) shell 或批處理腳本,這些腳本根據(jù)構建日志輸出在 Hudson
上執(zhí)行某些任務。如果日志文本與構建日志文件中的某處匹配,腳本將執(zhí)行。例如:如果在日志文本中指定了“IOException”,并且構建日志文件包含字符串“IOException”,則腳本將執(zhí)行。
允許使用 Java 正則表達式,并且可以將組用作腳本參數(shù)。如果文本是“Last Build : #(\d+)”并且腳本是“script.sh”,那么如果日志包含一行“Last
Build : #4”,腳本“script.sh 4”將被調用. -
MultiJob Phase
功能說明:上下游執(zhí)行
發(fā)送消息到飛書
預覽 1 (單Job)
-
示例圖
-
對應shell
#!/usr/bin/env bash
url1="https://open.feishu.cn/open-apis/bot/v2/hook/"
url2="https://open.feishu.cn/open-apis/bot/v2/hook/"
# 1. 消息 接收地址
webhook_list=($url1 $url2)
# ========================
# 2. 消息 參數(shù)預處理
# ========================
# 3. 消息 執(zhí)行發(fā)送
run_send_msg() {
echo -e "\n 復制發(fā)送消息腳本 $HOME -> ${WORKSPACE}" && cp "$HOME"/send_msg_to_feishu.py "${WORKSPACE}"
for ((i = 0; i < ${#webhook_list[@]}; i++)); do
webhook=${webhook_list[$i]}
echo -e "發(fā)送中 --> $webhook"
python3 send_msg_to_feishu.py "${webhook}" -job_url "${JOB_URL}"
done
echo -e "發(fā)送完成 \n\n"
}
run_send_msg
- 對應python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import argparse
import json
import subprocess
import time
import datetime
import requests
parser = argparse.ArgumentParser(description='Jenkins發(fā)送消息到飛書')
parser.add_argument('webhook', help='機器人webhookURL')
parser.add_argument('-job_url', '--JOB_URL', help='作業(yè)URL', required=True)
args = parser.parse_args()
WEBHOOK = args.webhook
JOB_URL = args.JOB_URL
BUILD_URL = JOB_URL + '/lastBuild'
def run_curl():
# 使用requests模塊代替subprocess執(zhí)行curl命令
try:
response = requests.get(JOB_URL + "api/json")
return response
except requests.exceptions.RequestException:
print(f"Error accessing {JOB_URL}")
return None
def check_network():
""" 檢查網絡 """
try:
response = requests.get(WEBHOOK)
return response.ok
except requests.exceptions.RequestException:
return False
def sending_alarms(text=None):
""" 發(fā)送信息 告警 """
network_status = subprocess.getoutput("networkctl status")
network_route = subprocess.getoutput("ip route show")
local_network_info = network_route
if text is None:
text = f"網絡異常: 無法訪問\n{BUILD_URL}\n{local_network_info}"
payload_message = {"msg_type": "text", "content": {"text": f"告警信息\n{text}"}}
headers = {"Content-Type": "application/json"}
res = requests.post(url=WEBHOOK, data=json.dumps(payload_message), headers=headers)
print(f"\t告警信息發(fā)送狀態(tài):{res.text}")
def get_build_info():
# 解析Jenkins構建信息
# ============ 數(shù)據(jù)獲取 ============
result = requests.get(f'{BUILD_URL}/api/json')
# print(result.request)
try:
result = result.json()
# ============ 數(shù)據(jù)獲取 ============
for index in result['actions']: # 節(jié)點:執(zhí)行
if index: # 節(jié)點:執(zhí)行 清洗節(jié)點 剔除空字段
# print(index)
if index['_class'] == 'hudson.model.CauseAction': # 查找節(jié)點 啟動信息
shortDescription = index['causes'][0]['shortDescription'] # 啟動著
print('\tshortDescription --> ', shortDescription)
if index['_class'] == 'jenkins.metrics.impl.TimeInQueueAction': # 查找節(jié)點 構建耗時
buildingDurationMillis = index['buildingDurationMillis'] # 時間毫秒
print('\tbuildingDurationMillis --> ', buildingDurationMillis)
if index['_class'] == 'hudson.plugins.git.util.BuildData': # 查找節(jié)點 git
for key, value in index['buildsByBranchName'].items(): # 設置變量 GIT_BRANCH
GIT_BRANCH = key
print('\tGIT_BRANCH --> ', GIT_BRANCH)
# print()
except (IndexError, KeyError, ValueError):
sending_alarms('jenkins 獲取數(shù)據(jù)異常')
exit(2)
if result['duration'] != 0: # 獲取字段
duration = int(result['duration']) // 1000 # 編譯持續(xù)時間
minutes, seconds = divmod(duration, 60)
hours, minutes = divmod(minutes, 60)
build_duration = f'{hours}h {minutes}m {seconds}s'
print(f"\t--> 通過響應值換算{build_duration}\n")
else: # 通過當前時間計算
build_duration = datetime.datetime.now() - datetime.datetime.fromtimestamp(result['timestamp'] / 1000)
build_duration = str(build_duration).split('.')[0]
print(f"\t--> 通過當前時間計算耗時{build_duration}\n")
fullDisplayName = result['fullDisplayName']
BUILD_STATUS = result['result'] # 編譯狀態(tài)
print('\tbuild_status --> ', BUILD_STATUS)
timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(result['timestamp'] / 1000)) # 編譯開始時間
print(f'\ttimestamp --> {timestamp}')
JOB_URL = result['url']
BUILD_DISPLAY_NAME = result['displayName']
JOB_NAME = fullDisplayName
return shortDescription, buildingDurationMillis, GIT_BRANCH, build_duration, fullDisplayName, BUILD_STATUS, timestamp, JOB_URL, BUILD_DISPLAY_NAME, JOB_NAME
def set_msg():
shortDescription, buildingDurationMillis, GIT_BRANCH, build_duration, fullDisplayName, BUILD_STATUS, timestamp, JOB_URL, BUILD_DISPLAY_NAME, JOB_NAME = get_build_info()
# ============ 設置樣式 ============
if "SUCCESS" in BUILD_STATUS: # 成功
template_color = "green"
elif "FAILURE" in BUILD_STATUS: # 失敗
template_color = "red"
elif "ABORTED" in BUILD_STATUS: # 終止
template_color = "yellow"
else:
template_color = "grey"
# ============ 卡片模板 ============
card = json.dumps({
"config": {
"wide_screen_mode": True
},
"elements": [
{
"tag": "markdown",
"content": f"觸發(fā)時間:{timestamp}\n"
f"分支名稱:{GIT_BRANCH}\n"
f"構建編號:{BUILD_DISPLAY_NAME}\n"
f"構建狀態(tài):<font color={template_color}>{BUILD_STATUS}</font>\n"
},
{
"tag": "note",
"elements": [
{
"tag": "img",
"img_key": f"{img_icon}",
"alt": {
"tag": "plain_text",
"content": f"{JOB_URL}"
}
},
{
"tag": "plain_text",
"content": f"{shortDescription}"
}
]
},
{
"tag": "hr"
},
{
"tag": "action",
"actions": [
{
"tag": "button",
"text": {
"tag": "plain_text",
"content": "報告鏈接"
},
"type": "primary",
"multi_url": {
"url": f"{BUILD_URL}/allure",
"pc_url": "",
"android_url": "",
"ios_url": ""
}
}
],
"layout": "bisected"
}
],
"header": {
"template": f"{template_color}",
"title": {
"content": f"Jenkins 構建狀態(tài)報告: {JOB_NAME}",
"tag": "plain_text"
}
}
})
body = json.dumps({"msg_type": "interactive", "card": card})
headers = {"Content-Type": "application/json"}
res = requests.post(url=WEBHOOK, data=body, headers=headers)
print(f'發(fā)送響應 --> {res.text}')
if __name__ == '__main__':
img_icon = 'img_v2_041b28e3-5680-48c2-9af2-497ace79333g' # img_key_id jenkinsIcon 上傳到飛書后得到的key
print(f"互聯(lián)網狀態(tài):{check_network()}")
if not check_network():
text = f"無法訪問互聯(lián)網"
print(text)
exit(2)
set_msg()
預覽 2 (多Job,概覽)1
Jenkins 需安裝
Multijob
插件
Multijob https://plugins.jenkins.io/jenkins-multijob-plugin/
-
對應shell
#!/usr/bin/env bash echo -e "\n\n 消息處理" # ======================== # 消息發(fā)送 # ======================== # 1. 消息 接收地址 # ----------------------- group='https://open.feishu.cn/open-apis/bot/v2/hook/' webhook_list=($group) py_send='SendMsgFeishu.py' # ======================== # 2. 文件處理 # ----------------------- echo -e "\n 復制發(fā)送消息腳本 $HOME -> ${WORKSPACE}" && cp "$HOME"/$py_send "${WORKSPACE}" # ======================== # 3. 消息 執(zhí)行發(fā)送 # ----------------------- run_send_msg() { for ((i = 0; i < ${#webhook_list[@]}; i++)); do webhook=${webhook_list[$i]} echo -e "發(fā)送中 --> $webhook" python3 $py_send "${webhook}" -job_url "${JOB_URL}" done echo -e "發(fā)送完成 \n\n" } run_send_msg
-
對應python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import argparse
import datetime
import json
import subprocess
import time
from json import JSONDecodeError
import requests
parser = argparse.ArgumentParser(description='Jenkins 發(fā)送消息到飛書',
epilog="執(zhí)行示例>>> python ${webhook} -job_url ${JOB_URL}")
parser.add_argument('webhook', help='機器人webhookURL') # 必填
parser.add_argument('-job_url', '--JOB_URL', help='作業(yè)URL', required=True, ) # 必填
webhook = parser.parse_args().webhook
JOB_URL = parser.parse_args().JOB_URL
BUILD_URL = JOB_URL + '/lastBuild'
job_name = [] # 運行名稱
job_duration = [] # 運行時長
job_status = [] # 運行狀態(tài)
job_url = [] # 運行結果
pass_rate = [] # 百分比顯示
print('修改時間:2023-07-04 10:02:43')
print(BUILD_URL.split('/')[2])
def get_jenkins_plugin():
""" 獲取信息 jenkins插件 """
url = f"http://{BUILD_URL.split('/')[2]}/pluginManager/api/json?depth=1"
print(url)
plugin_list = requests.get(url, auth=jenkins_auth).json()['plugins']
print(f"Jenkins插件安裝數(shù)量 {len(plugin_list)}")
# for plugin in plugin_list:
# name = plugin['longName']
# # print(name) # 插件名稱
return plugin_list
def get_base_info():
""" 獲取信息 system """
device_id = subprocess.getoutput('cat /etc/ding_issue')
version_os = subprocess.getoutput('cat /etc/issue')
version_browser = subprocess.getoutput('google-chrom-stable -version')
device_sn = subprocess.getoutput('hostname').split('-')[-1]
network_status = subprocess.getoutput("networkctl status")
network_route = subprocess.getoutput("ip route show")
print("設備信息".center(20, '-'))
print("\n設備唯一標識碼", device_id, "\n設備序列號", version_os)
print("\n系統(tǒng)版本", version_os, "\n瀏覽器版本", version_browser, )
print("\n本地網絡信息", network_status, "\n網絡優(yōu)先級(值小優(yōu)先)", network_route)
return device_id, version_os, version_browser, device_sn,
def sending_alarms(text=None):
""" 發(fā)送信息 告警 """
network_status = subprocess.getoutput("networkctl status")
network_route = subprocess.getoutput("ip route show")
local_network_info = network_route
if text is None:
text = f"網絡異常: 無法訪問\n{BUILD_URL}\n{local_network_info}"
payload_message = {"msg_type": "text", "content": {"text": f"告警信息\n{text}"}}
headers = {"Content-Type": "application/json"}
res = requests.post(url=webhook, data=json.dumps(payload_message), headers=headers)
print(f"告警信息發(fā)送狀態(tài):{res.text}")
def set_msg():
""" 發(fā)送信息 """
# ------------------
# ------ 數(shù)據(jù)獲取 ------
# ------------------
JENKINS_URL = BUILD_URL.split('job')[0] # JENKINS_URL
# ------ begin 登陸異常 ------
try:
get_result = requests.get(f'{BUILD_URL}/api/json', auth=jenkins_auth)
print(f"數(shù)據(jù)查詢API地址:{get_result.url}")
result = get_result.json()
# except JSONDecodeError: # json解析失敗 未登陸
# text = "Error 401 Unauthorized"
# sending_alarms(text)
# quit(text)
# except RecursionError:
# quit('遞歸錯誤:從 unicode 字符串解碼 JSON 對象時超出最大遞歸深度')
# except [OSError, TimeoutError]: # 異常列表: 網絡
# text = "No route to host"
# sending_alarms()
# quit(text)
except Exception as e: # 通用異常
text = f"發(fā)生異常: {type(e).__name__} --> {str(e)}\n檢查:{BUILD_URL}/api/json"
sending_alarms(text)
quit(text)
# res = requests.get(f'{BUILD_URL}/api/json', auth=jenkins_auth)
# if res.status_code == 401:
# quit('Error 401 Unauthorized')
# else:
# result = res.json()
# # ------ end 登陸異常 ------
shortDescription = result['actions'][0]['causes'][0]['shortDescription'] # 啟動者
timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(result['timestamp'] / 1000)) # 編譯開始時間
# ------ begin 獲取持續(xù)時間 ------
if result['duration'] != 0: # 獲取字段
duration = int(result['duration']) // 1000 # 編譯持續(xù)時間
minutes, seconds = divmod(duration, 60)
hours, minutes = divmod(minutes, 60)
build_duration = f'{hours}h {minutes}m {seconds}s'
print(f"--> 通過響應值換算{build_duration}\n")
else: # 通過當前時間計算
build_duration = datetime.datetime.now() - datetime.datetime.fromtimestamp(result['timestamp'] / 1000)
build_duration = str(build_duration).split('.')[0]
print(f"--> 通過當前時間計算耗時{build_duration}\n")
# ------ end 獲取持續(xù)時間 ------
total_count = len(result['subBuilds']) # 數(shù)量總計
print(f'======= 項目概覽 ======= \n'
f'shortDescription:{shortDescription}\nbuild duration:{build_duration}\ntotal_count:{total_count}\ntimestamp:{timestamp}\n')
print('提示: 沒有allure就報錯 無法運行 JSONDecodeError')
for index in result['subBuilds']: # 提取數(shù)據(jù)
# print(index)
# print(index['result'])
# 數(shù)據(jù)預處理
allure_summary = requests.get(f"{JENKINS_URL}{index['url']}/allure/widgets/summary.json")
allure_history_trend = requests.get(f"{JENKINS_URL}{index['url']}/allure/widgets/history-trend.json")
# print(allure_history_trend.request.url)
try: # 解析allure.json ## JAVA_HOME 導致allure報告生成失敗
allure_summary = allure_summary.json()['statistic']
try: # 獲取歷史數(shù)據(jù)
allure_history_trend = allure_history_trend.json()[1]
except IndexError:
print('沒有歷史數(shù)據(jù)')
allure_history_trend = allure_history_trend.json()[0]
# 計算百分比
if allure_summary['total'] != 0: # 除數(shù)不能為0
allure_pass_rate = allure_summary['passed'] / allure_summary['total']
else:
allure_pass_rate = 0 # 除數(shù)不能為0
if allure_history_trend['data']['total'] != 0:
allure_history = allure_history_trend['data']['passed'] / allure_history_trend['data']['total']
else:
allure_history = 0
except Exception as e:
text = f"解析Allure數(shù)據(jù)異常,請檢查\n發(fā)生異常: {type(e).__name__} --> {str(e)}\n檢查:{BUILD_URL}/api/json"
sending_alarms(text)
allure_pass_rate = allure_history = 0
allure_summary = {"total": 0, "passed": 0, }
# ------------------
# ------ 設置樣式 ------
# ------------------
if "SUCCESS" == index['result']: # 成功
color = "green"
elif "FAILURE" == index['result']: # 失敗
color = "red"
elif "ABORTED" == index['result']: # 中止
color = "yellow"
else: # 其他
color = "grey"
# 通過率對比
allure_change = allure_pass_rate - allure_history
print(f"{index['jobName']} --> 本次比上次通過率 {allure_change}")
if allure_pass_rate > allure_history:
allure_pass_rate = f'<font color=green>↑{allure_pass_rate:.2%}</font>'
elif allure_pass_rate < allure_history or allure_pass_rate == 0:
allure_pass_rate = f'<font color=red>↓{allure_pass_rate:.2%}</font>'
else:
allure_pass_rate = f' {allure_pass_rate:.2%}'
# ------------------
# ------ 載入數(shù)據(jù) ------
# ------------------
job_name.append({"tag": "markdown", "content": f"{index['jobName']}", "text_align": "center"})
job_duration.append({"tag": "markdown", "content": f"{index['duration']}", "text_align": "center"})
job_status.append(
{"tag": "markdown", "content": f"<font color={color}>{index['result']}</font>", "text_align": "center"})
job_url.append(
{"tag": "markdown", "content": f"[查看]({JENKINS_URL}{index['url']}/allure)", "text_align": "center"})
pass_rate.append(
{"tag": "markdown",
"content": f"{allure_summary['passed']}/{allure_summary['total']}{allure_pass_rate} ",
"text_align": "center"})
print(f'======= 項目詳情 ======= \n{job_name}\n{job_duration}\n{job_status}\n{pass_rate}\n{job_url}\n')
# ------------------
# ------ 卡片模板 ------
# ------------------
card = json.dumps({
"elements": [
{
"tag": "markdown",
"content": "**項目總覽**\n"
},
{
"tag": "column_set",
"flex_mode": "bisect",
"background_style": "grey",
"horizontal_spacing": "default",
"columns": [
{
"tag": "column",
"width": "weighted",
"weight": 1,
"elements": [
{
"tag": "markdown",
"text_align": "center",
"content": f"項目數(shù)量\n**{total_count}**\n"
}
]
},
{
"tag": "column",
"width": "weighted",
"weight": 1,
"elements": [
{
"tag": "markdown",
"text_align": "center",
"content": f"運行耗時\n**{build_duration}**"
}
]
},
{
"tag": "column",
"width": "weighted",
"weight": 1,
"elements": [
{
"tag": "markdown",
"text_align": "center",
"content": f"執(zhí)行時間\n**{timestamp}**\n"
}
]
}
]
},
{
"tag": "markdown",
"content": "**項目信息**"
},
{
"tag": "column_set",
"flex_mode": "none",
"background_style": "grey",
"columns": [
{
"tag": "column",
"width": "weighted",
"weight": 2,
"vertical_align": "top",
"elements": [
{
"tag": "markdown",
"content": "**項目名**",
"text_align": "center"
}
]
},
{
"tag": "column",
"width": "weighted",
"weight": 1,
"vertical_align": "top",
"elements": [
{
"tag": "markdown",
"content": "**運行時長**",
"text_align": "center"
}
]
},
{
"tag": "column",
"width": "weighted",
"weight": 1,
"vertical_align": "top",
"elements": [
{
"tag": "markdown",
"content": "**運行狀態(tài)**",
"text_align": "center"
}
]
},
{
"tag": "column",
"width": "weighted",
"weight": 1,
"vertical_align": "top",
"elements": [
{
"tag": "markdown",
"content": "**passed/total/通過率**",
"text_align": "center"
}
]
},
{
"tag": "column",
"width": "weighted",
"weight": 1,
"vertical_align": "top",
"elements": [
{
"tag": "markdown",
"content": "**Report**",
"text_align": "center"
}
]
}
]
},
{
"tag": "column_set",
"flex_mode": "none",
"background_style": "default",
"columns": [
{
"tag": "column",
"width": "weighted",
"weight": 2,
"vertical_align": "top",
"elements": job_name
},
{
"tag": "column",
"width": "weighted",
"weight": 1,
"vertical_align": "center",
"elements": job_duration
},
{
"tag": "column",
"width": "weighted",
"weight": 1,
"vertical_align": "top",
"elements": job_status
},
{
"tag": "column",
"width": "weighted",
"weight": 1,
"vertical_align": "top",
"elements": pass_rate
},
{
"tag": "column",
"width": "weighted",
"weight": 1,
"vertical_align": "top",
"elements": job_url
}
]
},
{
"tag": "hr"
},
{
"tag": "note",
"elements": [
{
"tag": "img",
"img_key": f"{img_icon}",
"alt": {
"tag": "plain_text",
"content": ""
}
},
{
"tag": "plain_text",
"content": f"{shortDescription}\n顏色代表對比上次執(zhí)行,綠色上升,紅色下降"
}
]
}
]
})
card_id = json.dumps({
"type": "template",
"data": {
"template_id": "ctp_AA6DZMfkJekh", # 卡片id,參數(shù)必填。可在搭建工具中通過“復制卡片ID”獲取
"template_variable": # 卡片中綁定的變量的取值。如沒有綁定變量,可不填此字段。
{
"total_count": "29",
"group_table": [
{
"jobName": "test001",
"duration": "小于1小時",
"build_url": "baidu.com",
"build_status": "SUCCESS",
"tmp_color": "green"
},
{
"jobName": "test002",
"duration": "2小時",
"build_url": "baidu.com",
"build_status": "FAILURE",
"tmp_color": "red"
},
{
"jobName": "test003",
"duration": "3小時",
"build_url": "baidu.com",
"build_status": "ABORTED",
"tmp_color": "yellow"
},
{
"jobName": "test004",
"duration": "3小時",
"build_url": "baidu.com",
"build_status": "UNSTABLE",
"tmp_color": "grey"
}
],
"duration": "15080",
"shortDescription": "Started by user admin",
"timestamp": "1686645721264",
"jobName": "",
"tmp_color": ""
}
}
})
body = json.dumps({"msg_type": "interactive", "card": card}) # 使用 當前模板
# body = json.dumps({"msg_type": "interactive", "card": card_id}) # 使用 預置模板
headers = {"Content-Type": "application/json"}
res = requests.post(url=webhook, data=body, headers=headers)
print(f'消息發(fā)送響應 -->\n\t {res.text}')
if __name__ == '__main__':
img_icon = 'img_v2_098e80ae-e583-4148-b822-f42a05298d3g' # img_key_id jenkinsIcon
# jenkins_auth = ('result', 'result') # jenkins User:Pwd
jenkins_auth = ('admin', '1') # jenkins User:Pwd
# get_base_info()
################################
# ## 方法一
set_msg()
################################
################################
# ## 方法二
# ## 判斷 jenkins 插件信息
# deviceid, os_version, browser_version, sn, = get_base_info()
# print("設備信息".center(20, '*'), deviceid, os_version, browser_version, sn, )
# if 'Multijob' in str(get_jenkins_plugin()): # 檢查插件是否存在
# print(True)
# # set_msg()
# else:
# quit('插件不存在,無法執(zhí)行')
################################
GitLab 根據(jù) labels 統(tǒng)計 issue
需要創(chuàng)建個人令牌(token)文章來源:http://www.zghlxwxcb.cn/news/detail-672755.html
#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import argparse
import csv
import datetime
import os
import time
"""
API文檔
https://docs.gitlab.cn/jh/api/api_resources.html
"""
try:
import pandas as pd
import requests
except ImportError:
print("依賴缺失 安裝依賴...")
os.system("pip install requests pandas")
import pandas as pd
import requests
parser = argparse.ArgumentParser(description='gitlab BUG列表查詢',
epilog="python get_gitlab_bug.py -t {token} -gid {$gid} -begin 2023-09-01 -status all "
"-labels 'bug solution::fixed' -grep {file}")
parser.add_argument('-t', '--token', help='個人令牌', required=True) # 必填
group = parser.add_mutually_exclusive_group()
group.add_argument('-gid', '--group_id', help='群組ID')
group.add_argument('-pid', '--project_id', help='項目ID')
parser.add_argument('-labels', '--labels', default='bug', help='可自定義輸入,快捷標簽(bug, test_request, bug_fix)')
parser.add_argument('-status', '--labels_status', default='opened', choices=['all', 'closed', 'opened'],
help='標簽狀態(tài)')
parser.add_argument('-begin', '--created_after', help='指定創(chuàng)建時間之后', )
parser.add_argument('-end', '--created_before', help='指定創(chuàng)建時間之前', )
parser.add_argument('-sort', '--sort', default='desc', choices=['desc', 'asc'],
help='按照 asc 或 desc 的順序返回議題。默認值是 desc', )
parser.add_argument('-grep', '--grep_author', help='匹配issue作者', )
token = parser.parse_args().token
labels = parser.parse_args().labels
labels_status = parser.parse_args().labels_status
group_id = parser.parse_args().group_id
project_id = parser.parse_args().project_id
created_after = parser.parse_args().created_after
created_before = parser.parse_args().created_before
sort = parser.parse_args().sort
grep_author = parser.parse_args().grep_author
session = requests.session()
session.headers.update({'PRIVATE-TOKEN': token})
base = "http://gitlab.example.com/api/v4/"
issue_list = [] # issue列表
list_id = [] # 項目ID
project_info = {} # 項目信息
write_name = f"export_{labels_status}-{labels}.{time.strftime('%Y-%m-%d', time.localtime())}.csv"
print(write_name)
'''查詢 類別及地址'''
if group_id is not None:
url_get_labels = f"{base}groups/{group_id}/labels" # 獲取群組 labels
url_get_issue = f"{base}groups/{group_id}/issues" # 獲取群組 issue
url_get_count = f"{base}groups/{group_id}/issues_statistics" # 群組信息統(tǒng)計
elif project_id is not None:
url_get_labels = f"{base}projects/{project_id}/labels" # 獲取項目 labels
url_get_issue = f"{base}projects/{project_id}/issues" # 獲取項目 issue
url_get_count = f"{base}projects/{project_id}/issues_statistics" # 項目信息統(tǒng)計
else:
print(f"project --> {project_id}\t group --> {group_id}")
print("請輸入 project_id 或 group_id")
quit()
'''快捷標簽 設置'''
if labels is None:
labels = "bug"
elif labels == "test_request":
labels = "Test Request"
elif labels == "release_request":
labels = "Release Request"
elif labels == "bug_fix":
labels = "bug solution::fixed"
'''查詢 參數(shù)'''
querystring = {"labels": labels, "scope": "all", "per_page": "100"} # 標簽 范圍 頁面數(shù)量
# 查詢 指定時間之后
if created_after is not None:
begin_time = f"{created_after}T08:00:00Z"
querystring.update({"created_after": begin_time})
if created_before is not None:
end_time = f"{created_before}T08:00:00Z"
querystring.update({"created_before": end_time})
def get_labels():
response = session.request("GET", url_get_labels).json()
print("當前標記列表".center(50, '-'))
for i, dic in enumerate(response):
print(f'{i + 1}\t{dic["name"]} --> {dic["description"]}')
def get_total_issue():
print("\n", "查詢概覽".center(50, '-'))
response = session.request("GET", url_get_count, params=querystring)
# print(response.status_code)
response = response.json()
# print("->>", response)
if '404' in str(response) or '401' in str(response):
print('查詢異常', response)
quit()
else:
print(f"\nIssue Total --> labels >> {labels}\n\t{response['statistics']['counts']}")
def get_label_time(issue, label_name='bug solution::fixed'):
label_create_time = '' # 標簽創(chuàng)建時間
# print(issue)
# print(get_issue['id'], get_issue['iid'], get_issue['project_id'], get_issue['state'])
if 'fix' in str(issue['labels']):
# print('\t fixed')
events = session.get(f"{base}projects/{issue['project_id']}/issues/{issue['iid']}/resource_label_events")
# print(events.request.url)
events = events.json()
# print(events)
for event in events:
if event['label']['name'] == label_name:
label_create_time = event['created_at']
# if 'result_fix' not in locals():
# label_create_time = ''
# print('\t fixed-->', label_create_time)
# return label_create_time
return label_create_time
def get_issue():
""" 獲取 labels """
querystring.update({"state": labels_status})
print(f"查詢參數(shù)--> {querystring}")
response = session.request("GET", url_get_issue, params=querystring)
# a = response.headers.get('Link') # 頁面鏈接
# b = response.headers.get('X-Next-Page') # 下一頁碼
# c = response.headers.get('X-Page') # 當前頁碼
# d = response.headers.get('X-Per-Page') # 頁面數(shù)量
e = response.headers.get('X-Total') # 總數(shù)據(jù)量
f = response.headers.get('X-Total-Pages') # 總頁數(shù)
# print(a, b, c, d, e, f, )
print(f"\n--> 查詢{labels}狀態(tài) {labels_status} 總數(shù)量 {e} 總頁數(shù) {f}")
for i in range(int(f)):
i += 1
querystring.update({"page": f"{i}"}) # 添加頁碼參數(shù)
# print(querystring)
data = session.request("GET", url_get_issue, params=querystring).json()
print(f"\t當前第 {i} 頁Issue數(shù)量", len(data))
issue_list.extend(data) # 列表添加元素
for item in data: # 提取項目ID
# print(item)
list_id.append(item["project_id"])
# print("run get group Issue OK")
def search_name(project):
""" 查詢項目信息 """
url = f"{base}projects/{project}" # 獲取項目信息
# print(url)
response = session.request("GET", url, ).json()
# print(response)
return response
def get_projectname():
""" 提取項目名稱 """
for index in set(list_id): # 列表 轉 集合
data = search_name(index)
# print(data["id"], data["name"])
project_info.update({data["id"]: data["name"]})
# print(len(issue_list))
today = datetime.date.today()
first_day_of_year = datetime.date(today.year, 1, 1)
diff = today - first_day_of_year
current_week = diff.days // 7 + 1
print(
f"\n--> 今天是第{current_week}周\t存在'{labels}'項目 數(shù)量{len(project_info.values())}\n\t{project_info.values()}")
def format_issue_list():
""" 提交人 匹配"""
if not grep_author:
print(f'\t提交人:不匹配指定作者')
print('\t如需匹配指定作者,請創(chuàng)建一個文件,內容為`所匹配的作者姓名` 執(zhí)行時添加對應參數(shù) `-grep {file_name}`')
pass
print(f'-->不匹配數(shù)量 {len(issue_list)}')
else:
with open('./group.txt', 'r') as read_file:
name_grep = read_file.read().split()
print(f'\t提交人:匹配指定作者 {name_grep}')
issue_list_tmp = issue_list.copy()
for issue in issue_list_tmp:
if issue["author"]["name"] not in str(name_grep):
issue_list.remove(issue)
print(f'-->匹配后剩余數(shù)量 {len(issue_list)}')
def save_csv():
""" 記錄到csv """
header_list = ["項目Name", "BUG_ID", "BUG標題", "狀態(tài)", "標簽", "提交人", "指派人", "打開天數(shù)", "web_url",
"打開時間", "修復時間", "關閉時間", '公式_標簽', ]
with open(write_name, mode="w", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, header_list)
writer.writeheader()
# print("寫入文件數(shù)據(jù)量", len(issue_list))
for idx, issue in enumerate(issue_list): # 獲取 項目信息
# print('issue_list索引-->', idx + 1, ' ', end='\r') # 簡易進度條
print(f'issue_list剩余數(shù)量-->', len(issue_list) - idx - 1, ' ', end='\r') # 簡易進度條
if issue["state"] == "opened":
count_days = int(
(time.time() - time.mktime(time.strptime(issue["created_at"].split('T')[0], '%Y-%m-%d'))) / (
24 * 60 * 60)) # 計算距今天數(shù)
# count_days = formula_since_now = (f'=today()-OFFSET(INDIRECT(ADDRESS(ROW(),COLUMN())),0,2)') # 公式_距今
else:
count_days = 0
assignees = []
for index in issue["assignees"]: # 指派人
assignees.append(index['name'])
created_time = issue['created_at'].split('T')[0] # 創(chuàng)建時間
fixed_time = get_label_time(issue).split('T')[0] # 修復時間
if issue['closed_at'] is None: # 關閉判斷
issue['closed_at'] = '' # open狀態(tài)的沒有關閉時間
closed_time = issue['closed_at'].split('T')[0] # 關閉時間
issue['title'] = str(issue['title']).replace('"', "'") # 替換標題中的雙引號
issue['title'] = f'=HYPERLINK("{issue["web_url"]}","{issue["title"]}")' # 設置超鏈接
# excel 過濾公式
# 判斷(統(tǒng)計判斷(單元格偏移(單元格引用文本(當前行, 當前列))), 列偏移量,行偏移量)
# formula_labels = (
# '=(IF(COUNTIF(OFFSET(INDIRECT(ADDRESS(ROW(),COLUMN())),0,-7),"*low*")=0,IF(COUNTIF(OFFSET(INDIRECT(ADDRESS(ROW(),COLUMN())),0,-7),"*medium*")=0,IF(COUNTIF(OFFSET(INDIRECT(ADDRESS(ROW(),COLUMN())),0,-7),"*high*")=0,"無等級標簽","高"),"中"),"低"))')
if 'bug' in str(issue['labels']):
if 'low' in str(issue['labels']):
formula_labels = '低'
elif 'medium' in str(issue['labels']):
formula_labels = '中'
elif 'high' in str(issue['labels']):
formula_labels = '高'
else:
formula_labels = '未知等級'
else:
formula_labels = '未知等級'
writer.writerow(
{"項目Name": project_info[issue["project_id"]], "BUG_ID": issue["iid"], "BUG標題": issue["title"],
"狀態(tài)": issue["state"], "標簽": issue["labels"], "提交人": issue["author"]["name"],
"指派人": assignees, "打開天數(shù)": count_days, "web_url": issue["web_url"], "打開時間": created_time,
"修復時間": fixed_time, "關閉時間": closed_time, '公式_標簽': formula_labels, })
pwd = os.getcwd().replace('\\', '/') # 獲取當前路徑
print(f'\n-------- write done -------> file://{pwd}/{write_name}')
def sort():
""" 排序 """
print("排序中 ...")
pd.set_option('display.max_columns', None)
pd.set_option('display.max_rows', None)
data = pd.read_csv(write_name, index_col="項目Name")
data.sort_values('項目Name', inplace=True)
# test = data.sort_values('項目Name', inplace=True)
# print(test)
data.to_csv(write_name)
print("\n按 項目Name 排序 done")
if __name__ == '__main__':
os.system("pip install requests pandas")
# get_labels() # 獲取 標記列表
get_issue()
get_projectname()
format_issue_list()
save_csv()
sort()
自用模板
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# @Time : 2023-08-04 10:02:43
import argparse
import datetime
import json
import subprocess
import time
import requests
parser = argparse.ArgumentParser(description='Jenkins多JOB集合 發(fā)送消息到飛書',
epilog="執(zhí)行示例>>> python ${webhook} -job_url ${JOB_URL}")
parser.add_argument('webhook', help='機器人webhookURL') # 必填
parser.add_argument('-job_url', '--JOB_URL', help='作業(yè)URL', required=True, ) # 必填
webhook = parser.parse_args().webhook
JOB_URL = parser.parse_args().JOB_URL
BUILD_URL = JOB_URL + '/lastBuild'
job_name = [] # 運行名稱
job_duration = [] # 運行時長
job_status = [] # 運行狀態(tài)
job_url = [] # 運行結果
pass_rate = [] # 百分比顯示
#########
print('腳本修改時間:2023-09-18 10:02:43') # 修改allure執(zhí)行報錯
#########
session = requests.Session()
def set_msg():
# ------------------
# ------ 數(shù)據(jù)獲取 ------
# ------------------
JENKINS_URL = BUILD_URL.split('job')[0] # JENKINS_URL
# ------ begin 登陸異常 ------
try:
get_result = requests.get(f'{BUILD_URL}/api/json', auth=jenkins_auth)
print(f"數(shù)據(jù)查詢API地址 {get_result.url}")
result = get_result.json()
# except JSONDecodeError: # 解析異常 未登錄
# quit('Error 401 Unauthorized')
except OSError: # 網絡異常
msg = {"msg_type": "text", "content": {"text": "獲取報告地址異常/無法訪問"}}
res_msg = requests.post(url=webhook, data=json.dumps(msg), headers={"Content-Type": "application/json"})
print(res_msg.text)
quit('獲取報告地址異常/無法訪問')
except Exception as e:
text = f"消息發(fā)送發(fā)生異常狀態(tài)\t{type(e).__name__} --> {str(e)}\n檢查API數(shù)據(jù)\t請手動前{BUILD_URL}查看報告"
msg = {"msg_type": "text", "content": {"text": text}}
res_msg = requests.post(url=webhook, data=json.dumps(msg), headers={"Content-Type": "application/json"})
print(res_msg.text)
quit(text)
# res = requests.get(f'{BUILD_URL}/api/json', auth=jenkins_auth)
# if res.status_code == 401:
# quit('Error 401 Unauthorized')
# else:
# result = res.json()
# # ------ end 登陸異常 ------
shortDescription = result['actions'][0]['causes'][0]['shortDescription'] # 啟動者
timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(result['timestamp'] / 1000)) # 編譯開始時間
# ------ begin 獲取持續(xù)時間 ------
if result['duration'] != 0: # 獲取字段
duration = int(result['duration']) // 1000 # 編譯持續(xù)時間
minutes, seconds = divmod(duration, 60)
hours, minutes = divmod(minutes, 60)
build_duration = f'{hours}h {minutes}m {seconds}s'
else: # 通過當前時間計算
print("\t通過當前時間 計算持續(xù)時間")
build_duration = datetime.datetime.now() - datetime.datetime.fromtimestamp(result['timestamp'] / 1000)
build_duration = str(build_duration).split('.')[0]
# ------ end 獲取持續(xù)時間 ------
total_count = len(result['subBuilds']) # 數(shù)量總計
print(f'\n\n======= 項目概覽 ======= \n'
f'shortDescription:{shortDescription}\nbuild duration:{build_duration}\ntotal_count:{total_count}\ntimestamp:{timestamp}\n')
print('****** 提示: 沒有allure就報錯 無法運行 JSONDecodeError ******')
for index in result['subBuilds']: # 提取數(shù)據(jù)
# print(index)
# print(index['result'])
# 數(shù)據(jù)預處理
allure_summary = requests.get(f"{JENKINS_URL}{index['url']}/allure/widgets/summary.json", auth=jenkins_auth)
allure_history_trend = requests.get(f"{JENKINS_URL}{index['url']}/allure/widgets/history-trend.json", auth=jenkins_auth)
try:
allure_summary = allure_summary.json()['statistic']
# print(allure_history_trend.request.url)
# allure_pass_rate = round(allure_pass_rate, 2) if allure_pass_rate != 100 else 100 # 保留2位小數(shù) 值為100不保留小數(shù)
try:
allure_history_trend = allure_history_trend.json()[1]
except IndexError:
print('沒有歷史數(shù)據(jù)')
allure_history_trend = allure_history_trend.json()[0]
if allure_summary['total'] != 0: # 除數(shù)不能為0
allure_pass_rate = allure_summary['passed'] / allure_summary['total']
else:
allure_pass_rate = 0 # 除數(shù)不能為0
if allure_history_trend['data']['total'] != 0:
allure_history = allure_history_trend['data']['passed'] / allure_history_trend['data']['total']
else:
allure_history = 0
except Exception as e:
print("獲取Allure報告異常", type(e).__name__, str(e))
allure_pass_rate = allure_history = 0
allure_summary = {'passed': None, 'total': None}
# print("數(shù)據(jù)獲取完成 按指定樣式格式化中")
# ------------------
# ------ 設置樣式 ------
# ------------------
# 通過率對比
allure_change = allure_pass_rate - allure_history
print(f"\t{index['jobName']} --> 本次比上次通過率 {allure_change}")
if allure_pass_rate > allure_history:
allure_pass_rate = f'<font color=green>↑{allure_pass_rate:.2%}</font>'
elif allure_pass_rate < allure_history or allure_pass_rate == 0:
allure_pass_rate = f'<font color=red>↓{allure_pass_rate:.2%}</font>'
else:
allure_pass_rate = f' {allure_pass_rate:.2%}'
# ------------------
# ------ 載入數(shù)據(jù) ------
# ------------------
job_name.append({"tag": "markdown", "content": f"{index['jobName']}", "text_align": "center"})
job_duration.append({"tag": "markdown", "content": f"{index['duration']}", "text_align": "center"})
job_status.append(
{"tag": "markdown", "content": f"{index['result']}", "text_align": "center"})
job_url.append(
{"tag": "markdown", "content": f"[查看]({JENKINS_URL}{index['url']}/allure)", "text_align": "center"})
pass_rate.append(
{"tag": "markdown",
"content": f"{allure_summary['passed']}/{allure_summary['total']}{allure_pass_rate}",
"text_align": "center"})
print(
f'\n\n======= 項目詳情 ======= \n{job_name}\n{job_duration}\n{job_status}\n{pass_rate}\n{job_url}\n\t{allure_change}')
# ------------------
# ------ 卡片模板 ------
# ------------------
card = json.dumps({
"elements": [
{
"tag": "markdown",
"content": f"**項目總覽**"
},
{
"tag": "column_set",
"flex_mode": "bisect",
"background_style": "grey",
"horizontal_spacing": "default",
"columns": [
{
"tag": "column",
"width": "weighted",
"weight": 1,
"elements": [
{
"tag": "markdown",
"text_align": "center",
"content": f"項目數(shù)量\n**{total_count}**\n"
}
]
},
{
"tag": "column",
"width": "weighted",
"weight": 1,
"elements": [
{
"tag": "markdown",
"text_align": "center",
"content": f"執(zhí)行時間\n**{timestamp}**\n"
}
]
},
{
"tag": "column",
"width": "weighted",
"weight": 1,
"elements": [
{
"tag": "markdown",
"text_align": "center",
"content": f"持續(xù)時間\n**{build_duration}**\n"
}
]
},
]
},
{
"tag": "markdown",
"content": "**項目信息**"
},
{
"tag": "column_set",
"flex_mode": "none",
"background_style": "grey",
"columns": [
{
"tag": "column",
"width": "weighted",
"weight": 3,
"vertical_align": "top",
"elements": [
{
"tag": "markdown",
"content": "**項目名**",
"text_align": "center"
}
]
},
{
"tag": "column",
"width": "weighted",
"weight": 2,
"vertical_align": "top",
"elements": [
{
"tag": "markdown",
"content": "**運行時長**",
"text_align": "center"
}
]
},
{
"tag": "column",
"width": "weighted",
"weight": 2,
"vertical_align": "top",
"elements": [
{
"tag": "markdown",
"content": "**passed/total/通過率**",
"text_align": "center"
}
]
},
{
"tag": "column",
"width": "weighted",
"weight": 1,
"vertical_align": "top",
"elements": [
{
"tag": "markdown",
"content": "**Report**",
"text_align": "center"
}
]
}
]
},
{
"tag": "column_set",
"flex_mode": "none",
"background_style": "default",
"columns": [
{
"tag": "column",
"width": "weighted",
"weight": 3,
"vertical_align": "top",
"elements": job_name
},
{
"tag": "column",
"width": "weighted",
"weight": 2,
"vertical_align": "center",
"elements": job_duration
},
{
"tag": "column",
"width": "weighted",
"weight": 2,
"vertical_align": "top",
"elements": pass_rate
},
{
"tag": "column",
"width": "weighted",
"weight": 1,
"vertical_align": "top",
"elements": job_url
}
]
},
{
"tag": "hr"
},
{
"tag": "note",
"elements": [
{
"tag": "img",
"img_key": f"{img_icon}",
"alt": {
"tag": "plain_text",
"content": ""
}
},
{
"tag": "plain_text",
"content": f"{shortDescription}\n通過率 與上次執(zhí)行變化:綠色為上升,紅色下降"
},
{
"tag": "plain_text",
"content": f"系統(tǒng):{os_version}\n瀏覽器:{browser_version}\nDeviceID:{deviceid}\nSN:{sn}"
}
]
}
]
})
body = json.dumps({"msg_type": "interactive", "card": card}) # 使用 當前模板
headers = {"Content-Type": "application/json"}
res = requests.post(url=webhook, data=body, headers=headers)
print(f'消息發(fā)送響應 -->\n\t {res.text}')
if __name__ == '__main__':
img_icon = 'img_v2_098e80ae-e583-4148-b822-f42a05298d3g' # img_key_id jenkinsIcon
jenkins_auth=("auto", "1")
set_msg()
-
python ??文章來源地址http://www.zghlxwxcb.cn/news/detail-672755.html
到了這里,關于【無標題】jenkins消息模板(飛書)的文章就介紹完了。如果您還想了解更多內容,請在右上角搜索TOY模板網以前的文章或繼續(xù)瀏覽下面的相關文章,希望大家以后多多支持TOY模板網!