需求背景

领导想知道目前 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()))

获取 Jenkins 上的 JOB 任务

# !/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("------------------------ 数据库写入结束 ---------------------------")

HTML 页面代码

<!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>

搭建 flask 服务

# !/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)

最终数据展示