Автор оригинала: Mike Driscoll.
Мы обсуждали, как использовать OpenVPN с Python в последних двух статьях. В этом заключительном посте я покажу, как принести все это вместе в графический интерфейс с кодом WXPYPHON. Я также собираюсь обсудить некоторые важные фрагменты.
Первый фрагмент, чтобы принять к сведению в Беги метод. Нам нужно убедиться, что служба OpenVPN работает при запуске нашей программы или файл журнала не будет обновлен. Итак, вы заметите, что используется метод StartService Win32ServiceUtil. Мы положили это в попробуйте Заявление, чтобы поймать ошибку, которая произойдет, если служба OpenVPN уже работает, не найден или не может быть запущен. Как правило, вы не должны использовать голые, кроме как можно маскировать другие ошибки в вашей программе; Тем не менее, я не смог найти соответствующий код ошибки для использования, поэтому я оставлю это для читателя.
def run(self):
"""
Run the openvpn.exe script
"""
vpnname='MCISVPN'
configfile='mcisvpn.conf'
defaultgw=''
vpnserver=''
vpnserverip = ''
print 'Starting OpenVPN Service...',
try:
win32serviceutil.StartService('OpenVPN Service', None)
except Exception, e:
print e
print 'success!'
delayedresult.startWorker(self._resultConsumer, self._resultProducer,
wargs=(self.jobID,self.abortEvent), jobID=self.jobID)
После попытки запустить службу OpenVPN, я использую модель резьбы, предоставленную wxpython, чтобы запустить код Watcher.py, который я написал о последнем времени, а также отслеживать мое местоположение в файле журнала.
Ниже приведен основной код GUI в полном объеме:
from vpnTBIcon import VPNIconCtrl
import os
import sys
import Queue
import threading
import time
import win32file
import win32con
import win32serviceutil
import wx
import wx.lib.delayedresult as delayedresult
ACTIONS = {
1 : "Created",
2 : "Deleted",
3 : "Updated",
4 : "Renamed to something",
5 : "Renamed from something"
}
def watch_path (path_to_watch, include_subdirectories=False):
FILE_LIST_DIRECTORY = 0x0001
hDir = win32file.CreateFile (
path_to_watch,
FILE_LIST_DIRECTORY,
win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE,
None,
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_BACKUP_SEMANTICS,
None
)
while 1:
results = win32file.ReadDirectoryChangesW (
hDir,
1024,
include_subdirectories,
win32con.FILE_NOTIFY_CHANGE_FILE_NAME |
win32con.FILE_NOTIFY_CHANGE_DIR_NAME |
win32con.FILE_NOTIFY_CHANGE_ATTRIBUTES |
win32con.FILE_NOTIFY_CHANGE_SIZE |
win32con.FILE_NOTIFY_CHANGE_LAST_WRITE |
win32con.FILE_NOTIFY_CHANGE_SECURITY,
None,
None
)
for action, file in results:
full_filename = os.path.join (path_to_watch, file)
if not os.path.exists (full_filename):
file_type = ""
elif os.path.isdir (full_filename):
file_type = 'folder'
else:
file_type = 'file'
yield (file_type, full_filename, ACTIONS.get (action, "Unknown"))
class Watcher (threading.Thread):
def __init__ (self, path_to_watch, results_queue, **kwds):
threading.Thread.__init__ (self, **kwds)
self.setDaemon (1)
self.path_to_watch = path_to_watch
self.results_queue = results_queue
self.start ()
def run (self):
for result in watch_path (self.path_to_watch):
self.results_queue.put (result)
# -------------------------------------------------------------------------
# GUI Code starts here
class vpnGUI(wx.App):
"""
wx application that wraps jrb's vpn script to allow it
to run in the system tray
"""
def __init__(self, redirect=False, filename=None):
wx.App.__init__(self, redirect, filename)
self.frame = wx.Frame(None, wx.ID_ANY, title='Voicemail', size=(800,500) )
self.panel = wx.Panel(self.frame, wx.ID_ANY)
self.abortEvent = delayedresult.AbortEvent()
# Set defaults
# ----------------------------------------------------------------------------------------
self.jobID = 0
# Create widget controls
# ----------------------------------------------------------------------------------------
# redirect stdout
self.log = wx.TextCtrl(self.panel, wx.ID_ANY, size=(1000,500),
style = wx.TE_MULTILINE|wx.TE_READONLY|wx.HSCROLL)
redir=RedirectText(self.log)
sys.stdout=redir
closeBtn = wx.Button(self.panel, wx.ID_ANY, 'Close')
mainSizer = wx.BoxSizer(wx.VERTICAL)
mainSizer.Add(self.log, 1, wx.ALL|wx.EXPAND, 5)
mainSizer.Add(closeBtn, 0, wx.ALL|wx.CENTER, 5)
self.panel.SetSizer(mainSizer)
# Bind events
# ----------------------------------------------------------------------------------------
self.Bind(wx.EVT_BUTTON, self.onClose, closeBtn)
self.Bind(wx.EVT_ICONIZE, self.onMinimize)
# create the system tray icon:
try:
self.tbicon = VPNIconCtrl(self.frame)
except Exception, e:
print 'Icon creation exception => %s' % e
self.tbicon = None
# comment this line out if you don't want to show the
# GUI when the program is run
self.frame.Show(True) # make the frame visible
self.run()
def run(self):
"""
Run the openvpn service
"""
vpnname='MCISVPN'
configfile='mcisvpn.conf'
defaultgw=''
vpnserver=''
vpnserverip = ''
print 'Starting OpenVPN Service...',
try:
win32serviceutil.StartService('OpenVPN Service', None)
except Exception, e:
print e
print 'success!'
delayedresult.startWorker(self._resultConsumer, self._resultProducer,
wargs=(self.jobID,self.abortEvent), jobID=self.jobID)
def _resultProducer(self, jobID, abortEvent):
"""
GUI will freeze if this method is not called in separate thread.
"""
PATH_TO_WATCH = [r'C:\Program Files\OpenVPN\log']
try: path_to_watch = sys.argv[1].split (",") or PATH_TO_WATCH
except: path_to_watch = PATH_TO_WATCH
path_to_watch = [os.path.abspath (p) for p in path_to_watch]
print "Watching %s at %s" % (", ".join (path_to_watch), time.asctime ())
files_changed = Queue.Queue ()
for p in path_to_watch:
Watcher (p, files_changed)
filepath = os.path.join(PATH_TO_WATCH[0], 'mcisvpn.log')
print 'filepath => ' + filepath
f = open(filepath)
for line in f.readlines():
print line
last_pos = f.tell()
f.close()
while not abortEvent():
try:
file_type, filename, action = files_changed.get_nowait ()
if action == 'Updated':
print 'Last pos => ', last_pos
f = open(filepath)
f.seek(last_pos)
for line in f.readlines():
if line != '\n':
print line
last_pos = f.tell()
f.close()
except Queue.Empty:
pass
time.sleep (1)
return jobID
def _resultConsumer(self, delayedResult):
jobID = delayedResult.getJobID()
assert jobID == self.jobID
try:
result = delayedResult.get()
except Exception, exc:
print "Result for job %s raised exception: %s" % (jobID, exc)
return
def onMinimize(self, event):
""" Minimize to tray """
self.frame.Hide()
def onClose(self, event):
"""
Close the program
"""
# recover stdout
sys.stdout=sys.__stdout__
# stop OpenVPN service
try:
print 'Stopping OpenVPN service...'
win32serviceutil.StopService('OpenVPN Service', None)
except Exception, e:
print e
# stop the threads
self.abortEvent.set()
# remove the icon from the tray
self.tbicon.Destroy()
# close the frame
self.frame.Close()
class RedirectText:
def __init__(self,textDisplay):
self.out=textDisplay
def write(self,string):
self.out.WriteText(string)
###### Run script! ######
if __name__ == "__main__":
app = vpnGUI()
app.MainLoop()
Вы заметите, что перенаправьте STDOUT в __init__ на нашем виджете управления текстовым управлением. Чтобы написать на виджета, мы используем печатное настроек Python. Мы сбрасываем STDOUT в OnClose обработчик события. Этот обработчик также останавливает службу OpenVPN, уничтожает значок системного лотка и закрывает программу.
И это все там действительно есть. Ниже приведены некоторые ссылки для тех из вас, кто хочет выкопать в этих инструментах.
Ресурсы
Источник для этих примеров
- vpn.zip
- vpn.tar.tar.