model_script/client/XxlJobClient.py

import json

import requests

header = {
    "Cookie": "Hm_lvt_7f5c31e86e4cc8a706f7b3957d81f353=1714270679; "
              "XXL_JOB_LOGIN_IDENTITY=7b226964223a312c22757365726e616d65223a2261646d696e222c2270617373776f7264"
              "223a226531306164633339343962613539616262653536653035376632306638383365222c22726f6c65223a312c2270"
              "65726d697373696f6e223a6e756c6c7d",
    "content-type": "application/x-www-form-urlencoded; charset=UTF-8"
}


class XxJobClient(object):

    def __init__(self, base_url):
        self.base_url = base_url

    def get_all_job_log(self, job_group=0, job_id=0, log_status=-1) -> list[dict]:
        """
        获取job的执行日志
        :param job_group: 执行器ID, 0-全部
        :param job_id: 任务ID, 0-全部
        :param log_status:  任务状态，-1-全部, 1-成功, 2-失败, 3-进行中
        :return:
        """
        url = f"{self.base_url}/joblog/pageList"
        param = {
            "start": 0,
            "length": 1000,
            "jobGroup": job_group,
            "jobId": job_id,
            "logStatus": log_status,
            "filterTime": "2024-09-27 00:00:00 - 2024-09-27 23:59:59"
        }
        response = requests.post(url, data=param, headers=header)
        return json.loads(response.content)['data']

    def get_all_job_group(self) -> list[dict]:
        """
        获取所有的执行器
        :return:
        """

        url = f"{self.base_url}/jobgroup/pageList"
        param = {
            "start": 0,
            "length": 1000,
        }
        response = requests.post(url, data=param, headers=header)
        return json.loads(response.content)['data']

    def get_all_job_info(self, job_group=0, trigger_status=-1) -> list[dict]:
        """
        获取所有的job信息
        :param job_group: 执行器ID
        :param trigger_status: 状态 0-关，1-开，-1-全部
        :return:
        """
        url = f"{self.base_url}/jobinfo/pageList"
        param = {
            "jobGroup": job_group,
            "start": 0,
            "length": 1000,
            "triggerStatus": trigger_status
        }
        response = requests.post(url, data=param, headers=header)
        return json.loads(response.content)['data']