Skip to content

Instantly share code, notes, and snippets.

@MatrixManAtYrService
Created May 13, 2019 18:17
Show Gist options
  • Save MatrixManAtYrService/8ff1cb41c5cba865e8d43aaf3e558d4d to your computer and use it in GitHub Desktop.
Save MatrixManAtYrService/8ff1cb41c5cba865e8d43aaf3e558d4d to your computer and use it in GitHub Desktop.
parse_bot_commands
for event in slack_events:
if event['type'] == 'message':
mentiond_id, message = parse_direct_mention(event['text'])
# is it for me?
if mentiond_id == self_id:
# classify recipient
from_user = 'subtype' not in event
try:
from_bot = (event['subtype'] == 'bot_message')
except KeyError:
from_bot = False
from_self = False
try:
from_self = event['username'] == bot_user.display_name
except:
pass
# extract originating user
if from_user:
from_user = lookup_user(event['user'])
elif from_bot and not from_self:
from_user = lookup_user(event['bot_id'])
else:
print("Unknown message source, disregarding$")
print(event)
break
return message, event['channel'], from_user
user_cache = {}
# provide api_header as either 'bots' or 'users'
# provide either user ids or user names
def _lookup(api_header, id_str):
# find by userid
if api_header == 'bots':
response = slack_client.api_call(f"{api_header}.info", bot=id_str)
return response['bot']
else:
response = slack_client.api_call(f"{api_header}.info", user=id_str)
# if not found, list them all and search by username
if 'user' not in response:
list_response = slack_client.api_call(f"{api_header}.list")
ids_by_name = { x['name'] : x['id'] for x in list_response['members'] }
user_id = ids_by_name[id_str]
# if found, now get them by userid
response = slack_client.api_call(f"{api_header}.info", user=user_id)
return response['user']
class User:
def __init__(self, api_dict):
self.id = api_dict["id"]
self.mention = f'<@{self.id}>'
try:
self.display_name = api_dict['profile']['display_name']
except KeyError:
self.display_name = api_dict['name']
try:
self.tz_offset = api_dict['tz_offset']
except KeyError:
self.tz_offset = 0
def local_time_str(self, from_time_utc):
local_time = datetime.fromtimestamp(from_time_utc + self.tz_offset)
return local_time
def __str__(self):
return self.display_name
def lookup_user(id_str):
global user_cache
# if they're cached, return their details
try:
return user_cache[id_str]
# otherwise, look them up and cache them
except KeyError:
user = None
try:
response = _lookup('users', id_str)
except KeyError:
response = _lookup('bots', id_str)
user = User(response)
user_cache[id_str] = user
user_cache[user.id] = user
user_cache[user.display_name] = user
return user_cache[id_str]
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment