自动执行Yearning工单

import requests
import json
import logging

LOG_FORMAT = "%(asctime)s - %(levelname)s - %(message)s"
DATE_FORMAT = "%Y-%d-%m %H:%M:%S"

logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT, datefmt=DATE_FORMAT)


class YearningOp:
    def __init__(self, username, password):
        self.username = username
        self.password = password
        self.headers = self.gen_headers()

    def gen_headers(self):
        token = self.login().get('token')
        headers = {
            'content-type': 'application/json; charset=utf-8',
            'Authorization': 'Bearer {}'.format(token)
        }
        return headers

    def login(self):
        headers = {'content-type': 'application/json; charset=utf-8'}
        req_data = json.dumps({
            'username': self.username,
            'password': self.password
        })
        url = 'http://yearning.xyb2b.com.cn/login'
        resp = requests.post(url, headers=headers, data=req_data)
        return resp.json()

    def work_list(self):
        url = 'http://yearning.xyb2b.com.cn/api/v2/audit'
        req_data = '''{"page":1,"find":{"picker":[],"valve":false,"text":""}}'''
        resp = requests.put(url, headers=self.headers, data=req_data)
        return resp.json()['data']

    # 未执行工单 id 列表
    def nonexecution_work_id_lst(self):
        return [obj['work_id'] for obj in self.work_list() if obj['status'] == 2]

    def test(self, work_id):
        url = 'http://yearning.xyb2b.com.cn/api/v2/audit/test'
        req_data = json.dumps({
            'work_id': work_id
        })
        resp = requests.post(url, headers=self.headers, data=req_data)

        is_ok = False
        for r in resp.json():
            error = r.get('error')
            is_ok = error == ''
            if not is_ok:
                logging.error('工单[{}]测试失败, err: {}'.format(work_id, error))
                break
        return is_ok

    def exec(self, work_id):
        url = 'http://yearning.xyb2b.com.cn/api/v2/audit/execute'
        req_data = json.dumps({
            'work_id': work_id
        })
        resp = requests.post(url, headers=self.headers, data=req_data)
        print(resp.text)

    def run(self):
        lst = self.nonexecution_work_id_lst()

        if len(lst) == 0:
            logging.info('暂无待执行的工单')
            exit(0)

        lst.sort()
        for work_id in lst:
            logging.info('测试工单[{}] - begin'.format(work_id))
            test_ok = self.test(work_id)
            logging.info('测试工单[{}] - end'.format(work_id))

            if test_ok:
                logging.info('执行工单[{}] - begin'.format(work_id))
                self.exec(work_id)
                logging.info('执行工单[{}] - end'.format(work_id))

if __name__ == '__main__':
    YearningOp('zze', 'xxx').run()