Slackbot development in Python

Introduction

This article is the 24th day article of SLP KBIT Advent Calendar 2019. It's been a long time, I'm Gakki. This time, I created Slackbot, so I will write a record of it.

Development history

In my circle, Slack is used as the main contact tool with ** free plan **. Under such circumstances, many members use it as a workspace, so Sometimes mentions are made on unrelated topics, so I wondered if I could alleviate that. I decided to create a bot that can manage user groups. (It doesn't make sense to have notifications enabled for most messages ...)

Development environment and technology used

• Ubuntu 18.04 LTS
• Python3.6
  • slackbot
  • slack-client

Introducing bots to your workspace

I referred to the site of here.

Development flow

From here, we will start the main subject. I will write it separately in the following contents.

--Permission setting --Folder structure --Coding

Permission setting

In order to manage the user group created this time, Since it was not possible to do it only with the authority as a bot user, we set the necessary authority as follows.

Folder structure

The folder structure is as follows. The processing beyond the function of the bot user is described in my_mention.py and subMethod.py, and it is imported by run.py of main.

coding

This time, we implemented a group of commands for managing user groups. In addition, I also created a command related to the questionnaire, so I will write a little about it. The main functions are as follows.

--Create user group --Delete user group --Edit user group members --User group mention

First, as a premise, I will explain from the concept of the user group created this time. As mentioned earlier, the workspace with the bot is a free plan, so API methods related to user groups cannot be used. Therefore, create a dictionary with key as the user group name and value as an array of group members. I decided to make it a pseudo user group by saving it in a static file. Therefore, please understand that it is the case when the contents related to the dictionary appear in the subsequent programs.

Creating a user group

The program is as follows.

@respond_to('create\s([a-zA-Z0-9]*)\s([\w\s,]+)')
def create_usergroup(message, usergroup_name, member):
    usergroup = subMethod.get_usergroup_list()
    member_list = subMethod.get_member()['members']
    for usergroup_dict in usergroup:
        if usergroup_dict['usergroup_name'] == usergroup_name:
            message.send("`" + usergroup_name+' is already exist.`\n> please choose another name.')
            return
    data = {}
    member_id = []
    data['usergroup_name'] = usergroup_name
    try:
        member_name = [x.strip() for x in member.split(',')]
    except AttributeError:
        member_name = []
        member_id = member
    ml_id = [ml['id'] for ml in member_list]
    ml_name = [ml['name'] for ml in member_list]
    ml_rname = [ml['real_name'] if 'real_name' in ml else 'no_name' for ml in member_list]
    ml_dname = [ml['profile']['display_name'] for ml in member_list]
    for mn in member_name:
        if mn in ml_name:
            member_id.append(ml_id[ml_name.index(mn)])
        elif mn in ml_rname:
            member_id.append(ml_id[ml_rname.index(mn)])
        elif mn in ml_dname:
            member_id.append(ml_id[ml_dname.index(mn)])
        else:
            message.send("`" + mn + " is not in this channel`")
    data['member'] = member_id
    usergroup.append(data)
    subMethod.set_usergroup_list(usergroup)
    message.send('Created a usergroup')

Executed when a message of the format create {user group name} {member name,…} is received. As an internal process, first check if the specified user group has already been created. After checking, if it does not exist yet, create it and add members. At this time, it is confirmed at the same time whether the specified member exists in the workspace. Users existing in the workspace can get a list by using the user_list method. However, the problem here is the format. In slack, users can have values in three formats: userID, fullname, and displayname. Since it is difficult to send an ID when sending a message, I think that you will specify full name or display name, so I made it possible to perform mutual conversion there. After this process, if the user exists, the ID is added, and if it does not exist, an error message is sent. The reason for adding the ID will be explained in the section on group mention below. When executed, it will be as below.

Delete user group

The program is as follows.

@respond_to('delete_usergroup\s([a-zA-Z0-9]*)')
def delete_usergroup(message, usergroup_name):
    usergroup = subMethod.get_usergroup_list()
    usergroup_name_list = [x['usergroup_name'] for x in usergroup]
    if usergroup_name not in usergroup_name_list:
        message.send("`" + usergroup_name + ' is not exist.`\n> type `@secretary list` and check usergroup_name')
        return
    new_usergroup = []
    for usergroup_dict in usergroup:
        if usergroup_dict['usergroup_name'] == usergroup_name:
            continue
        new_usergroup.append(usergroup_dict)
    subMethod.set_usergroup_list(new_usergroup)
    message.send('Deleted a usergroup')

Receives and executes a message in the format delete_usergroup {user group name}. In this regard, it's almost just a dictionary operation. If there is a specified user group, just delete it from the dictionary. If not, an error message will be sent. When I try to use it, it looks like this.

Editing user group members

The program is as follows.

@respond_to('add\s([a-zA-Z0-9]*)\s([\w\s,]+)')
def add_member(message, usergroup_name, member):
    usergroup = subMethod.get_usergroup_list()
    usergroup_name_list = [usergroup_dict['usergroup_name'] for usergroup_dict in usergroup]
    if usergroup_name not in usergroup_name_list:
        message.send("`" + usergroup_name + " is not exist`\n> please type `@secretary list` and check usergroup_name.")
        return
    member_list = subMethod.get_member()['members']
    usergroup_member = subMethod.get_usergroup_member(usergroup_name)

    member_id = []
    try:
        member_name = [x.strip() for x in member.split(',')]
    except AttributeError:
        member_name = []
        member_id = member
    add_member_name = []
    for mn in member_name:
        if mn not in usergroup_member:
            add_member_name.append(mn)
        else:
            message.send("`" + mn + ' already belongs`')
    ml_id = [ml['id'] for ml in member_list]
    ml_name = [ml['name'] for ml in member_list]
    ml_rname = [ml['real_name'] if 'real_name' in ml else 'no_name' for ml in member_list]
    ml_dname = [ml['profile']['display_name'] for ml in member_list]
    for mn in add_member_name:
        if mn in ml_name:
            member_id.append(ml_id[ml_name.index(mn)])
        elif mn in ml_rname:
            member_id.append(ml_id[ml_rname.index(mn)])
        elif mn in ml_dname:
            member_id.append(ml_id[ml_dname.index(mn)])
        else:
            message.send("`" + mn + " is not in this channel`")
    if len(member_id) == 0:
        message.send("`No one will add`")
        return
    for usergroup_dict in usergroup:
        if usergroup_dict['usergroup_name'] == usergroup_name:
            usergroup_dict['member'].extend(member_id)
            usergroup_dict['member'] = list(set(usergroup_dict['member']))
            break
    subMethod.set_usergroup_list(usergroup)
    message.send('Added some member')

@respond_to('delete\s([a-zA-Z0-9]*)\s([\w\s,]+)')
def delete_member(message, usergroup_name, member):
    usergroup = subMethod.get_usergroup_list()
    usergroup_name_list = [usergroup_dict['usergroup_name'] for usergroup_dict in usergroup]
    if usergroup_name not in usergroup_name_list:
        message.send("`" + usergroup_name + " is not exist`\n> type `@secretary list` and check usergroup_name")
        return
    member_list = subMethod.get_member()['members']
    member_id = []
    try:
        member_name = [x.strip() for x in member.split(',')]
    except AttributeError:
        member_name = []
        member_id = member
    ml_id = [ml['id'] for ml in member_list]
    ml_name = [ml['name'] for ml in member_list]
    ml_rname = [ml['real_name'] if 'real_name' in ml else 'no_name' for ml in member_list]
    ml_dname = [ml['profile']['display_name'] for ml in member_list]
    for mn in member_name:
        if mn in ml_name:
            member_id.append(ml_id[ml_name.index(mn)])
        elif mn in ml_rname:
            member_id.append(ml_id[ml_rname.index(mn)])
        elif mn in ml_dname:
            member_id.append(ml_id[ml_dname.index(mn)])
        else:
            message.send("`" + mn + " is not in this channel`")
    if len(member_id) == 0:
        message.send("`No one will delete`")
        return
    for usergroup_dict in usergroup:
        if usergroup_dict['usergroup_name'] == usergroup_name:
            for mi in member_id:
                if mi not in usergroup_dict['member']:
                    message.send("`" + ml_name[ml_id.index(mi)] + " doesn't belong to this`")
                else:
                    usergroup_dict['member'].remove(mi)
            break
    subMethod.set_usergroup_list(usergroup)
    message.send('Deleted some member')

Add a member to the user group for the message ʻadd {user group name} {member name,…} Deletes the member from the user group for the messagedelete {user group name} {member name,…}`. The additional processing is the same as the creation processing if the specified user group exists. In the deletion process, when the specified user group exists, the members belonging to that user group are deleted one by one. It is an error if the specified user group and members do not exist.

User group mention

The program is as follows.

@listen_to('@[a-zA-Z0-9]+\s([\s\S]*)')
def reply_to_thread(message, text):
    usergroup = subMethod.get_usergroup_list()
    message.body['text'].replace('\n', ' ')
    mention = message.body['text'].split()[0].strip('@')
    mention_dict = []
    for dictionary in usergroup:
        if dictionary['usergroup_name'] == mention:
            mention_dict = dictionary
            break
    if len(mention_dict) == 0:
        message.send('`' + mention + ' is not exist`')
        return
    sentence = ""
    for member in mention_dict['member']:
        sentence = sentence + "<@" + member + "> "
    sentence = sentence + "\n"
    message.send(sentence, 
            thread_ts=message.thread_ts)

This method is different from what I wrote earlier. The methods up to this point can only be executed in conjunction with the mention to the bot, This method responds to messages of the form @ {user group name} {message} without it. The method written as respond requires mention, the method written as listen is unnecessary. Although it is an internal process, the @ {user group name} included at the beginning of the message is extracted and Extracts the value of the element with the specified user group name as the key. As explained earlier, value is an array format, so the for statement takes out the elements one by one and combines them to create a message. This is the message for mention. Here, ID is used in slack when mentioning. Therefore, it was necessary to memorize the ID. If the created message is sent to TL, it will only get in the way if there are too many members in the user group. Therefore, this time, the message is sent in the form of a thread for the original message. The method for sending to a thread is message.send (sentence, thread_ts = message.thread_ts). The actual screen looks like the one below.

The methods related to user groups are as follows. In addition to the ones mentioned above, a list of groups and members, There are things like changing the group name and combining user groups, but I will omit it because it can be done with the combination so far.

bonus

When I took a questionnaire on slack, I was trying to get a reaction to the message. With this format, it was very troublesome to see the reactions one by one and confirm who voted where. When you create a user group, if you group the people who will answer the questionnaire into a user group, I thought that the bot could handle it, so I made about two functions. (It seems that a simple poll has been released recently ...)

Questionnaire total

The program is as follows.

@respond_to('count')
def count_up_reaction(message):
    response = subMethod.get_message(message.body['channel'], 
                                    message.thread_ts)
    if not response:
        message.direct_reply("Can't use count method in DM")
        return
    sentence = ''
    if 'reactions' in response['messages'][0]:
        data = response['messages'][0]['reactions']
        sorted_data = sorted(data, reverse=True, key=lambda x:x['count'])
        sentence = response['messages'][0]['text'] + '\n\n*Result*\n'
        for datum in sorted_data:
            sentence = sentence + ":" + datum['name'] + ":" + " "
            for user in datum['users']:
                sentence = sentence + "<@" + user + "> "
            sentence = sentence + "\n"
    else:
        sentence = 'No reactions'
    message.direct_reply(sentence)

It is executed by sending count to the thread of the message for which you want to aggregate the reactions. As a process, when you get the message at the top of the thread, there is a part that summarizes the reaction data, so organize it. The reaction data is included in the form below. After all, since it is a dictionary, it acquires data, molds it, and sends it to the user who sent count by DM. The execution screen is as shown below.

Questionnaire difference aggregation

The program is as follows.

@respond_to('diff')
def check_reactor(message):
    response = subMethod.get_message(message.body['channel'],
                                    message.thread_ts)
    if not response:
        message.direct_reply("Can't use count method in DM")
        return
    target_usergroup = response['messages'][0]['text'].replace('\n', ' ').split()[0].strip('@')
    all_target_audience = subMethod.get_usergroup_member_id(target_usergroup)
    if len(all_target_audience) == 0:
        sentence = 'No specified user group'
    elif 'reactions' in response['messages'][0]:
        data = response['messages'][0]['reactions']
        reacted_users = []
        reacted_users.extend([user for datum in data for user in datum['users']])
        target_audience = []
        target_audience.extend([user for user in all_target_audience if user not in reacted_users])
        sentence = "*Hasn't yet reacted*\n"
        for user in target_audience:
            sentence = sentence + "<@" + user + ">\n"
    else:
        sentence = "*Hasn't yet reacted*\n"
        for user in all_target_audience:
            sentence = sentence + "<@" + user + ">\n"
    message.direct_reply(sentence)

As with user group mentions, specify the target user group at the beginning of the message. Then, send diff to the thread of the target message to perform aggregation. First, extract the target user group from the message and get an array of members. Next, get a list of users with reactions in some way. If you can get these two, extract the users who exist in only one of the two arrays. Arrange them and send them by DM. The execution screen is as shown below.

in conclusion

The slackbot can be created intuitively using the GUI, and the installation itself can be done intuitively. As for the function implementation, I had the impression that it is relatively easy to work on Python because there is a good library. Currently, there are still few functions, so I will continue to code in the future. Also, as always, I write the same process in various places, Since the file name is also in a terrible state, it is in a terrible state. I want to do the refactoring firmly. Finally, this code has been uploaded on GitHub, so please have a look if you like. I've also written about Docker, so if you have an environment where you can use Docker, I think that you can use it as soon as you set the token.