Export Slack messages from your Slack workspace

slack_export.py

"""
Author: Gabriele Marini @Gabryxx7

How to run it:
1. Create a Slack App (https://api.slack.com/start/overview#creating)
2. Give it all the permissions you need (usually all the read ones except for the admin persmissions)
3. Create a new config.yaml file that looks like this:

    SLACK_BOT_TOKEN: "xoxp-123abvcsafas-xxxxx-xxxxxx-..."
    timeframe:
      from: 2021-01-01
      to: 2021-02-02
    public_channels_file: public_channels.csv
    private_channels_file: private_channels.csv
    group_messages_file: group_messages.csv
    logger_name: slack_bot_log.txt

4. Get the conversation you want with:
    convos = get_conversations(types="public_channel,private_channel,im,mpim")
5. Export the data to csb with:
    export_all_conversations_history(convos)
"""

import os
# Import WebClient from Python SDK (github.com/slackapi/python-slack-sdk)
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
import yaml
import logging
import utils
import csv
import os

def get_conversations(**kwargs):
    try:
        convos = []
        # Call the conversations.list method using the WebClient
        result = client.conversations_list(types="public_channel,private_channel,mpim,im")
        for response in result:
            convos = convos + (response['channels'])
        return convos
    except SlackApiError as e:
        print(f"Error in retreiving conversations: {e}")

"""
 channel_id: ID of the channel you want to send the message to
"""
def get_conversation_history(conversation, **kwargs):
    channel_id = conversation["id"]
    conversation_history = []
    try:
        # Call the conversations.history method using the WebClient
        # conversations.history returns the first 100 messages by default
        # These results are paginated, see: https://api.slack.com/methods/conversations.history$pagination
        result = client.conversations_history(channel=channel_id, **kwargs)
        conversation_history = result["messages"]
        # Print results
        logger.info("{} messages found in {}".format(len(conversation_history), id))
        return conversation_history
    except SlackApiError as e:
        logger.error("Error creating conversation: {}".format(e))

def get_conversation_by_id(channel_id, conversations):
    res = next(filter(lambda x: 'id' in x and channel_id == x['id'], conversations), None)
    return res

def get_conversation_by_name(channel_name, conversations):
    res = next(filter(lambda x: 'name' in x and channel_name == x['name'], conversations), None)
    return res

def get_user_info(user_id):
    try:
        result = client.users_info(user=users_list)
        print(result)
    except SlackApiError as e:
        logger.error("Error creating conversation: {}".format(e))

# Put users into the dict
def users_to_dict(users_array):
    for user in users_array:
        # Key user info on their unique user ID
        user_id = user["id"]
        # Store the entire user object (you may not need all of the info)
        users_list[user_id] = user["name"]

def get_users_list(force_overwrite=False):
    global users_list
    users_list = {}
    csv_users_file = "slack_users_list.csv"
    try:
        with open(csv_users_file, mode='r', newline='') as infile:
            reader = csv.reader(infile)
            for rows in reader:
                users_list[rows[0]] = rows[1]
    except IOError as e:
        print(f"Error reading users file: {e}")
    if not users_list or force_overwrite:
        print("Downloading user dictionary")
        users_list = {}
        try:
            result = client.users_list()
            users_to_dict(result["members"])
            try:
                with open(csv_users_file, 'w', newline='') as csvfile:
                    csvwriter = csv.writer(csvfile, delimiter=',')
                    csvwriter.writerow(["user_id", "user_name"])
                    for key, value in users_list.items():
                        csvwriter.writerow([key, value])
            except IOError:
                print(f"Error writing users file: {e}")
        except SlackApiError as e:
            logger.error("Error creating conversation: {}".format(e))
    return users_list

def print_conversations_list(convos):
    for conversation in convos:
        # print(conversation)
        try:
            name = utils.get_conversation_name(conversation, users_list=users_list)
            type = utils.get_conversation_type_string(conversation)
            print(f"{name}\t{type}")
        except KeyError as e:
            print(f"KeyError in retreiving key: {e}")

def write_messages(messages, csvwriter, prefix=None):
    prefix = prefix if prefix != None else []
    for message in messages:
        write_message(message, csvwriter, prefix)

def write_message(message, csvwriter, prefix=None):
    prefix = prefix if prefix != None else []
    try:
        row = []
        # row.append(message["message_id"])
        row.append(message["text"])
        try:
            row.append(message["user"])
        except Exception as e:
            row.append("None")
        try:
            row.append(users_list[message["user"]])
        except:
            row.append("None")
        row.append(message["ts"])
        row.append(utils.ts_to_dt(message["ts"]))
        # print(f"{utils.ts_to_dt(message['ts'])}\t{message['client_msg_id']}\t{message['type']}\t{message['user']}\t{message['team']}\t{message['text']}")
        csvwriter.writerow(prefix+row)
    except Exception as e:
        print(f"error while writing to file: {e}")

def export_conversation_history(conversation, csv_writer=None, messages=None, **kwargs):
    convo_name = utils.get_conversation_name(conversation, users_list=users_list)
    filename = f"slack_{convo_name}_export_{utils.formatted_now()}.csv"
    if messages == None:
        messages = get_conversation_history(conversation)
    csvwriter = csv_writer
    if csv_writer is None:
        f = open(filename, 'w', newline='', encoding='utf-8')
        csvwriter = init_csv_writer(f)

    convo_info_prefix = [conversation["id"], convo_name, utils.get_conversation_type_string(conversation)]
    write_messages(messages, csvwriter=csvwriter, prefix=convo_info_prefix)
    if csv_writer == None:  # if we created a new writer inside here then we need to close the file at the end
        print(f"------- EXPORT {filename} COMPLETED AT {utils.formatted_now()} -------")
        f.close()

def export_all_conversations_history(conversations, **kwargs):
    filename = f"slack_export_{utils.formatted_now()}.csv"
    f = open(filename, 'w', newline='', encoding='utf-8')
    csvwriter = init_csv_writer(f)
    for conversation in conversations:
        export_conversation_history(conversation, csv_writer=csvwriter, messages=None, **kwargs)
    print(f"------- EXPORT {filename} COMPLETED AT {utils.formatted_now()} -------")
    f.close()

def init_csv_writer(f):
    csvwriter = csv.writer(f, delimiter=',', quoting=csv.QUOTE_ALL)
    csvwriter.writerow(["convo_id", "convo_name", "convo_type", "msg_text", "msg_user_id", "msg_user_name", "msg_timestamp", "msg_datetime"])     # header
    return csvwriter

def init():
    # WebClient instantiates a client that can call API methods
    # When using Bolt, you can use either `app.client` or the `client` passed to listeners.
    # The config data is stored in config.yaml

    global config
    config = yaml.safe_load(open("config.yaml"))
    global client
    client = WebClient(token=config["SLACK_BOT_TOKEN"])
    global users_list
    get_users_list()
    global logger
    logger = logging.getLogger(config["logger_name"])

def main():
    init()
    convos = get_conversations(types="public_channel,private_channel,im,mpim")
    print_conversations_list(convos)
    channnel = get_conversation_by_name("channel_name", convos)
    channel_messages = get_conversation_history(channnel,
        inclusive=True,
        oldest=utils.dt_to_ts("2018-01-01 01:00:00"),
        limit=1000000)
    # print(len(channel_messages))
    # print(channel_messages[0].keys())
    # export_conversation_history(channnel, messages=channel_messages)
    export_all_conversations_history(convos)


if __name__ == "__main__":
    main()