Pexpect 是一個用來啟動子程序并對其進(jìn)行自動控制的純 Python 模塊。 Pexpect 可以用來和像 ssh、ftp、passwd、telnet 等命令行程序進(jìn)行自動交互。本文主要是針對ssh遠(yuǎn)程登錄，然后執(zhí)行簡單的新建文件夾與拷貝任務(wù)Pexpect 的安裝：下載：https://pypi.python.org/pypi/pexpect/解壓后在目錄下運(yùn)行：python setup.py installPexpect 的簡單使用：from pexpect import *user = 'user'host = 'host'password = 'password'#實現(xiàn)遠(yuǎn)程登錄host機(jī)器并新建/home/download/wangling/test目錄command = 'sudo ssh -l '+user+' '+host+' sudo mkdir -p /home/download/wangling/test'child = spawn(command , timeout=10 ) child.sendline(password)#實現(xiàn)遠(yuǎn)程文件拷貝(將本機(jī)1.txt文件拷貝到host機(jī)器test2目錄下2.txt)command1 = 'sudo scp /home/download/wangling/test1/1.txt '+user+'@'+host+':/home/download/wangling/test2/2.txt'child = spawn(command1 , timeout=10 )child.sendline(password)#!/usr/bin/env python #-*-coding:utf-8-*- #pexpect庫向文件發(fā)送數(shù)據(jù) from pexpect.fdpexpect import fdspawnf=open('/home/acm506/桌面/python數(shù)據(jù)庫/with.py','ab+')child=fdspawn(f) child.sendline('age sister') f.seek(0) age=child.expect('age') #成功的標(biāo)志 if age==0:child.sendline('age success!') child.close()#!/usr/bin/env python #-*-coding:utf-8-*- #ftp 協(xié)議進(jìn)行文件管理 import os import sys import re import time import os.path import pexpect#用戶登入 def login_ftp():ftp=pexpect.spawn('ftp',cwd=cwd)if ftp.expect(prmpt)!=0:sys.exit()#連接ftps服務(wù)器ftp.sendline(''.join(('open ',ftps)))#提示用戶名不成功if ftp.expect('Name')!=0:sys.exit()#發(fā)送用戶名ftp.sendline(ftpuser)#提示用戶名密碼輸入if ftp.expect('Password:')!=0:sys.exit()ftp.sendline(ftppw)if ftp.expect('230')!=0 or ftp.expect(prmpt)!=0:sys.exit()return ftp#獲取服務(wù)器下文件下所有的文件 def get_server_files(ftp):ftp.sendline('ls')if ftp.expect('226')!=0:sys.exit()#文件列表filelsts=ftp.beforefilelsts=filelsts.split('\n')#匹配多個空格remtch=re.compile('\s+')filelsts=[remtch.subn('',item.strip('\r'))[0] for item in filelsts if 'group' in item]filedict=dict()for item in filelsts:datas=item.split('')filedict[datas[-1]]={'mon':mons.index(datas[-4])+1,'day':int(datas[-3]),'time':datas[-2]}return filedict#獲得本地文件信息 def get_local_files():localfiles=os.listdir(cwd)localfilesdict=dict()for file in localfiles:t=time.ctime(os.stat(os.path.join(cwd,file)).st_mtime)#創(chuàng)建時間，修改時間datas=t.split()localfilesdict[file]={'mon':mons.index(datas[-4])+1,'day':int(datas[-3]),'time':datas[-2][:5]}return localfilesdict#文件同步到服務(wù)器 def sync_files(ftp,localfilesdict,filedict):#需要同步的文件addfile=[]for file in localfilesdict.keys():if file not in filedict:addfile.append(file)if file in filedict:if localfiledict[file]['mon']>filedict[file]['mon'] or localfiledict[file]['day']>filedict[file]['day'] or localfilesdict[file]['time']>filedict[file]['time']:addfile.append(file)#刪除服務(wù)器上有但是本地沒有的文件信息delfile=set(filedict.keys())-set(localfilesdict.keys())#上傳文件if addfile:for f in addfile:ftp.sendline('put '+f)if ftp.expect(['226',pexpect.EOF])==0:print('Upload success:',f)else:sys.exit()#刪除文件if delfile:for f in delfile:ftp.sendline('delete '+f)if ftp.expect(['250',pexpect.EOF])==0:print('Del:',f)else:print('Permission denied:')sys.exit()#退出ftp服務(wù)器 def exit_ftp(ftp):if ftp:ftp.sendcontrol('d')print(ftp.read().decode())if __name__=='__main__':mons=('Jan','Feb','Mar','Apr','May','Jun','Jul','Aug','Sep','Oct','Nov','Dec')cwd='/home/acm506/桌面/python數(shù)據(jù)庫'#服務(wù)器提示符prmpt=['ftp>',pexpect.EOF]#本機(jī)ipftps='192.168.1.167'#用戶名匿名ftpuser='anoneymous'#ftpuser='hang'#ftppw='hang'ftppw='123'#登入ftp服務(wù)器ftp=login_ftp()#獲取文件列表filedict=get_server_files(ftp)localfilesdict=get_local_files()#文件同步sync_files(ftp,localfilesdict,filedict)#退出服務(wù)器exit_ftp(ftp)#!/usr/bin/env python #-*-coding:utf-8-*- #測試網(wǎng)路window下 from pexpect.popen_spawn import PopenSpawn import pexpect.popen_spawndef test_ip(ipaddr):child=PopenSpawn('cmd')child.sendline('ping %s' % ipaddr)child.sendline('exit')child.expect(pexpect.EOF)out=child.before.decode('gbk')per=out[:out.find('%')][-2:]per=[ch for ch in per if ch.isdigit()]per=int(''.join(per))if per>=100:print('網(wǎng)絡(luò)不通！',ipaddr)elif 80>=per>=30:print('網(wǎng)絡(luò)不穩(wěn)定！',ipaddr)else:print('網(wǎng)絡(luò)正常！',ipaddr)if __name__=='__main__':addrlst=['192.168.1.167','192.168.1.1','8.8.8.8']for ip in addrlst:test_ip(ip)#!/usr/bin/env python #-*-coding:utf-8-*- #寫入python腳本 import pexpect.replwrapchild=pexpect.replwrap.python()print(child.run_command('2*1'))child=pexpect.replwrap.bash() print(child.run_command('ls'))#!/usr/bin/env python #-*-coding:utf-8-*- #ssh命令連接遠(yuǎn)程 from pexpect.pxssh import pxssh import getpasshostname='192.168.1.167' user='root' pw=getpass.getpass() s=pxssh() s.login(hostname,user,pw) s.sendline('ls -l') s.prompt() print(s.before.decode()) #磁盤使用情況 s.sendline('df') s.prompt() print(s.before.decode()) s.sendline('poweroff')s.logout() #!/usr/bin/env python #-*-coding:utf-8-*- #逐個登錄制定的多臺遠(yuǎn)程主機(jī)，監(jiān)控遠(yuǎn)程主機(jī)并依據(jù)相關(guān)信息要求用戶處理 #登錄多臺指定的遠(yuǎn)程主機(jī) #獲取遠(yuǎn)程主機(jī)的系統(tǒng)狀態(tài) #對遠(yuǎn)程主機(jī)狀態(tài)進(jìn)行檢查 #遠(yuǎn)程主機(jī)狀態(tài)良好則輸出相關(guān)信息 #遠(yuǎn)程主機(jī)負(fù)載過重則顯示其狀態(tài)信息，并由用戶選擇是否進(jìn)入交互模式處理，同時記錄處理日志 #退出登錄 from pexpect.pxssh import pxssh import pexpect#登錄遠(yuǎn)程 def login_host(host):s=pxssh()#連接遠(yuǎn)程if s.login(host[0],host[1],host[2]):return s#獲取遠(yuǎn)程主機(jī)CPU數(shù)量 def get_cpus(sshc):#cpu信息sshc.sendline('cat /proc/cpuinfo')#匹配res=sshc.expect(['cpu cores.*\r\n'.pexpect.EOF])if res==0:#匹配結(jié)構(gòu)data=sshc.after.decode().split('\r\n')#cpu數(shù)量data=data[0]data=data[data.index(':')+1:]#數(shù)量cpucores=int(data)sshc.prompt()return cpucores #遠(yuǎn)程主機(jī)負(fù)載狀況 def get_cpu_load(sshc):sshc.sendline('uptime')if sshc.prompt():data=sshc.before.decode()data=data.strip('\r\n')data=data[data.rfind(':')+1:]data=data.split(',')return (float(data[0]),float(data[1]),float(data[2]))#獲取負(fù)載的信息 def get_cpu_stat(sshc):sshc.sendline('vmstat')sshc.prompt()print(sshc.before.decode())#登入用戶后，處理一定的信息 def user_deal(host,logfilename):s=login_host(host)if not s:print('Login Failure:',host[0])returntry:#cpu的數(shù)量cpucores=get_cpus(s)if not cpucores:print('Do not get cpucores:',host[0])return cpu_load=get_cpu_load(s)if cpu_load[2]>=cpucores or 1:get_cpu_stat(s)print("System is not healthy.Do you want to deal?(yes/no)")yn=input()if yn=='yes':with open(logfilename,'ab+') as f:s.logfile=fs.interact()s.prompt()s.logfile=Noneelse:print('System is healthy:',host[0])except:print('Failure:',host[0])finally:s.logout()if __name__=='__main__':hosts=[('192.168.1.22','root','123'),]logfilename='log.txt'for host in hosts:user_deal(host,logfilename)

總結(jié)

以上是生活随笔為你收集整理的Python的Pexpect库的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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