話不多說直接上代碼

封裝連接

@staticmethod

def connect(ip, server_user, server_port, server_path):

"""

連接服務(wù)器

:param :

:return:

"""

ssh = paramiko.SSHClient()

private_key = paramiko.RSAKey.from_private_key_file('{}.ssh/id_rsa'.format(server_path))

ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

try:

ssh.connect(ip, port=server_port, username=server_user, pkey=private_key, timeout=5, allow_agent=True)

return ssh

except Exception as e:

app.logger.info('connect_error info is {}'.format(e.__repr__()))

return {

'result': 1,

'ip': ip,

'msg': e.__repr__(),

}

在服務(wù)器執(zhí)行命令

def fabric_run_cmd(self, ip, server_user, server_port, server_path, cmd):

"""

批量在服務(wù)器執(zhí)行命令

:return:

"""

conn = self.connect(ip, server_user, server_port, server_path)

try:

if type(conn) == dict:

return conn

else:

stdin, stdout, stderr = conn.exec_command(cmd)

app.logger.info('fabric_run_cmd_stdout info is {}'.format(stdout.readlines()))

app.logger.info('fabric_run_cmd_stderr info is {}'.format(stderr.readlines()))

channel = stdout.channel

status = channel.recv_exit_status()

conn.close()

app.logger.info('fabric_run_cmd_status info is {}'.format(status))

if status == 0:

return {

'ip': ip,

'msg': '執(zhí)行成功',

'result': 0

}

else:

return {

'ip': ip,

'msg': stderr.readlines(),

'result': 1

}

except Exception as e:

app.logger.info('fabric_run_cmd_error info is {}'.format(e.__repr__()))

return {

'ip': ip,

'msg': e.__repr__(),

'result': 1

}

將文件發(fā)送到服務(wù)器

def fabric_put_file(self, ip, src_file, dst_file, server_user, server_port, server_path):

"""

發(fā)送文件到服務(wù)器

:return:

"""

conn = self.connect(ip, server_user, server_port, server_path)

if type(conn) == dict:

return conn

else:

try:

ftp = conn.open_sftp()

ftp.put(dst_file, src_file) # 發(fā)送文件之前先判斷有沒有目標(biāo)文件夾 如果沒有現(xiàn)在服務(wù)器上創(chuàng)建文件夾

ftp.close() # 在此判斷文件是否發(fā)送成功

return {

'ip': ip,

'msg': '上傳成功',

'result': 0

}

except Exception as e:

app.logger.info('fabric_put_file_error info is {}'.format(e.__repr__()))

return {

'ip': ip,

'msg': e.__repr__(),

'result': 1

}

能力有限,如有不妥請(qǐng)留言指正

以上就是python如何利用paramiko執(zhí)行服務(wù)器命令的詳細(xì)內(nèi)容，更多關(guān)于python 執(zhí)行服務(wù)器命令的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

總結(jié)

以上是生活随笔為你收集整理的cmd运行python服务器,python如何利用paramiko执行服务器命令的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。