1
\$\begingroup\$

simpler pattern/extension built on top of telepot (telegram bot api wrapper)

filter usage

from nimmy import Bot
from re import match, search

bot = Bot("TELEGRAM_BOT_API_TOKEN")

# filter model function returns, keys return None on failure
def adminOnly(msg):
    return {
        'admin': match('fatzombi', msg.chat['username'])
    }


def ping(msg):
  return {
    'text': search("^ping\s(.*)", msg.text)
    }


# executes only when filtering contents match takes filter function and check type: any or all
@bot.filter(adminOnly)
@bot.filter(ping)
def handle_this(msg):
  from subprocess import check_output

  hostname = msg.filter['text'].group(1)
  response = check_output(['ping', '-c 1', hostname])

  bot.sendMessage(msg.chat['id'], response)

  # feed the monitor!
  print("ping -> " + hostname)

# takes filtered functions as handle, binded on message_loop
bot.add_handle(handle_this)

# a simple run forever loop, takes message_loop args as loop={}
bot.run_forever()

from telepot import Bot
from functools import wraps
from nimmy.utils import Map

telepot.Bot extended

class Bot(Bot):
    """
    extending telepot.Bot
    """
    bot_functions = {}

    def _filter(self, filter_model, message, check):
        # checks for any exceptions with filters, find None, true or false
        try:
            filters = filter_model(message)

            if check(filters.values()):
                return filters
        except:
            return False

    def filter(self, filter_model, check=all):
        # argumented decoration, returns func with msg
        def decorator(func):
            # prevents cloning
            @wraps(func)
            def wrapper(msg):
                # converts dict to object with utils.Map
                message = Map(**msg)
                filterd = self._filter(filter_model, message, check)

                if filterd:
                    message.filter = filterd
                    return func(message)
            return wrapper
        return decorator

    def run_forever(self, sleep_for=10, **loop):
        # telepot message_loop with dummy_handler
        print('Loading ...')
        self.message_loop(self.dummy_handler, **loop)

        from time import sleep
        # simple loop sleeps: sleep_for
        while 1:
            sleep(sleep_for)

    def add_handle(self, handle):
        # add filter handles to a list
        try:
            Bot.bot_functions[id(self)]
        except KeyError:
            Bot.bot_functions[id(self)] = []

        Bot.bot_functions[id(self)].append(handle)

    def dummy_handler(self, msg):
        # acts like a handle returns a list of handles and execute
        return [h(msg) for h in Bot.bot_functions[id(self)]]
\$\endgroup\$
  • \$\begingroup\$ Does this work? Because I'd expect you to have various issues due to shadowing Bot. \$\endgroup\$ – jonrsharpe Dec 28 '16 at 18:43
  • \$\begingroup\$ yes, there was but, it works well for my uses, but i didn't seen any problems with current solution \$\endgroup\$ – FATZOMBI Dec 28 '16 at 19:18

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.