dingxin_toolbox_drf/update/utils/cmd_extcute.py

import paramiko
from update.models import UpdateCommand
from django.db.models import Q


class UpdateCommandUtil:
    def __init__(self):
        self.client = paramiko.SSHClient()
        self.model = UpdateCommand

    def connect(self, cinema_ip):
        cinema_config = {'hostname': cinema_ip, 'port': 22, 'username': 'root', 'password': 'cine123456'}
        self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        self.client.connect(**cinema_config)

    def disconnect(self):
        self.client.close()

    def get_all_cmd(self):
        return self.model.objects.filter(is_delete=False).order_by('id').all()

    def get_sys_cmd(self):
        return self.model.objects.filter(Q(is_delete=False) & Q(is_sys=True)).order_by('id').all()

    def get_no_sys_cmd(self):
        return self.model.objects.filter(Q(is_delete=False) & Q(is_sys=False)).order_by('id').all()

    def get_no_sys_setup_cmd(self):
        return self.model.objects.filter(Q(is_delete=False) & Q(is_sys=False) & Q(process='setup')).order_by('id').all()

    def get_no_sys_teardown_cmd(self):
        return self.model.objects.filter(Q(is_delete=False) & Q(is_sys=False) & Q(process='teardown')).order_by('id').all()

    def get_checked_cmd(self, cmd_list):
        sys_result = self.get_sys_cmd()
        checked_setup_result = self.model.objects.filter(
            Q(is_delete='0') & Q(id__in=cmd_list) & Q(process='setup')).order_by('id').all()
        checked_teardown_result = self.model.objects.filter(
            Q(is_delete='0') & Q(id__in=cmd_list) & Q(process='teardown')).order_by(
            id).all()
        sys_cmd = [sys['command'] for sys in sys_result]
        checked_setup_cmd = [setup['command'] for setup in checked_setup_result]
        checked_teardown_cmd = [teardown['command'] for teardown in checked_teardown_result]
        return {'sys': sys_cmd, 'setup': checked_setup_cmd, 'teardown': checked_teardown_cmd}

    def exec_cmd(self, _exec_cmd_list):
        self.connect()
        for exec_cmd in _exec_cmd_list:
            stdin, stdout, stderr = self.client.exec_command(exec_cmd)
            print(stdout.read().decode('utf-8'))
        self.disconnect()

    def exec_setup_cmd(self, cmd_list, _release):
        exec_cmd_list = []
        exec_cmd_data = self.get_checked_cmd(cmd_list)
        setup_cmd = exec_cmd_data['sys'] + exec_cmd_data['setup']
        for cmd in setup_cmd:
            if '<param>' in cmd:
                cmd = cmd.replace('<param>', _release)
                exec_cmd_list.append(cmd)
            elif cmd.startswith('<multi>'):
                cmd = cmd.replace('<multi>', _release)
                for c in cmd.split('###'):
                    exec_cmd_list.append(c)
            else:
                exec_cmd_list.append(cmd)
        self.exec_cmd(exec_cmd_list)

    def exec_teardown_cmd(self, cmd_list, _release):
        exec_cmd_list = []
        exec_cmd_data = self.get_checked_cmd(cmd_list)
        teardown_cmd = exec_cmd_data['teardown']
        for cmd in teardown_cmd:
            if '<param>' in cmd:
                cmd = cmd.replace('<param>', _release)
                exec_cmd_list.append(cmd)
            elif cmd.startswith('<multi>'):
                cmd = cmd.replace('<multi>', _release)
                for c in cmd.split('###'):
                    exec_cmd_list.append(c)
            else:
                exec_cmd_list.append(cmd)
        self.exec_cmd(exec_cmd_list)