Pexpect 是一個用來啟動子程序并對其進行自動控制的純 Python 模塊。 Pexpect 可以用來和像 ssh、ftp、passwd、telnet 等命令行程序進行自動交互。本文主要是針對ssh遠程登錄，然后執行簡單的新建文件夾與拷貝任務Pexpect 的安裝：下載：https://pypi.python.org/pypi/pexpect/解壓后在目錄下運行：python setup.py installPexpect 的簡單使用：from pexpect import *user = 'user'host = 'host'password = 'password'#實現遠程登錄host機器并新建/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)#實現遠程文件拷貝(將本機1.txt文件拷貝到host機器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庫向文件發送數據 from pexpect.fdpexpect import fdspawnf=open('/home/acm506/桌面/python數據庫/with.py','ab+')child=fdspawn(f) child.sendline('age sister') f.seek(0) age=child.expect('age') #成功的標志 if age==0:child.sendline('age success!') child.close()#!/usr/bin/env python #-*-coding:utf-8-*- #ftp 協議進行文件管理 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服務器ftp.sendline(''.join(('open ',ftps)))#提示用戶名不成功if ftp.expect('Name')!=0:sys.exit()#發送用戶名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#獲取服務器下文件下所有的文件 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)#創建時間，修改時間datas=t.split()localfilesdict[file]={'mon':mons.index(datas[-4])+1,'day':int(datas[-3]),'time':datas[-2][:5]}return localfilesdict#文件同步到服務器 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)#刪除服務器上有但是本地沒有的文件信息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服務器 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數據庫'#服務器提示符prmpt=['ftp>',pexpect.EOF]#本機ipftps='192.168.1.167'#用戶名匿名ftpuser='anoneymous'#ftpuser='hang'#ftppw='hang'ftppw='123'#登入ftp服務器ftp=login_ftp()#獲取文件列表filedict=get_server_files(ftp)localfilesdict=get_local_files()#文件同步sync_files(ftp,localfilesdict,filedict)#退出服務器exit_ftp(ftp)#!/usr/bin/env python #-*-coding:utf-8-*- #測試網路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('網絡不通！',ipaddr)elif 80>=per>=30:print('網絡不穩定！',ipaddr)else:print('網絡正常！',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命令連接遠程 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-*- #逐個登錄制定的多臺遠程主機，監控遠程主機并依據相關信息要求用戶處理 #登錄多臺指定的遠程主機 #獲取遠程主機的系統狀態 #對遠程主機狀態進行檢查 #遠程主機狀態良好則輸出相關信息 #遠程主機負載過重則顯示其狀態信息，并由用戶選擇是否進入交互模式處理，同時記錄處理日志 #退出登錄 from pexpect.pxssh import pxssh import pexpect#登錄遠程 def login_host(host):s=pxssh()#連接遠程if s.login(host[0],host[1],host[2]):return s#獲取遠程主機CPU數量 def get_cpus(sshc):#cpu信息sshc.sendline('cat /proc/cpuinfo')#匹配res=sshc.expect(['cpu cores.*\r\n'.pexpect.EOF])if res==0:#匹配結構data=sshc.after.decode().split('\r\n')#cpu數量data=data[0]data=data[data.index(':')+1:]#數量cpucores=int(data)sshc.prompt()return cpucores #遠程主機負載狀況 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]))#獲取負載的信息 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的數量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)

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。