Created
May 13, 2019 18:17
-
-
Save MatrixManAtYrService/8ff1cb41c5cba865e8d43aaf3e558d4d to your computer and use it in GitHub Desktop.
parse_bot_commands
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
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 |
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
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