Referenced Material
https://github.com/MMacLaine/Slack-Bot – All scripts reference can be found here

Background information
I recently found myself working for a company which uses Slack as their primary IM tool, this being a new tool for me as someone who has a background in Microsoft and as you can safely assume Microsoft Teams. We started to run into an issue with Slack, when I started working we had just under 300 members of staff and by the end of 2021 this number will be nearer 3000. With this huge amount of growth this leads to a very ugly workspace here on Slack and as time has gone on we now have a problem:

  • Too many channels which are no longer in use.
  • Channels which were set up with not much thought and no longer serve their original purpose.
  • Channels which were created by staff who longer work here and as a result have not had their channel archived when having their account disabled.
  • Channels are having their own naming conventions which are making it harder to find what people are looking for.
  • Many other niche problems.

With this problem at hand and my interest in learning Python I decided to have a look into creating a Python bot for slack which operated using commands to perform certain actions. The method that these would work through was primarily based user interaction based on Webhooks.

Starting process
With the background explained and this being a learning opportunity for myself as I had never touched Python up until this point, I decided it would be best to set up my own Slack Sandbox and complete testing here. Ultimately I want to interact with prod as little as possible, even with my now complete script if changes are needed these are tested multiple times in my sandbox environment before pushing into prod.

When I originally started looking into this I, rather naively, assumed the majority of the work would result in scripting the functions and how they interact with API calls but quickly learned that this wouldn’t be the case. Some elements I had to consider:

  • Security
  • Local deployment for testing
  • prod deployment, in my case I opted for AWS which was new for me.
  • Access Management
  • Version control
  • How to check the status of the bot without interacting with the bot

I also quickly started to realise that this project would involve a lot more than Python. All in all I ended up using a number of new languages/software which I wasn’t expecting to at the start of this project. What you will end up needing:

There were a few more elements but the list would get a bit too long. The point is that what I imagined going into it and what was actually needed were vastly different, a good thing to consider when doing similar projects.

What does the bot actually do?
This is the important part of the bot, what does it actually do? Well it will take the command that you send it using Flask and Webhooks to detect what you’ve sent and then action said command. So what are the commands that the bot can do?

– Guidelines
This command will check the naming conventions of each channel within Slack. Firstly this will go through every channel’s description and look for the word “social” which should be used to state if a channel should be skipped on checking or not. If Social is not contained it will be looped through again and check the beginning of the name of each channel to confirm if it matches the Ragex rule which must be followed. The current rules are to check the beginning of the channel must contain:

all|announcement|se|fr|uk|no|de|global|ask|bet|proj|fdbk|team|fun

As an example “All-Hands” would pass, “Al-Hands” would fail. This is easily changed and added to as it simply detects the beginning word is matching. If a channel fails it will report this back into the main channel #Slack-Management. Important to note, the logic to post a warning into each channel does exist as well as a direct message to the channel creator with a default message. These functions have been commented out and are easily enabled by removing the comments and changing the messages as required.  Optional functions to message the slack channel or channel creator are commented out and can be enabled / edited as needed.

– Members
The Members command simply goes through every channel and lists how many members are in them. It will print it out as  Channel – #MemberCount

– History
The History command will go through all channels and will collect 4 bits of key information from each: The channel name, the name of the user who last messaged into the channel, how many days ago it was since this message was made and what the last message was. This is printed into #Slack-Management channel. All channels over 365 days with no activity will be archived.

– Empty
The Empty command is the main and will likely be the most used command. The bot will firstly check for all channels which are 0 members in them, this will automatically prompt these channels to be archived. It will then check for all channels which contain 50 or less members and read the chat history. If the channel has not been used in 90 days it will automatically archive the channel involved. If the channel has over 50 members this limit will increase to 180 days.

All channels which are archived will be reported on by the bot and shared in the main #slack-management channel. Important to note this targets every single public channel on slack regardless of social or non-social use.

– Warning
This command works off the same logic as Empty, however, this will simply identify who the channel creator is, when doing so they will be messaged by the bot giving them warning about what channel will soon be archived, allowing the owner some time to action keeping the channel from being removed.

– Help
This command is a simple print text command which only the user who prompts it will be able to see. The message is a simple explanation of the supported commands and appears like: 

*Outdated screenshot, some functions added since

– Greetings
This command was purely for testing but has been left in, it’s a simple command which replies with a welcome message when sent to the bot, the supported greetings are Hello, Hello there, Hey, Hej However due to how python and slack progress commands hi was removed due to triggering when history was used as a command cursing issues.

Security Considerations
This bot has the ability to perform a number of actions which could be problematic if used incorrectly as well as the fact it has certain scopes via api.slack.com which are needed meaning maintaining and hosting the bot is incredibly important to be done correctly.

From a perspective of how we manage the bot within the slack environment we have created a function which will detect where a user is running any command from before attempting to run a command. If the command is not being run via #slack-management, a private and locked channel, they will be asked to check where the command is being run from and rejected. This way only the correct people may run commands.

From an API/hosting perspective this bot has been added into the an AWS environment for hosting. It pulls directly from the GitHub environment which is where the code is security stored and maintained. The bot makes use of three different security keys which connecting from AWS to Slack:

Access Management
Access management is 100% controlled via the slack channel ID you assign in the script. To have the ability to run any commands you must be a member of this slack channel, anyone outside this (you should ensure this is a private channel) will not be able to run any commands. Adding a new member is simply and can be done by inviting a user to the channel.

Local Deployment
Local deployment is useful when running tests via our slack sandbox environment, I would recommend all features and testing be completed here before pushing to production. Important to keep in mind is when using local deployment you will need to change the 3 secret variables in the docker file, these are commented out and can easily be stored in whatever SDK/IDE you wish to use.

For local hosting we use ngrok which allows for easy local hosting. You can run this by using the command this will push a local web url which should look something like this:  https://b01cf107226d.ngrok.io -> http://localhost:8086   

From here you will need to go to https://api.slack.com/apps and select the test app, in our case this is currently named Starter bot located in your workspace. You will then navigate to event subscriptions which is where you will be able to to add the url: 

IMPORTANT: You must add api/slack-bot/events in front of the URL for Event Subscriptions to detect the change.

AWS Deployment
You should avoid using AWS or whatever platform unless you are somewhat knowledgeable, in the instance of AWS we manage this as:

  • Push/RP on main branch
  • AWS codebuild: run x build with parameters, overriding the branch to main
  • Check aws ECS for x and wait for “Reached steady state” message

This project has been deployed using the service-template. If you wish to introduce CI/CD this is a must.

The Script itself (main.py)

# Required modules
import os
import re
import json
from flask import Flask, Response
from slackeventsapi import SlackEventAdapter
from threading import Thread
from slack import WebClient
from slack.errors import SlackApiError
from datetime import datetime

# This `app` represents your existing Flask app
app = Flask(__name__)

slack_token = os.environ['SLACK_BOT_TOKEN']
SLACK_SIGNING_SECRET = os.environ['SLACK_BOT_SIGNING_SECRET']
VERIFICATION_TOKEN = os.environ['SLACK_BOT_VERIFICATION_TOKEN']

# instantiating slack client
slack_client = WebClient(slack_token)


@app.route("/api/slack-bot")
def root():
    return 'Welcome to Slack-Bot'


# Flask health check
@app.route("/api/slack-bot/_health")
def health():
    return {"status": 200}


@app.route("/api/slack-bot/events")
def event_hook(request):
    json_dict = json.loads(request.body.decode("utf-8"))
    if json_dict["token"] != VERIFICATION_TOKEN:
        return {"status": 403}

    if "type" in json_dict:
        if json_dict["type"] == "url_verification":
            response_dict = {"challenge": json_dict["challenge"]}
            return response_dict
    return {"status": 500}


slack_events_adapter = SlackEventAdapter(
    SLACK_SIGNING_SECRET, "/api/slack-bot/events", app
)


@slack_events_adapter.on("app_mention")
def handle_message(event_data):
    def send_reply(value):
        message = value["event"]

        if message.get("subtype") is None:
            command = message.get("text")
            channel_id = message["channel"]

            # check if user is member of required slack channel, edit to channel as needed
            if message["user"] in get_conversation_members(
                    ["C01T006CUAU"]):

                # commands currently added into the bot, will check for command sent via slack          
                if any(item in command.lower() for item in ["help"]):
                    help_message(channel_id, message["user"])

                elif any(item in command.lower() for item in ["empty"]):
                    text = archive_empty_channels()
                    slack_client.chat_postMessage(channel=channel_id, text=text)

                elif any(item in command.lower() for item in ["history"]):
                    text = check_channels_latest_message_history()
                    slack_client.chat_postMessage(channel=channel_id, text=text)

                elif any(item in command.lower() for item in ["warning"]):
                    text = send_channel_owner_warning()
                    slack_client.chat_postMessage(channel=channel_id, text=text)

                elif any(item in command.lower() for item in ["members"]):
                    text = check_number_of_members()
                    slack_client.chat_postMessage(channel=channel_id, text=text)

                elif any(item in command.lower() for item in ["hello", "hello there", "hey", "hej"]):
                    text = (
                            "Hello <@%s>! :tada:"
                            % message["user"]
                    )
                    slack_client.chat_postMessage(channel=channel_id, text=text)

                elif any(item in command.lower() for item in ["guidelines"]):
                    text = check_channels_guidelines()
                    slack_client.chat_postMessage(channel=channel_id, text=text)
                else:
                    help_message(channel_id, message["user"])
            else:
                slack_client.chat_postEphemeral(channel=channel_id, user=message["user"],
                                                text="You are not in the right slack channel")

    thread = Thread(target=send_reply, kwargs={"value": event_data})
    thread.start()
    return Response(status=200)


def help_message(channel_id, user):
    help_text = """
                Welcome to Slack-Bot
                You need to be a member of following slack group: XYZ
                You can use the following commands:
                    - hej: say hello (others might work)
                    - Warning: give users a warning before archiving
                    - help: print out this message
                    - empty: get a list of empty channels
                    - members: prints every channel along with their member count
                    - guidelines: get a list of channels according to guidelines
                    - history: get a list of channels with their latest chat message, archive correct channels"""
    slack_client.chat_postEphemeral(channel=channel_id, user=user, text=help_text)


# Check for channels with 0 members. Skipping all channels that are marked as social. Archive all channels found
def archive_empty_channels():
    result = slack_client.conversations_list(limit=1000, types="public_channel")

    channels_to_archive = []

    for channel in result["channels"]:
        if channel["is_channel"] and not channel["is_archived"]:
            channel_id = channel["id"]
            channel_name = channel["name"]

            if channel_name not in ["general"]:
                time_since_message = get_time_since_last_message_in_channel(channel_id)

                if channel["num_members"] <= 50:
                    if time_since_message.days >= 90:
                        channels_to_archive.append((channel_name, channel_id))

                if channel["num_members"] > 50:
                    if time_since_message.days >= 180:
                        channels_to_archive.append((channel_name, channel_id))

    for _, channel_id in channels_to_archive:
        try:
            slack_client.conversations_join(channel=channel_id)
            slack_client.conversations_archive(channel=channel_id)
        except SlackApiError as e:
            print(e.response)

    return "Archiving the following channels \n" + "\n".join(channel[0] for channel in channels_to_archive)


# check if the channel is adhering to the guidelines. (naming convention, having a description)
def check_channels_guidelines():
    result = slack_client.conversations_list(limit=1000)
    channels = [(channel["name"], channel["id"], channel["creator"]) for channel in result["channels"] if
                channel["is_channel"] and "social" not in channel["purpose"]["value"] and not channel["is_archived"]]

    out = ""

    for name, channel_id, creator in channels:
        if re.match("^(all|announcement|se|fr|uk|no|de|global|ask|bet|proj|fdbk|team|fun)-\w+$", name):
            out += f"{name:25s} :white_check_mark:\n"
        else:
            out += f"{name:25s} :x:\n"
            slack_client.conversations_join(limit=1, channel=channel_id)
            # slack_client.chat_postMessage(channel=channel_id, text="To be added")
            # slack_client.chat_postMessage(channel=creator, text="Hi")

    return out


# check if the channel is adhering to the guidelines. (naming convention, having a description)
def check_channels_latest_message_history():
    result = slack_client.conversations_list(limit=1000)
    channels_history = [(channel_history["name"], channel_history["id"]) for channel_history in result["channels"] if
                        channel_history["is_channel"] and not channel_history["is_archived"]]

    header = f"Channel name, Channel user, Days, last message\n"
    out = "Archived \n" + header + "\n"
    out_not_archived = "Not archived \n" + header + "\n"

    for channel_name, channel_id in channels_history:
        slack_client.conversations_join(limit=1, channel=channel_id)
        history = slack_client.conversations_history(limit=1, channel=channel_id)

        last_message = history['messages'][0]
        username = slack_client.users_info(user=last_message['user'])['user']["real_name"]
        time_since_message = datetime.now() - datetime.fromtimestamp(float(last_message['ts']))

        if time_since_message.days >= 365:
            out += f"{channel_name}, {username}, {time_since_message.days} days, {last_message['text']}\n"
            slack_client.conversations_archive(channel=channel_id)
        else:
            out_not_archived += f"{channel_name}, {username}, {time_since_message.days} days, {last_message['text']}\n"

    return out_not_archived + "\n" + out


# Create a list of all channels along with their membership count.
def check_number_of_members():
    result = slack_client.conversations_list(limit=1000, exclude_archived=1, types="public_channel, private_channel")
    all_channel_names_with_numbers = [(channel["name"], channel["num_members"]) for channel in result["channels"] if
                                      channel["is_channel"]]
    return "\n".join("(%s: %s)" % tup for tup in all_channel_names_with_numbers)


def get_time_since_last_message_in_channel(channel_id):
    slack_client.conversations_join(limit=1, channel=channel_id)
    history = slack_client.conversations_history(limit=1, channel=channel_id)

    last_message = history['messages'][0]
    return datetime.now() - datetime.fromtimestamp(float(last_message['ts']))


def get_conversation_members(conversation_ids):
    members = []
    for conversation_id in conversation_ids:
        try:
            members += slack_client.conversations_members(limit=1000, channel=conversation_id).get("members")
        except SlackApiError:
            pass
    return members


# send warning to owner of channel which meet archive needs
def send_channel_owner_warning():
    result = slack_client.conversations_list(limit=1000, types="public_channel")
    warning_text = """
     Hello @channel-owner! We've noticed that this Slack channel (#{})
     has not been actively used for a few months, and has 
     therefore been scheduled to be archived. Once this has been 
     done, all messages will be deleted and the channel will no 
     longer exist after a few more weeks. If this channel is still in use and 
     should not be deleted, anyone can stop the deletion by posting a new
     message in this channel. If no action is taken soon after
     receiving this message, the channel will be archived.
     If you have any questions regarding the bot or the cleanup,
     or have a request to un-archive you can contact us here:
     """
    out = "Following channels: creators have been warned:\n"
    channels = [(channel["name"], channel["id"], channel["creator"], channel["num_members"]) for channel in
                result["channels"] if channel["is_channel"] and not channel["is_archived"]]

    for channel_name, channel_id, channel_creator, channel_num_members in channels:

        if channel_name not in ["general"]:
            time_since_message = get_time_since_last_message_in_channel(channel_id)

            real_name = slack_client.users_profile_get(user=channel_creator)["profile"]["real_name"] or "Unknown"

            if channel_num_members <= 50:
                if time_since_message.days >= 90:
                    out += f"{channel_name}: {real_name}\n"
                    slack_client.chat_postMessage(channel=channel_creator, text=warning_text.format(channel_name))

            if channel_num_members > 50:
                if time_since_message.days >= 180:
                    out += f"{channel_name}: {real_name}\n"
                    slack_client.chat_postMessage(channel=channel_creator, text=warning_text.format(channel_name))

    return out


if __name__ == "__main__":
    app.run(host='0.0.0.0', port=8086)

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out /  Change )

Google photo

You are commenting using your Google account. Log Out /  Change )

Twitter picture

You are commenting using your Twitter account. Log Out /  Change )

Facebook photo

You are commenting using your Facebook account. Log Out /  Change )

Connecting to %s