baseimage-tutorial-framework/lib/tfw/components/source_code_event_handler.py

128 lines
4.5 KiB
Python
Raw Normal View History

from os.path import isfile, join, relpath, exists, isdir
from glob import glob
from tfw.event_handler_base import TriggerlessEventHandler
from tfw.components.directory_monitor import DirectoryMonitor
from tfw.config.logs import logging
log = logging.getLogger(__name__)
class FileManager:
def __init__(self, working_directory, selected_file=None, exclude=None):
self.exclude = exclude if exclude is not None else []
self._workdir = None
self.workdir = working_directory
2018-02-15 15:32:41 +00:00
self.filename = selected_file or self.files[0]
@property
def workdir(self):
return self._workdir
@workdir.setter
def workdir(self, directory):
if not exists(directory) or not isdir(directory):
raise EnvironmentError('"{}" is not a directory!'.format(directory))
self._workdir = directory
def select_file(self, filename):
if not filename in self.files:
raise EnvironmentError('No such file in workdir!')
self.filename = filename
@property
def files(self):
2018-02-08 15:56:30 +00:00
return [self._relpath(file) for file in glob(join(self._workdir, '**/*'), recursive=True)
if isfile(file) and
not any(word in file for word in self.exclude)]
@property
def file_contents(self):
with open(self._filepath(self.filename), 'r', errors='surrogateescape') as ifile:
return ifile.read()
@file_contents.setter
def file_contents(self, value):
with open(self._filepath(self.filename), 'w', errors='surrogateescape') as ofile:
ofile.write(value)
def _filepath(self, filename):
return join(self._workdir, filename)
def _relpath(self, filename):
return relpath(self._filepath(filename), start=self._workdir)
class SourceCodeEventHandler(TriggerlessEventHandler):
def __init__(self, key, directory, selected_file=None, exclude=None):
super().__init__(key)
self.filemanager = FileManager(directory, selected_file=selected_file, exclude=exclude)
self.commands = {'read': self.read,
'write': self.write,
'select': self.select,
'selectdir': self.select_dir}
self._monitor = None
self.reload_monitor()
@property
def monitor(self):
return self._monitor
def reload_monitor(self):
if self._monitor:
try: self._monitor.stop()
except KeyError: logging.debug('Working directory was removed ignoring...')
self._monitor = DirectoryMonitor(self.filemanager.workdir)
self._monitor.watch() # This runs on a separate thread
def read(self, data):
try: data['content'] = self.filemanager.file_contents
except PermissionError: data['content'] = 'You have no permission to open that file :('
except FileNotFoundError: data['content'] = 'This file was removed :('
except Exception: data['content'] = 'Failed to read file :('
return data
def write(self, data):
self.monitor.eventhandler.ignore = 1
try: self.filemanager.file_contents = data['content']
except Exception: log.exception('Error writing file!')
del data['content']
return data
def select(self, data):
try: self.filemanager.select_file(data['filename'])
except EnvironmentError: log.exception('Failed to select file "{}"'.format(data['filename']))
return data
def select_dir(self, data):
try:
self.filemanager.workdir = data['directory']
self.reload_monitor()
try:
self.filemanager.select_file(self.filemanager.files[0])
self.read(data)
except IndexError:
data['content'] = 'No files in this directory :('
except EnvironmentError:
log.exception('Failed to select directory "{}"'.format(data['directory']))
return data
def attach_fileinfo(self, data):
data['filename'] = self.filemanager.filename
data['files'] = self.filemanager.files
data['directory'] = self.filemanager.workdir
def handle_event(self, key, message):
try:
data = message['data']
message['data'] = self.commands[data['command']](data)
self.attach_fileinfo(data)
return message
except KeyError:
log.error('IGNORING MESSAGE: Invalid message received: {}'.format(message))
2018-02-13 14:38:46 +00:00
def cleanup(self):
self.monitor.stop()