difflib paramiko模組
阿新 • • 發佈：2018-12-09
difflib模組
import difflib
# 1. 以字串方式展示兩個文字的不同， 效果如下:
text1 = '''
1. Beautiful is better than ugly.
2. Explicit is better than implicit.
3. Simple is better than complex.
4. Complex is better than complicated.
'''.splitlines(keepends=True)
text2 = '''
1. Beautifu is better than ugly.
2. Explicit is better than implicit.
3. Simple is better than complex.
4. Complex is better than complicated.
''' 
import difflib
file1 = '/etc/passwd' paramiko模組
paramiko遠端密碼連線
基於ssh用於連線遠端伺服器做操作:遠端執行命令， 上傳檔案， 下載檔案
import paramiko
# 建立一個ssh物件; 
基於公鑰的遠端連線
import paramiko
from paramiko.ssh_exception import NoValidConnectionsError,AuthenticationException
def connect(cmd,hostname,port=22,user='root'):
# 建立一個ssh物件;
client = paramiko.SSHClient()
# 返回一個私鑰物件
private_key = paramiko.RSAKey.from_private_key_file('id_rsa')
# 2. 解決問題:如果之前沒有；連線過的ip, 會出現
# Are you sure you want to continue connecting (yes/no)? yes
# 自動選擇yesclient.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
# 3. 連線伺服器
client.connect(hostname=hostname,
port=port,
username=user,
pkey=private_key)
# 4. 執行操作
stdin, stdout, stderr = client.exec_command(cmd)
except NoValidConnectionsError as e:
print("連線失敗")
except AuthenticationException as e:
print("密碼錯誤")
else:
# 5. 獲取命令的執行結果;
result = stdout.read().decode('utf-8')
print(result)
finally:
# 6. 關閉連線
client.close()
if __name__=='__main__':
#批量連線，操作
# for count in range(254):
# host = '172.25.254.%s' % (count + 1)
# print(host.center(50, '*'))
# connect('hostname', host)
connect('hostname','172.25.254.246')
基於使用者名稱密碼的上傳和下載
import paramiko
transport = paramiko.Transport(('172.25.254.246',22))
transport.connect(
username='root',
password='westos'
)
sftp = paramiko.SFTPClient.from_transport(transport)
# 上傳檔案, 包含檔名
sftp.put('/home/kiosk/Desktop/rhcsa-exam','/mnt/rhcsa-exam')
sftp.get('/mnt/rhcsa-exam','/home/kiosk/Desktop/rhcsa1-exam')
基於金鑰上傳下載
import paramiko
# 返回一個私鑰物件
private_key = paramiko.RSAKey.from_private_key_file('id_rsa')
transport = paramiko.Transport(('172.25.254.246', 22))
transport.connect(username='root',pkey=private_key)
sftp = paramiko.SFTPClient.from_transport(transport)
# 上傳檔案, 包含檔名
sftp.put('/home/kiosk/Desktop/rhcsa-exam', '/mnt/rhcsa2-exam')
sftp.get('/mnt/rhcsa2-exam', '/home/kiosk/Desktop/rhcsa2-exam')
transport.close()
paramiko的封裝
import os
import paramiko
from paramiko.ssh_exception import NoValidConnectionsError, AuthenticationException, SSHException
class SshRemoteHost(object):
def __init__(self, hostname, port, user, passwd, cmd):
# 指的不是shell命令
# cmd shell命令
# put
# get
self.hostname = hostname
self.port = port
self.user = user
self.passwd = passwd
self.cmd = cmd
def run(self):
"""預設呼叫的內容"""
# cmd hostname
# put
# get
cmd_str = self.cmd.split()[0] # cmd
# 類的反射, 判斷類裡面是否可以支援該操作?
if hasattr(self, 'do_'+ cmd_str): # do_cmd
getattr(self, 'do_'+cmd_str)()
else:
print("目前不支援該功能")
def do_cmd(self):
# 建立一個ssh物件;
client = paramiko.SSHClient()
# 2. 解決問題:如果之前沒有；連線過的ip, 會出現
# Are you sure you want to continue connecting (yes/no)? yes
# 自動選擇yes
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
# 3. 連線伺服器
client.connect(hostname=self.hostname,
port=self.port,
username=self.user,
password=self.passwd)
print("正在連線主機%s......." % (self.hostname))
except NoValidConnectionsError as e:
print("連線失敗")
except AuthenticationException as e:
print("密碼錯誤")
else:
# 4. 執行操作
# cmd uname
# cmd ls /etc/
# *******注意：
cmd = ' '.join(self.cmd.split()[1:])
stdin, stdout, stderr = client.exec_command(cmd)
# 5. 獲取命令的執行結果;
result = stdout.read().decode('utf-8')
print(result)
# 6. 關閉連線
client.close()
def do_put(self):
# put /tmp/passwd /tmp/passwd
# put /tmp/passwd /tmp/pwd
# put /tmp/passwd # 將本機的/tmp/passwd檔案上傳到遠端主機的/tmp/passwd;
print("正在上傳.....")
try:
transport = paramiko.Transport((self.hostname, int(self.port)))
transport.connect(username=self.user, password=self.passwd)
except SSHException as e:
print("連線失敗")
else:
sftp = paramiko.SFTPClient.from_transport(transport)
newCmd = self.cmd.split()[1:]
if len(newCmd) == 2:
# 上傳檔案, 包含檔名
sftp.put(newCmd[0], newCmd[1])
print("%s檔案上傳到%s主機的%s檔案成功" %(newCmd[0],
self.hostname, newCmd[1]))
else:
print("上傳檔案資訊錯誤")
transport.close()
def do_get(self):
# 2. 根據選擇的主機組， 顯示包含的主機IP/主機名;
# 3. 讓使用者確認資訊， 選擇需要批量執行的命令；
# - cmd shell命令
# - put 本地檔案 遠端檔案
# - get 遠端檔案 本地檔案
print("正在下載.....")
try:
transport = paramiko.Transport((self.hostname, int(self.port)))
transport.connect(username=self.user, password=self.passwd)
except SSHException as e:
print("連線失敗")
else:
sftp = paramiko.SFTPClient.from_transport(transport)
newCmd = self.cmd.split()[1:]
if len(newCmd) == 2:
# 上傳檔案, 包含檔名
sftp.get(newCmd[0], newCmd[1])
print("下載檔案成功" )
else:
print("下載檔案資訊錯誤")
transport.close()
def main():
# 1. 選擇操作的主機組：eg: mysql, web, ftp
groups = [file.rstrip('.conf') for file in os.listdir('conf')]
print("主機組顯示:".center(50, '*'))
for group in groups: print('\t', group)
choiceGroup = input("清選擇批量操作的主機組(eg:web):")
# 2. 根據選擇的主機組， 顯示包含的主機IP/主機名;
# 1). 開啟檔案conf/choiceGroup.conf
# 2). 依次讀取檔案每一行，
# 3). 只拿出ip
print("主機組包含主機:".center(50, '*'))
with open('conf/%s.conf' %(choiceGroup)) as f:
for line in f:
print(line.split(':')[0])
f.seek(0,0) # 把指標移動到檔案最開始
hostinfos = [line.strip() for line in f.readlines()]
# 3. 讓使用者確認資訊， 選擇需要批量執行的命令；
print("批量執行指令碼".center(50, '*'))
while True:
cmd = input(">>:").strip() # cmd uname
if cmd:
if cmd == 'exit' or cmd =='quit':
print("執行結束， 退出中......")
break
# 依次讓該主機組的所有主機執行
for info in hostinfos:
# 'ip:port:user:passwd'
host, port, user, passwd = info.split(":")
print(host.center(50, '-'))
clientObj = SshRemoteHost(host, port, user, passwd, cmd)
clientObj.run()
if __name__ == '__main__':
main()
