This module implements the commands a user can give. It also contains the global cmnds object to which all commands are added.
example: cmnds.add(‘hello’, handle_hello, ‘USER’)
Bases: gozerbot.commands.Commands
commands for the bot .. dispatch with (bot, ircevent) as arguments.
dispatch on event passing bot and ievent as arguments.
Bases: object
implements a command.
Bases: dict
the commands object is a dict containing the commands. dict key is the command (1 word).
activate commands of a plugin.
add a command.
search for command.
de-activate all commands of a plugin.
dispatch a command.
return command matching txt.
get all commands of a plugin.
get all function names of commands in a plugin.
get options of a command.
list commands with permission perm.
return options dict of command.
overload permission of function with funcname.
get permission of command.
set permission of command.
number of commands.
unload plugin commands.
return all possible permissions.
locate plugin a command is registered in.
# gozerbot/commands.py # # """ This module implements the commands a user can give. It also contains the global cmnds object to which all commands are added. example: cmnds.add('hello', handle_hello, 'USER') """ __status__ = "seen"
from gozerbot.stats import stats from utils.generic import makeoptions from eventbase import defaultevent from config import config from utils.log import rlog from utils.trace import calledfrom from utils.exception import handle_exception from utils.locking import lockdec from runner import cmndrunners from threads.thr import start_new_thread, start_bot_command
import sys import re import copy import types import thread
commandlock = thread.allocate_lock() locked = lockdec(commandlock)
class Command(object): """ implements a command. """ def __init__(self, func, perm, plugname, speed=5, threaded=False, allowqueue=True, options={'--speed': 5, '--chan': '', '--filter': '', '--how': 'msg'}): self.name = str(func) # function name is string representation of the function self.func = func # function to call if type(perm) == types.ListType: self.perms = list(perm) else: self.perms = [perm, ] self.plugname = plugname # plugin name self.speed = copy.deepcopy(speed) # speed to execute command with self.threaded = copy.deepcopy(threaded) # set if threaded exec is required self.allowqueue = copy.deepcopy(allowqueue) # set if command is allowed to be used in pipeline self.options = dict(options) # options set in the command self.activate = True
class Commands(dict): """ the commands object is a dict containing the commands. dict key is the command (1 word). """ def __setitem__(self, name, value): """ set command. """ dict.__setitem__(self, name, value) def __delitem__(self, name): """ delete command. """ dict.__delitem__(self, name) def size(self): """ number of commands. """ return len(self) def activate(self, plugname): """ activate commands of a plugin. """ cp = dict(self) for i in cp.values(): if i.plugname == plugname: i.activate = True def disable(self, plugname): """ de-activate all commands of a plugin. """ cp = dict(self) for i in cp.values(): if i.plugname == plugname: i.activate = False def whatperms(self): """ return all possible permissions. """ result = [] cp = dict(self) for i in cp.values(): for j in i.perms: if j not in result: result.append(j) return result def list(self, perm): """ list commands with permission perm. """ result = [] if type(perm) != types.ListType: perm = perm.upper() perms = [perm, ] else: perms = perm cp = dict(self) for name, cmnd in cp.items(): for i in perms: if i in cmnd.perms: result.append(name) return result def getfuncnames(self, plug): """ get all function names of commands in a plugin. """ result = [] cp = dict(self) for i in cp.values(): if i.plugname == plug: result.append(i.func.func_name) return result def getoptions(self, command): """ get options of a command. """ cp = dict(self) for name, cmnd in cp.iteritems(): if name == command: return makeoptions(defaultevent, cmnd.options) def permoverload(self, name, perms): """ overload permission of function with funcname. """ perms = [perm.upper() for perm in perms] for com in self.values(): try: if com.func.func_name == name: com.perms = perms return True except AttributeError: rlog(10, 'commands', "permoverload: no %s function" % name) return False def add(self, cmnd, func, perm, speed=5, threaded=False, allowqueue=True, options={'--speed': 5, '--chan': '', '--filter': '', '--how': 'msg'}): """ add a command. """ plugname = calledfrom(sys._getframe(0)) if config['loadlist'] and plugname not in config['loadlist']: rlog(1, 'commands', 'NOT LOADING %s' % plugname) return rlog(-3, 'commands', 'added %s (%s) ' % (cmnd, plugname)) self[cmnd.lower()] = Command(func, perm, plugname, speed, threaded, allowqueue, options) self[cmnd.lower()].name = cmnd.lower() def apropos(self, txt, perms=[]): """ search for command. """ result = [] cp = dict(self) for name, cmnd in cp.iteritems(): if perms: go = False for i in perms: if i in cmnd.perms: go = True if not go: continue if re.search(txt, name): result.append(name) return result def unload(self, plugname): """ unload plugin commands. """ results = [] for name, cmnd in self.iteritems(): if cmnd.plugname == plugname: results.append(name) got = False for name in results: try: del self[name] rlog(-3, 'commands', 'unloaded %s (%s)' % (name, plugname)) got = True except KeyError: got = False if got: return True def whereis(self, command): """ locate plugin a command is registered in. """ result = [] cp = dict(self) for name, cmnd in cp.iteritems(): if name == command: if not cmnd.plugname in result: result.append(cmnd.plugname) return result def perms(self, name): """ get permission of command. """ name = name.lower() if self.has_key(name): return self[name].perms else: return [] def setperm(self, command, perm): """ set permission of command. """ command = command.lower() perm = perm.upper() if self.has_key(command): if perm not in self[command].perms: self[command].perms.append(perm) return True return False def getcommand(self, txt): """ return command matching txt. """ textlist = txt.split() if not textlist: return None cmnd = textlist[0].lower() if self.has_key(cmnd): return self[cmnd] def options(self, command): """ return options dict of command. """ try: return self[command].options except KeyError: pass def getcommands(self, plugin): """ get all commands of a plugin. """ result = [] tmp = dict(self) for name, cmnd in tmp.iteritems(): if cmnd.plugname == plugin: result.append(name) return result def dispatch(self, com, txt, wait=False): """ dispatch a command. """ if com.threaded: thread = start_new_thread(com.func, (txt, )) if wait: thread.join() else: cmndrunners[10-com.speed].put(com.name, com.func, txt) return 1
class Botcommands(Commands): """ commands for the bot .. dispatch with (bot, ircevent) as arguments. """ def dispatch(self, com, bot, ievent, wait=False): """ dispatch on event passing bot and ievent as arguments. """ if not com.activate: return False if bot.stopped: return False stats.up('cmnds', com.name) stats.up('cmnds', com.plugname) stats.up('cmnds', 'speed%s' % com.speed) if com.threaded or ievent.threaded: thread = start_bot_command(com.func, (bot, ievent)) #if thread and wait: thread.join() else: speed = ievent.speed or com.speed ievent.speed = speed cmndrunners.put(com.name, com.func, bot, ievent) return True
cmnds = Botcommands()