您的位置:首页 > 运维架构 > Linux

python利用wxpython实现ssh连接linux进展

2014-08-01 09:56 686 查看
加了宽口图标,sqllite支持,历史命令,及上下键显示历史命令

#coding:utf-8
import wx
import paramiko
import  time
import sqlite3
class SSHMainWindow(wx.App):

    def OnInit(self):

        #初始一个frame
        frame = wx.Frame(parent=None, title='SHHWindow', size=(1000,600))

        self.panel = wx.Panel(frame, -1)
        self.text = wx.TextCtrl(self.panel, -1,"", style=wx.TE_RICH|wx.TE_MULTILINE)
        self.icon = wx.Icon('./images/screen.ico', wx.BITMAP_TYPE_ICO)
        frame.SetIcon(self.icon)
        #最小化窗体时的值
        self.text.SetMinSize((900,500))
        #菜单
        menuBar = wx.MenuBar()
        self.menu = wx.Menu()
        login = self.menu.Append(-1, 'Login')
        menuBar.Append(self.menu, '&File')
        frame.SetMenuBar(menuBar)
        #组件布局
        self.startBtn = wx.Button(self.panel, -1,u"执行命令")
        self.Bind(wx.EVT_BUTTON, self.OnStartButton, self.startBtn)
        self.showhisBtn = wx.Button(self.panel, -1,u"历史命令")
        self.Bind(wx.EVT_BUTTON, self.OnShowIsBtnButton, self.showhisBtn)
        self.username=wx.StaticText(self.panel,-1,' User Name ')
        self.usernametext = wx.TextCtrl(self.panel,-1,size=(100,-1))
        self.password=wx.StaticText(self.panel,-1,' Password ')
        self.passwordtext = wx.TextCtrl(self.panel,-1,size=(100,-1),style=wx.TE_PASSWORD)
        self.IPadress=wx.StaticText(self.panel,-1,' IP ')
        self.IP_adress=wx.TextCtrl(self.panel,-1,size=(100,-1))
        self.shell=wx.StaticText(self.panel,-1,' SHELL ')
        self.command = wx.TextCtrl(self.panel,-1,size=(200,-1),style=wx.TE_PROCESS_ENTER)
        #按回车键时
        self.Bind(wx.EVT_TEXT_ENTER ,self.OnStartButton,self.command)
        #按向下键
        self.Bind(wx.EVT_KEY_DOWN ,self.OnKeyDown,self.command)
        inner = wx.BoxSizer(wx.HORIZONTAL)

        inner.Add(self.username,0,wx.ALIGN_CENTER_VERTICAL)
        inner.Add(self.usernametext,0,wx.ALIGN_CENTER_VERTICAL)
        inner.Add(self.password,0,wx.ALIGN_CENTER_VERTICAL)
        inner.Add(self.passwordtext,0,wx.ALIGN_CENTER_VERTICAL)
        inner.Add(self.IPadress,0,wx.ALIGN_CENTER_VERTICAL)
        inner.Add(self.IP_adress, 0, wx.ALIGN_CENTER_VERTICAL)
        inner.Add(self.shell,0,wx.ALIGN_CENTER_VERTICAL)
        inner.Add(self.command, 0, wx.ALIGN_CENTER_VERTICAL)
        inner.Add(self.startBtn, 0, wx.ALIGN_CENTER_VERTICAL)
        inner.Add(self.showhisBtn, 0, wx.ALIGN_CENTER_VERTICAL)
        main = wx.BoxSizer(wx.VERTICAL)
        main.Add(inner, 0, wx.EXPAND|wx.ALL,5)
        main.Add(self.text, 1, wx.EXPAND|wx.ALL, 5)
        self.panel.SetSizer(main)
        main.Fit(frame)
        #居中显示窗体
        frame.Centre()
        frame.Show()
        return True
    #方向键执行历史命令
    def OnKeyDown(self,event):
        keycode = event.GetKeyCode()
        if keycode == wx.WXK_UP or keycode == wx.WXK_DOWN:
            try:
                conn = sqlite3.connect('./commandlog.db')
                print "Opened database successfully";
                cur = conn.cursor()
                global  ID
                if self.command.GetValue()=='':
                    cur.execute('SELECT max(ID) ID,command FROM COMMANDLOG ORDER BY ID')
                    for row in cur.fetchall():
                        ID=row[0]+1
                        self.command.WriteText(row[1])
                if ID !=None and self.command.GetValue()!='':
                    ID=ID-1
                    t=(ID,)
                    cur.execute('SELECT  ID,command FROM COMMANDLOG  where ID=? ORDER BY ID',t)

                    self.command.Clear()
                    for row in cur.fetchall():
                        ID=row[0]
                        print ID
                        self.command.WriteText(row[1])
            except Exception, e:
                print e
                self.command.WriteText(u"执行失败!")
                return  None
        else:
           event.Skip()
    #执行命令的函数
    def OnShowIsBtnButton(self,event):
        try:

             conn = sqlite3.connect('./commandlog.db')
             print "Opened database successfully";
             cur = conn.cursor()
             cur.execute('SELECT * FROM COMMANDLOG ORDER BY ID DESC')
             self.text.Clear()
             for row in cur.fetchall():

                  self.text.WriteText("%s \t %s \t %s \n" %('command:',row[1],row[2]))

        except Exception, e:
                print e
                self.text.WriteText(u"执行失败!")
                return  None

    def OnStartButton(self,event):
        try:

            ssh = paramiko.SSHClient()
            ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            ssh.connect(self.IP_adress.GetValue(), 22, self.usernametext.GetValue(), self.passwordtext.GetValue(), timeout=5)
            stdin, stdout, stderr = ssh.exec_command(self.command.GetValue())
            strout=(stdout.read()+stderr.read()).decode('utf-8', 'ignore')
            timestr=time.strftime('%Y-%m-%d %A %X ',time.localtime(time.time())).decode('utf-8', 'ignore')

            self.text.Clear()
            self.text.WriteText('\n')
            self.text.WriteText(strout)

            conn = sqlite3.connect('./commandlog.db')
            print "Opened database successfully";
            cur = conn.cursor()

            for t in[(self.command.GetValue() ,timestr)]:
                cur.execute('INSERT INTO COMMANDLOG (ID, COMMAND, TIME) VALUES ( NULL,?,?)',t)
            conn.commit()
            cur.execute('SELECT * FROM COMMANDLOG')

            print cur.fetchall()
            cur.execute("SELECT * FROM sqlite_master WHERE type = 'trigger';")
            print cur.fetchall()

            ssh.close()
        except Exception, e:
                print e
                self.text.WriteText(u"执行失败!")
                return  None
app = SSHMainWindow()
app.MainLoop()
内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签: