领导想知道目前 Jenkins 运行的所有自动化任务的最新情况，包括这些任务是否都执行了，执行的结果如何等等，需要通过时间可以过滤出来。
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
@File: job_monitor_db.py
@Time: 2022-11-17 13:16
@Desc: 创建数据库
'''
import time
import pymysql
def connect_db():
"""
"""
# url = "127.0.0.1"
# name = "root"
# pwd = "123456"
# dataBase = "data_monitor"
# return pymysql.connect(host=url, port=3306, user=name, passwd=pwd, db=dataBase)
url = "xxx"
name = "xxx"
pwd = "xxxx"
dataBase = "automation"
return pymysql.connect(host=url, port=9015, user=name, passwd=pwd, db=dataBase)
def query_database(db, sql):
cursor = db.cursor()
try:
if isinstance(sql, str):
cursor.execute(sql)
result = list(cursor.fetchall())
else:
result = []
for sq in sql:
cursor.execute(sq)
result.append(list(cursor.fetchall()))
except Exception as err:
result = ''.join(('An db query exception happened: ', str(err)))
db.close() # 关闭数据库连接
return result
def update_db(db, sql):
cursor = db.cursor()
try:
if isinstance(sql, str):
cursor.execute(sql)
db.commit()
else:
print('sql 不是一个合格的字符串：{}'.format(sql))
except Exception as err:
result = ''.join(('An db update exception happened: ', str(err)))
db.rollback()
print(result)
# 数据库数据插入更新
def db_insert(db, sql):
cursor = db.cursor()
i = 0
try:
cursor.execute(sql)
db.commit()
result = 'db insert success'
except Exception as err:
db.rollback()
result = 'An db insert exception happened: ' + str(err) + ' ' + str(i)
db.close() # 关闭数据库连接
return result
def gov_insert(db, sql):
cursor = db.cursor()
i = 0
try:
cursor.execute(sql)
db.commit()
result = 'db insert success'
except Exception as err:
db.rollback()
result = 'An db insert exception happened: ' + str(err) + ' ' + str(i)
return result
def close_db(db):
try:
db.close()
except Exception as err:
result = ''.join(('An db closed exception happened: ', str(err)))
print(result)
def get_current_time():
return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
# !/usr/bin/python
# -*- coding: utf-8 -*-
"""
@File : job_case_monitor.py
@Create Time: 2024-07-25 13:47
@Description: Jenkins上执行的所有job任务
"""
import json
import time
from datetime import datetime
import requests
import base64
import logging
# 配置日志
from job_monitor_db import *
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def get_jobs_recursively(root_url, jobs, auth_header, current_name=''):
"""递归获取 Jenkins 作业信息"""
try:
response = requests.get(root_url, headers=auth_header)
response.raise_for_status()
data = response.json()
except requests.exceptions.RequestException as e:
logging.error(f"请求失败：{e}")
return
for item in data.get('jobs', []):
current_name = item['name']
if '_class' in item and 'folder' in item['_class']:
new_path_url = root_url.replace('/api/json', f"/job/{current_name}/api/json")
get_jobs_recursively(new_path_url, jobs, auth_header, current_name)
elif '_class' in item and 'FreeStyleProject' in item['_class']:
full_root_url = root_url.replace('/api/json', f"/job/{current_name}/api/json")
get_jobs_recursively(full_root_url, jobs, auth_header, current_name)
if 'lastBuild' in data and data['lastBuild']:
last_build_data = data['lastBuild']
last_build_url = root_url.replace('/api/json', f"/{last_build_data['number']}/api/json")
try:
response = requests.get(last_build_url, headers=auth_header)
response.raise_for_status()
build_data = response.json()
logging.info(f"Build data: {build_data}") # 打印build_data查看结构
except requests.exceptions.RequestException as e:
logging.error(f"请求失败：{e}")
return
try:
last_report_url = root_url.replace('/api/json', f"/HTMLReport/")
response = requests.get(last_report_url, headers=auth_header)
if response.status_code == 200:
# 保存 last_report_url
logging.info(f"Saving URL: {last_report_url}")
else:
# 尝试另一个URL
last_report_url = root_url.replace('/api/json', f"/HTML_20Report/")
response = requests.get(last_report_url, headers=auth_header)
if response.status_code == 200:
# 保存 last_report_url
logging.info(f"Saving URL: {last_report_url}")
else:
logging.error("Neither URL returned a 200 status code.")
except Exception as e:
logging.error(f"An error occurred: {e}")
last_report_url = "" # 在异常情况下也设置为默认值
trigger_name = "jenkins timer"
for action in build_data.get('actions', []):
if action.get('_class') == 'hudson.model.CauseAction':
for cause in action.get('causes', []):
if cause.get('_class') == 'hudson.model.Cause$UserIdCause':
trigger_name = cause.get('userName')
break
last_timestamp = datetime.fromtimestamp(build_data['timestamp'] / 1000).strftime('%Y-%m-%d %H:%M:%S')
last_result = build_data.get('result', 'UNKNOWN')
job_info = {
"trigger_name": trigger_name,
"current_project": current_name,
"last_build_url": last_build_url,
"last_timestamp": last_timestamp,
"last_result": last_result,
"last_report_url": last_report_url
}
jobs.append(job_info)
# logging.info(f"Job info added: {job_info}")
def fetch_jenkins_jobs(jenkins_url, username, token):
"""获取 Jenkins 作业信息"""
auth_string = f"{username}:{token}"
auth_bytes = auth_string.encode('ascii')
auth_base64_bytes = base64.b64encode(auth_bytes)
auth_header = {'Authorization': f'Basic {auth_base64_bytes.decode("ascii")}'}
jobs = []
root_url = f"{jenkins_url}/api/json"
get_jobs_recursively(root_url, jobs, auth_header, '')
return jobs
# 使用示例
jenkins_url = 'http://xxxxxx:8080'
username = 'admin'
token = 'xxxxxxxxxx'
if __name__ == '__main__':
logging.info("正在请求中，请等待。。。。。。。。。。。。。")
jobs = fetch_jenkins_jobs(jenkins_url, username, token)
# 将获取到的作业信息发送到 Flask 应用
flask_url = 'http://127.0.0.1:5000/api/tasks/batch' # Flask 应用运行的 URL
headers = {
'Content-Type': 'application/json; charset=utf-8' # 设置内容类型和字符编码
}
execid = int(time.time())
# # 收集所有作业信息到一个列表
# all_jobs_info = []
for job in jobs:
logging.info(f"job: {job}")
trigger_name = job["trigger_name"]
current_project = job["current_project"]
last_build_url = job["last_build_url"]
last_timestamp = job["last_timestamp"]
last_result = job["last_result"]
last_report_url = job["last_report_url"]
# 连接数据库
db = connect_db()
logging.info("------------------------ 开始写入数据库 ---------------------------")
message_sql = "insert INTO Jenkins_Jobs (execid, trigger_name, current_project, last_build_url, last_timestamp, last_result, last_report_url) VALUES ('{}','{}','{}','{}','{}','{}','{}');".format(execid, trigger_name, current_project, last_build_url, last_timestamp, last_result, last_report_url)
db_insert(db, message_sql)
logging.info("------------------------ 数据库写入结束 ---------------------------")
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Jenkins Job Report</title>
<script src="https://cdn.bootcdn.net/ajax/libs/jquery/3.7.1/jquery.min.js"></script>
<link rel="stylesheet" href="https://cdn.bootcdn.net/ajax/libs/font-awesome/6.1.1/css/all.min.css">
<style>
body {
font-family: 'Arial', sans-serif;
color: #333;
background-color: #f4f4f4;
margin: 0;
padding: 20px;
line-height: 1.6;
}
.container {
position: relative; /* 设置为相对定位，作为 LOGO 的定位上下文 */
max-width: 1200px;
margin: auto;
padding: 0 20px;
}
h1, h2 {
text-align: center;
margin-bottom: 20px;
}
form {
margin-bottom: 20px;
text-align: center;
}
input[type="date"], button {
padding: 8px 15px;
margin: 5px;
border: 1px solid #ddd;
border-radius: 5px;
cursor: pointer;
}
button {
background-color: #007bff;
color: white;
border: none;
border-radius: 5px;
padding: 10px 20px;
cursor: pointer;
}
button:hover {
background-color: #0056b3;
}
table {
width: 100%;
border-collapse: collapse;
margin-bottom: 20px;
}
th, td {
padding: 10px;
text-align: left;
border-bottom: 1px solid #ddd;
}
th {
background-color: #333;
color: #fff;
}
tr:hover {
background-color: #f5f5f5;
}
.success { background-color: #d4edda; } /* 成功 */
.failure { background-color: #f8d7da; } /* 失败 */
.aborted { background-color: #fff3cd; } /* 中止 */
.unexecuted { background-color: #e9ecef; } /* 未执行 */
.loading {
color: red;
text-align: center;
margin-top: 20px;
display: none;
}
/* LOGO 样式 */
.logo {
position: absolute; /* 使用绝对定位 */
float: left; /* 让 LOGO 浮动到容器的左侧 */
width: 100px; /* 控制 LOGO 的宽度 */
height: auto; /* 高度自动，保持宽高比 */
margin-right: 20px; /* 与页面其他内容的间距 */
}
/* 清除浮动 */
.clear {
clear: both;
}
</style>
</head>
<body>
<div class="container">
<div class="container">
<!-- LOGO 图像 -->
<img src="XXXXX_logo_blue_tcm156-121992.svg" alt="Company Logo" class="logo">
<!-- 报告标题 -->
<h1>Jenkins Job Report</h1>
<div class="clear"></div> <!-- 清除浮动，确保后续内容正常显示 -->
<form id="reportForm">
<label for="startDate">Start Date:</label>
<input type="date" id="startDate" name="startDate" required>
<label for="endDate">End Date:</label>
<input type="date" id="endDate" name="endDate" required>
<button type="button" id="generateReport">Generate Report</button>
<p class="loading">Loading, please wait...</p>
</form>
<h2>Job Status List</h2>
<table id="jobReport">
<thead>
<tr>
<th>No.</th> <!-- 新增的序号列 -->
<th>Current Project</th>
<th>Triggered By</th>
<th>Last Time</th>
<th>Last Result</th>
<th>Execution Status</th>
<th>Link</th>
<th>Report</th>
</tr>
</thead>
<tbody>
<!-- Data will be inserted here -->
</tbody>
</table>
</div>
</div>
<script>
$(document).ready(function() {
$("#generateReport").click(function(event) {
event.preventDefault();
var startDate = $("#startDate").val();
var endDate = $("#endDate").val();
if (!startDate || !endDate) {
alert("Please provide both start and end dates");
return;
}
// 显示加载提示信息
$(".loading").show();
// 发起 AJAX 请求获取数据
$.ajax({
url: 'http://xxxxxxx:5000/fetch-data',
type: 'GET',
data: { start: startDate, end: endDate},
dataType: 'json',
success: function(data) {
console.log('Data received:', data); // 打印返回的数据
// 隐藏加载提示信息
$(".loading").hide();
// 渲染表格数据
var tableBody = $("#jobReport tbody");
tableBody.empty(); // 清空表格
var index = 1; // 初始化序号
data.forEach(function(job) {
// 根据作业的执行结果添加相应的CSS类
var jobRowClass = getJobStatusClass(job.last_result);
// 创建图标链接
var linkIcon = $("<a>")
.addClass("link-icon")
.attr("href", job.last_build_url)
.attr("target", "_blank")
.append($("<i>").addClass("fas fa-external-link-alt"));
// 创建报告链接图标
var reportLinkIcon = $("<a>")
.addClass("report-link-icon") // 可以为这个链接添加一个特定的类，以便在CSS中设置样式
.attr("href", job.last_report_url) // 确保这个属性名是正确的
.attr("target", "_blank") // 在新标签页中打开链接
.append($("<i>").addClass("fas fa-file-code")); // 使用Font Awesome的代码文件图标
// 创建表格行并添加数据
var jobRow = $("<tr>")
.addClass(jobRowClass) // 添加CSS类
.append($("<td>").text(index)) // 添加序号
.append($("<td>").text(job.current_project))
.append($("<td>").text(job.trigger_name))
.append($("<td>").text(job.last_timestamp))
.append($("<td>").text(job.last_result))
.append($("<td>").text(job.execution_status))
.append($("<td>").append(linkIcon)) // 新增图标链接列
.append($("<td>").append(reportLinkIcon)); // 报告链接图标
// 如果作业未执行，添加'unexecuted'类
if (job.execution_status === '未执行') {
jobRow.addClass('unexecuted');
}
tableBody.append(jobRow);
index++; // 增加序号
});
},
error: function(xhr, status, error) {
// 隐藏加载提示信息
$(".loading").hide();
alert("Failed to load data: " + error);
}
});
});
// 根据项目结果返回相应的CSS类
function getJobStatusClass(result) {
switch(result) {
case 'SUCCESS':
return 'success';
case 'FAILURE':
return 'failure';
case 'ABORTED':
return 'aborted';
default:
return 'unexecuted'; // 默认为未执行
}
}
});
</script>
</body>
</html>
# !/usr/bin/python
# -*- coding: utf-8 -*-
"""
@File : job_flask_server.py
@Create Time: 2024-07-25 14:04
@Description: 创建Flask服务，进行整体数据查询展示
"""
import datetime
from flask import *
from job_monitor_db import connect_db, query_database
from flask_cors import CORS
app = Flask(__name__)
# 允许所有域名访问您的应用
CORS(app)
@app.route('/tasks', methods=['GET'])
def manage_tasks_batch():
# 连接数据库
db = connect_db()
try:
message_sql = '''
SELECT
trigger_name,
current_project,
last_build_url,
last_timestamp,
last_result
FROM Jenkins_Jobs
WHERE execid IN (
SELECT MAX(execid) FROM Jenkins_Jobs
);
'''
# 执行查询
data = query_database(db, message_sql)
# 将查询结果转换为字典列表
results = []
for row in data:
# 由于 row 是元组，我们使用列名作为索引来创建字典
result = {
'trigger_name': row[0],
'current_project': row[1],
'last_build_url': row[2],
# 假设 last_timestamp 是 datetime 对象，我们将其转换为字符串
'last_timestamp': row[3].strftime('%Y-%m-%d %H:%M:%S') if hasattr(row[3], 'strftime') else row[3],
'last_result': row[4]
}
results.append(result)
# 返回 JSON 响应
return jsonify(results)
except Exception as e:
print(e)
db.close()
return jsonify({'error': 'An error occurred'}), 500
# 新的路由处理 GET 请求，根据时间范围过滤任务数据
@app.route('/fetch-data', methods=['GET'])
def fetch_data():
start_str = request.args.get('start')
end_str = request.args.get('end')
# 检查日期参数是否存在
if not start_str or not end_str:
return jsonify({"error": "Please provide both 'start' and 'end' date parameters."}), 400
try:
# 将字符串转换为日期对象
start_date = datetime.datetime.strptime(start_str, "%Y-%m-%d").date()
end_date = datetime.datetime.strptime(end_str, "%Y-%m-%d").date()
# 构造包括结束日期整天的 SQL 查询
# 假设 start_date 和 end_date 已经被定义为 datetime.date 对象
next_day = end_date + datetime.timedelta(days=1)
start_date_str = start_date.strftime("%Y-%m-%d 00:00:00")
end_date_str = next_day.strftime("%Y-%m-%d 00:00:00")
# 执行查询
# 这个 SQL 查询旨在从 `Jenkins_Jobs` 表中检索每个 `trigger_name` 的最新 `execid` 对应的记录，
# 并根据记录的 `last_timestamp` 是否在指定的日期范围内来标记每条记录的执行状态（'已执行' 或 '未执行'）。
# 它首先通过一个子查询为每个 `trigger_name` 找出最大的 `execid`，
# 然后使用 `INNER JOIN` 将这个结果与原表连接，确保只选择每个触发器最新执行的记录。
# 最后，查询结果按照执行状态和时间戳降序排序，以便优先显示最新且已执行的记录。
message_sql = f"""
SELECT
j1.trigger_name AS trigger_name,
j1.current_project AS current_project,
j1.last_build_url AS last_build_url,
j1.last_timestamp AS last_timestamp,
j1.last_result AS last_result,
j1.last_report_url AS last_report_url,
CASE
WHEN j1.last_timestamp >= '{start_date_str}' AND j1.last_timestamp < '{end_date_str}' THEN '已执行'
ELSE '未执行'
END AS execution_status
FROM
Jenkins_Jobs j1
INNER JOIN (
SELECT
trigger_name,
MAX(execid) AS max_execid
FROM
Jenkins_Jobs
GROUP BY
trigger_name
) j2 ON j1.trigger_name = j2.trigger_name AND j1.execid = j2.max_execid
ORDER BY
execution_status, -- 先按执行状态排序
j1.last_timestamp DESC; -- 再按时间戳降序排序，确保最新的记录排在前面
"""
db = connect_db()
data = query_database(db, message_sql)
print(data)
# 将查询结果转换为字典列表
results = [
{
'trigger_name': row[0],
'current_project': row[1],
'last_build_url': row[2],
'last_timestamp': row[3].strftime('%Y-%m-%d %H:%M:%S'), # 格式化时间戳
'last_result': row[4],
'last_report_url': row[5], # 添加 last_report_url 到字典中
'execution_status': row[6] # 添加 execution_status 到字典中
}
for row in data
]
return jsonify(results)
except ValueError:
# 如果日期格式不正确，返回错误信息
return jsonify({"error": "Invalid date format. Please use YYYY-MM-DD."}), 400
except Exception as e:
# 其他错误
return jsonify({"error": str(e)}), 500
@app.route('/report')
def display():
"""
整体页面展示
:return:
"""
return render_template("/jenkins_job_report_new.html")
if __name__ == '__main__':
app.run(host='0.0.0.0', port='5000', debug=True)
