diff options
Diffstat (limited to 'plugins')
| -rw-r--r-- | plugins/alias.py | 40 | ||||
| -rw-r--r-- | plugins/day_change.py | 31 | ||||
| -rw-r--r-- | plugins/exec.py | 43 | ||||
| -rw-r--r-- | plugins/link.py | 48 | ||||
| -rw-r--r-- | plugins/screen_detach.py | 46 | ||||
| -rw-r--r-- | plugins/test.py | 20 | ||||
| -rw-r--r-- | plugins/translate.py | 34 |
7 files changed, 262 insertions, 0 deletions
diff --git a/plugins/alias.py b/plugins/alias.py new file mode 100644 index 00000000..eb596332 --- /dev/null +++ b/plugins/alias.py @@ -0,0 +1,40 @@ +from plugin import BasePlugin + +class Plugin(BasePlugin): + def init(self): + self.add_command('alias', self.command_alias, '/alias <alias> <command> <args>\nAlias: create an alias command') + self.add_command('unalias', self.command_unalias, '/unalias <alias>\nUnalias: remove a previously created alias') + self.commands = {} + + def command_alias(self, line): + arg = line.split() + if len(arg) < 2: + self.core.information('Alias: Not enough parameters', 'Error') + return + alias = arg[0] + command = arg[1] + post_args = ' '.join(arg[2:]) + + if alias in self.core.commands or alias in self.commands: + self.core.information('Alias: command already exists', 'Error') + return + self.commands[alias] = lambda args: self.get_command(command)(post_args+args) + self.add_command(alias, self.commands[alias], 'This command is an alias for /%s %s' % (command, post_args)) + self.core.information('Alias /%s successfuly created' % alias, 'Info') + + def command_unalias(self, alias): + if alias in self.commands: + del self.commands[alias] + self.del_command(alias) + self.core.information('Alias /%s successfuly deleted' % alias, 'Info') + + def get_command(self, name): + """Returns the function associated with a command""" + def dummy(args): + """Dummy function called if the command doesn’t exist""" + pass + if name in self.core.commands: + return self.core.commands[name][0] + elif name in self.core.current_tab().commands: + return self.core.current_tab().commands[name][0] + return dummy diff --git a/plugins/day_change.py b/plugins/day_change.py new file mode 100644 index 00000000..cac69a75 --- /dev/null +++ b/plugins/day_change.py @@ -0,0 +1,31 @@ +from gettext import gettext as _ +from plugin import BasePlugin +import datetime +import tabs +import timed_events + +class Plugin(BasePlugin): + def init(self): + self.schedule_event() + + def cleanup(self): + self.core.remove_timed_event(self.next_event) + + def schedule_event(self): + day_change = datetime.datetime.combine(datetime.date.today(), datetime.time()) + day_change += datetime.timedelta(1) + self.next_event = timed_events.TimedEvent(day_change, self.day_change) + self.core.add_timed_event(self.next_event) + + def day_change(self): + msg = datetime.date.today().strftime(_("Day changed to %x")) + + for tab in self.core.tabs: + if (isinstance(tab, tabs.MucTab) or + isinstance(tab, tabs.PrivateTab) or + isinstance(tab, tabs.ConversationTab)): + room = tab.get_room() + room.add_message(msg) + + self.core.refresh_window() + self.schedule_event() diff --git a/plugins/exec.py b/plugins/exec.py new file mode 100644 index 00000000..f7f451df --- /dev/null +++ b/plugins/exec.py @@ -0,0 +1,43 @@ +# A plugin that can execute a command and send the result in the conversation + +from plugin import BasePlugin +import os +import common +import shlex +import subprocess + +class Plugin(BasePlugin): + def init(self): + self.add_command('exec', self.command_exec, "Usage: /exec [-o|-O] <command>\nExec: Execute a shell command and prints the result in the information buffer. The command should be ONE argument, that means it should be between \"\". The first argument (before the command) can be -o or -O. If -o is specified, it sends the result in the current conversation. If -O is specified, it sends the command and its result in the current conversation.\nExample: /exec -O \"uptime\" will send “uptime\n20:36:19 up 3:47, 4 users, load average: 0.09, 0.13, 0.09” in the current conversation.") + + def command_exec(self, args): + args = common.shell_split(args) + if len(args) == 1: + command = args[0] + arg = None + elif len(args) == 2: + command = args[1] + arg = args[0] + else: + self.core.command_help('exec') + return + try: + cut_command = shlex.split(command) + except Exception as e: + self.core.information('Failed to parse command: %s' % (e,), 'Error') + return + try: + process = subprocess.Popen(cut_command, stdout=subprocess.PIPE) + except OSError as e: + self.core.information('Failed to execute command: %s' % (e,), 'Error') + return + result = process.communicate()[0].decode('utf-8') + if arg and arg == '-o': + if not self.core.send_message('%s' % (result,)): + self.core.information('Cannot send result (%s), this is not a conversation tab' % result) + elif arg and arg == '-O': + if not self.core.send_message('%s:\n%s' % (command, result)): + self.core.information('Cannot send result (%s), this is not a conversation tab' % result) + else: + self.core.information('%s:\n%s' % (command, result), 'Info') + return diff --git a/plugins/link.py b/plugins/link.py new file mode 100644 index 00000000..0cd1a0f1 --- /dev/null +++ b/plugins/link.py @@ -0,0 +1,48 @@ +# A plugin that adds the /link command, letting you open links that are pasted +# in the conversation, without having to click them. + +import os +import re + +from plugin import BasePlugin, PluginConfig +from xhtml import clean_text +import common + +url_pattern = re.compile(r'\b(http[s]?://(?:\S+))\b', re.I|re.U) + +class Plugin(BasePlugin): + def init(self): + self.add_command('link', self.command_link, "Usage: /link\nLink: opens the last link from the conversation into a browser.") + + def find_link(self, nb): + messages = self.core.get_conversation_messages() + if not messages: + return None + for message in messages[::-1]: + match = url_pattern.search(clean_text(message.txt)) + if match: + self.core.information('[%s]' % (match.groups(),)) + for url in list(match.groups())[::-1]: + if nb == 1: + return url + else: + nb -= 1 + return None + + def command_link(self, args): + args = common.shell_split(args) + if len(args) == 1: + try: + nb = int(args[0]) + except: + return self.core.command_help('link') + else: + nb = 1 + link = self.find_link(nb) + if link: + self.core.exec_command('%s %s' % (self.config.get('browser', 'firefox'), link)) + else: + self.core.information('No URL found.', 'Warning') + + def cleanup(self): + del self.config diff --git a/plugins/screen_detach.py b/plugins/screen_detach.py new file mode 100644 index 00000000..6ee96896 --- /dev/null +++ b/plugins/screen_detach.py @@ -0,0 +1,46 @@ +from plugin import BasePlugin +import os +import stat +import pyinotify + +SCREEN_DIR = '/var/run/screen/S-%s' % (os.getlogin(),) + +class Plugin(BasePlugin): + def init(self): + self.timed_event = None + sock_path = None + self.thread = None + for f in os.listdir(SCREEN_DIR): + path = os.path.join(SCREEN_DIR, f) + if screen_attached(path): + sock_path = path + self.attached = True + break + + # Only actually do something if we found an attached screen (assuming only one) + if sock_path: + wm = pyinotify.WatchManager() + wm.add_watch(sock_path, pyinotify.EventsCodes.ALL_FLAGS['IN_ATTRIB']) + self.thread = pyinotify.ThreadedNotifier(wm, default_proc_fun=HandleScreen(plugin=self)) + self.thread.start() + + def cleanup(self): + if self.thread: + self.thread.stop() + + def update_screen_state(self, socket): + attached = screen_attached(socket) + if attached != self.attached: + self.attached = attached + status = 'available' if self.attached else 'away' + self.core.command_status(status) + +def screen_attached(socket): + return (os.stat(socket).st_mode & stat.S_IXUSR) != 0 + +class HandleScreen(pyinotify.ProcessEvent): + def my_init(self, **kwargs): + self.plugin = kwargs['plugin'] + + def process_IN_ATTRIB(self, event): + self.plugin.update_screen_state(event.path) diff --git a/plugins/test.py b/plugins/test.py new file mode 100644 index 00000000..0d5cdb0a --- /dev/null +++ b/plugins/test.py @@ -0,0 +1,20 @@ +from plugin import BasePlugin + +class Plugin(BasePlugin): + def init(self): + self.add_command('plugintest', self.command_plugintest, 'Test command') + self.add_event_handler('message', self.on_message) + self.core.information("Plugin loaded") + self.core.connect('enter', self.on_enter) + + def cleanup(self): + self.core.information("Plugin unloaded") + + def on_enter(self, line): + self.core.information('Text sent: {}'.format(line)) + + def on_message(self, message): + self.core.information("Test plugin received message: {}".format(message)) + + def command_plugintest(self, args): + self.core.information("Command! With args {}".format(args)) diff --git a/plugins/translate.py b/plugins/translate.py new file mode 100644 index 00000000..520d02b4 --- /dev/null +++ b/plugins/translate.py @@ -0,0 +1,34 @@ +from plugin import BasePlugin +import urllib.request +from urllib.parse import urlencode +import xhtml +import json + +TARGET_LANG = 'en' + +def translate(s, target=TARGET_LANG, source=''): + f = urllib.request.urlopen('http://ajax.googleapis.com/ajax/services/language/translate', urlencode({ 'v': '1.0', 'q': s, 'langpair': '%s|%s' % (source, target) })) + response = json.loads(str(f.read(), 'utf-8'))['responseData'] + return (response['translatedText'], response['detectedSourceLanguage']) + +class Plugin(BasePlugin): + def init(self): + self.add_event_handler('groupchat_message', self.on_groupchat_message) + + def on_groupchat_message(self, message): + try: + room_from = message.getMucroom() + if message['type'] == 'error': + return + + if room_from == 'poezio@muc.poezio.eu': + nick_from = message['mucnick'] + body = xhtml.get_body_from_message_stanza(message) + room = self.core.get_room_by_name(room_from) + text, lang = translate(body) + if lang != TARGET_LANG: + room.add_message(text, nickname=nick_from) + self.core.refresh_window() + except Exception as e: + import traceback + self.core.information("Exception in translator! %s" % (traceback.format_exc(),)) |