Last active
July 2, 2017 02:35
-
-
Save xtacocorex/009c158836a3475c7ad93ff978af936b to your computer and use it in GitHub Desktop.
The TacoBot for Slack. Need to fill out __author__ and the testbench channel code
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# TACOBOT - A FUN SLACK BOT | |
# ROBERT WOLTERMAN, 2017 | |
# YAY | |
import os | |
import sys | |
import slackclient | |
import threading | |
import time | |
import json | |
import re | |
import random | |
# CUSTOMIZE THE AUTHOR, THIS IS USED BY THE BOT FOR THE MAINTENANCE COMMAND | |
__author__ = "" | |
# ENVIRONMENT VARIABLES | |
SLACK_TOKEN = os.environ.get('SLACK_TOKEN', None) | |
# GLOBALS | |
# CUSTOMIZE THESE WITH YOUR SPECIFIC SLACK GROUP | |
# LEAVE KNOWN_BOTS AS AN EMPTY LIST IF YOUR SLACK HAS NO BOTS | |
BOT_NAME = "tacobot" | |
BOT_EMOJI = ":taco:" | |
KNOWN_BOTS = [] | |
# CLASSES | |
class TacoBot(threading.Thread): | |
# WE WANT THE GOOGLE QUERY FOR ALL THE THINGS | |
GOOGLE_QUERY = "https://www.google.com/#q={0}" | |
# LOOP DELAY WHEN READING | |
READ_WEB_DELAY = 0.5 | |
# ERROR CODES | |
ERROR_CODES = {"NULL_TOKEN" : -1, | |
"RTM_CONNECT_FAIL" : -2} | |
# ===================================================================================== | |
# CUSTOMIZE HERE FOR YOUR PURPOSES | |
# TESTBENCH IS THE CHANNEL YOU WANT TO USE TO TEST THIS BEFORE GOING LIVE | |
# THIS BOT WILL TRY TO SEND MESSAGES TO TESTBENCH ON STARTUP AND SHUTDOWN | |
# POPULATE WITH CHANNEL NAME AND CODE | |
TESTBENCH = "testbench" | |
TESTBENCH_CODE = "" | |
# REGEX FOR THINGS YOU WANT THE BOT TO RESPOND TO | |
BESTTACOS = re.compile("(who has the best tacos in|where can i find the best tacos in) (.*)") | |
BESTBURRITOS = re.compile("(who has the best burritos in|where can i find the best burritos in) (.*)") | |
SUPERHERO = re.compile("who is your favorite superhero") | |
BURRITO = re.compile("burrito") | |
TACOSHIRT = re.compile("(taco shirt|tacoshirt)") | |
FEATURES = re.compile("!features") | |
ILOVETACOS = re.compile("i love tacos") | |
# RANDOM MENTION RESPONSE LIST | |
RANDOM_RESPONSE_LIST = [ | |
[True, "what you talkin bout {0}"], | |
[False, "why do i keep hearing my name?"], | |
[False, "sometimes you all get on my nerves"], | |
[False, "stop talking and go eat some tacos!"] | |
] | |
# ===================================================================================== | |
def __init__(self, botname, emoji, token, known_bots=[]): | |
threading.Thread.__init__(self) | |
if not token: | |
print("NULL TOKEN, EXITING") | |
sys.exit(self.ERROR_CODES["NULL_TOKEN"]) | |
self.slackclient = slackclient.SlackClient(token) | |
self.botname = botname | |
self.known_bots = known_bots + [self.botname] | |
self.emoji = emoji | |
self.dead = False | |
self.connected = False | |
self.last_ts = 0.0 | |
self.channel_names_ids = {self.TESTBENCH : self.TESTBENCH_CODE} | |
# ===================================================================================== | |
# CUSTOMIZE | |
# REGEX FOR THINGS YOU WANT THE BOT TO RESPOND TO THAT REQUIRE CLASS INFORMATION | |
self.BOTNAMES = re.compile("(" + "|".join(self.known_bots) + ")") | |
self.BOTNAMEHELP = re.compile("(" + "|".join(self.known_bots) + ") help") | |
self.CREATOR = re.compile(self.botname + "(|,) (who is your creator|who made you|who created you)") | |
self.MAINTENANCE = re.compile(self.botname + " maintenance") | |
self.PING = re.compile(self.botname + " ping") | |
# ===================================================================================== | |
def kill(self): | |
self.send_me_message(self.channel_names_ids[self.TESTBENCH], "powering down") | |
self.dead = True | |
def is_alive(self): | |
return not self.dead | |
def __list_channels(self): | |
channels_call = self.slackclient.api_call("channels.list") | |
if channels_call['ok']: | |
return channels_call['channels'] | |
return None | |
def __build_channel_name_dict(self): | |
all_chans = self.__list_channels() | |
if all_chans: | |
for chan in all_chans: | |
self.channel_names_ids[chan['name'].encode('ascii', 'ignore')] = chan['id'] | |
def __channel_info(self, channel_id): | |
channel_info = self.slackclient.api_call("channels.info", channel=channel_id) | |
if channel_info: | |
return channel_info['channel'] | |
return None | |
def __get_channel_messages(self): | |
rtm_data = self.slackclient.rtm_read() | |
if rtm_data and len(rtm_data) > 0: | |
# LOOP THROUGH THE LIST | |
for dat in rtm_data: | |
# SEE IF THE TYPE IS A MESSAGE | |
if 'message' in dat['type'] and 'subtype' not in dat.keys() and float(dat['ts']) > self.last_ts: | |
self.last_ts = float(dat['ts']) | |
return dat | |
return None | |
def __get_username(self, user_id): | |
resp = self.slackclient.api_call("users.info", user=user_id) | |
if "user" in resp.keys(): | |
if "name" in resp["user"].keys(): | |
return resp['user']['name'] | |
else: | |
return None | |
else: | |
return None | |
def send_message(self, channel_id, message): | |
#print("SENDING MESSAGE TO: {0}".format(channel_id)) | |
return self.slackclient.api_call( | |
"chat.postMessage", | |
channel=channel_id, | |
text=message, | |
username=self.botname, | |
icon_emoji=self.emoji | |
) | |
def send_me_message(self, channel_id, message): | |
#print("SENDING ME MESSAGE TO {0}".format(channel_id)) | |
return self.slackclient.api_call( | |
"chat.meMessage", | |
channel=channel_id, | |
text=message, | |
icon_emoji=self.emoji) | |
def __handle_bot_help(self, channel_id): | |
# TODO: UPDATE | |
self.send_message(channel_id, "you are not worthy for help") | |
def __handle_bot_features(self, channel_id): | |
# TODO: UPDATE | |
response = """i can do the following:\n | |
- taco shirt | |
- what is your favorite superhero | |
- who has the best tacos in | |
- where can i find the best tacos in | |
- tacobot help | |
- tacobot who is your creator | |
- tacobot who made you | |
- burrito | |
- who has the best burritos in | |
- where can i find the best burritos in | |
- i love tacos | |
""" | |
self.send_message(channel_id, response) | |
# ===================================================================================== | |
# CUSTOMIZE THIS FUNCTION TO DEAL WITH THE REGEX FOR WHAT YOU WANT THE BOT TO | |
# DO AFTER IT HAS RECEIVED THE DATA FROM THE RTM CALL | |
def __handle_chan_text(self, chan_text): | |
# LOCAL VARIABLES TO MAKE OUR LIVES EASIER | |
# WE WANT THE CHANNEL TEXT LOWER CASE SO WE CAN EASILY KEY OFF IT | |
ctext = chan_text["text"].lower() | |
# GET CHANNEL ID | |
channel_id = chan_text["channel"] | |
# GET THE USER NAME OF THE SENDER | |
username = self.__get_username(chan_text['user']) | |
# CHECK TO SEE IF THE BOTNAME IS ANYWHERE IN THE TEXT | |
# REGEX ALL UP FRONT | |
bothelp = self.BOTNAMEHELP.match(ctext) | |
besttacos = self.BESTTACOS.search(ctext) | |
creator = self.CREATOR.search(ctext) # WANT THIS TO BE A SEARCH IN CASE OF ? | |
maintenance = self.MAINTENANCE.match(ctext) | |
superhero = self.SUPERHERO.search(ctext) | |
burrito = self.BURRITO.search(ctext) | |
tacoshirt = self.TACOSHIRT.search(ctext) | |
features = self.FEATURES.search(ctext) | |
bestburritos = self.BESTBURRITOS.search(ctext) | |
ilovetacos = self.ILOVETACOS.search(ctext) | |
ping = self.PING.match(ctext) | |
if bothelp: | |
# FIRST OFF, IF WE FIND OTHER BOTS AND THEY ARE ASKED FOR HELP | |
# RESPONE WITH /me | |
if bothelp.group(1) == self.botname: | |
# ME HELP | |
self.__handle_bot_help(channel_id) | |
else: | |
# KNOWN BOT HELP | |
self.send_me_message(channel_id, "is so lonely") | |
elif ping: | |
# PONG | |
if username: | |
self.send_message(channel_id, "you called {0}".format(username)) | |
else: | |
self.send_message(channel_id, "pong") | |
elif features: | |
# FEATURES | |
self.__handle_bot_features(channel_id) | |
elif besttacos: | |
# BEST TACOS | |
url = self.GOOGLE_QUERY.format(besttacos.group(0).strip().replace(",","").replace(" ","+")) | |
resp = self.send_message(channel_id, url) | |
elif bestburritos: | |
# BEST BURRITOS | |
url = self.GOOGLE_QUERY.format(bestburritos.group(0).strip().replace(",","").replace(" ","+")) | |
resp = self.send_message(channel_id, "while i prefer tacos, if you must know:\n{0}".format(url)) | |
elif creator: | |
# CREATOR | |
self.send_message(channel_id, "we all know {0} is my creator".format(__author__)) | |
elif maintenance: | |
# MAINTENANE | |
if username == __author__: | |
# ONLY DO STUFF IF IT'S THE CREATOR | |
self.send_message(channel_id, "NOOOOOO!!!!") | |
self.kill() | |
else: | |
self.send_me_message(channel_id, "{0} doesn't have that power :stuck_out_tongue_winking_eye:".format(username)) | |
elif superhero: | |
# SUPERHERO | |
self.send_message(channel_id,"my favorite is !batman") | |
elif burrito: | |
# BURRITO | |
if username: | |
self.send_message(channel_id, "what is wrong with you {0}, tacos are better!".format(username)) | |
elif tacoshirt: | |
# TACO SHIRT | |
self.send_message(channel_id, "https://cottonbureau.com/products/every-day-is-taco-tuesday") | |
elif ilovetacos: | |
# I LOVE TACOS | |
self.send_message(channel_id, "ME TOO! :flying_taco:") | |
elif self.botname in ctext: | |
if username: | |
loc = random.randint(0, len(self.RANDOM_RESPONSE_LIST)-1) | |
if self.RANDOM_RESPONSE_LIST[loc][0]: | |
self.send_message(channel_id, self.RANDOM_RESPONSE_LIST[loc][1].format(username)) | |
else: | |
self.send_message(channel_id, self.RANDOM_RESPONSE_LIST[loc][1]) | |
# ===================================================================================== | |
def run(self): | |
# BUILD CHANNEL LIST | |
self.__build_channel_name_dict() | |
# CONNECT RTM INTERFACE | |
self.connected = self.slackclient.rtm_connect() | |
if not self.connected: | |
print("ERROR CONNECTING, EXITING") | |
sys.exit(self.ERROR_CODES["RTM_CONNECT_FAIL"]) | |
# NOTICE | |
print("ENSURE BOT IS INVITED TO CHANNELS YOU WANT TO IT TO ACCESS") | |
# LOOP UNTIL DEAD | |
print("STARTING RUN LOOP") | |
self.send_me_message(self.channel_names_ids[self.TESTBENCH], "powering up") | |
while not self.dead: | |
try: | |
# LOOP ON OUR CHAN KEYS | |
chan_text = self.__get_channel_messages() | |
if chan_text: | |
# DEBUG | |
#print(chan_text) | |
# DETERMINE WHAT TO DO | |
self.__handle_chan_text(chan_text) | |
# SLEEP A BIT | |
time.sleep(self.READ_WEB_DELAY) | |
except KeyboardInterrupt: | |
print("HIT KEYBOARD INTERRUPT INSIDE TacoBot.run()") | |
self.dead = True | |
# FUNCTIONS | |
def main(): | |
bot = TacoBot(BOT_NAME, BOT_EMOJI, SLACK_TOKEN, KNOWN_BOTS) | |
bot.start() | |
try: | |
# LOOP WHILE THE BOT IS ALIVE, IT CAN BE KILLED FROM THE | |
# __author__ USER COMMANDING IT VIA SLACK | |
while bot.is_alive(): | |
time.sleep(1) | |
except KeyboardInterrupt: | |
bot.kill() | |
if __name__ == '__main__': | |
main() | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment