Delete old slack messages at a specific channel.

readme.md

Delete old slack messages at a specific channel.

Delete old slack messages at a specific channel.

Usage

• Runtime: Python 3.x

Run at command line

$ API_TOKEN=xoxb-XXXXXXXXXXXXXXXXXXXXXXXXXXX python3 delete_messages.py Cxxxxxxxxx

Run at AWS Lambda

You can specify multiple channel ids with following JSON data.

{
  "target_ch_ids": [
    "CHANNEL_1",
    "CHANNEL_2",
    "CHANNEL_3"
  ]
}

delete_messages.py

"""Delete old Slack messages at specific channel."""
from datetime import datetime
from time import sleep
import json
import re
import sys
import urllib.parse
import urllib.request
import os

DELETE_URL = "https://slack.com/api/chat.delete"
HISTORY_URL = "https://slack.com/api/conversations.history"
API_TOKEN = os.environ["API_TOKEN"]
TERM_SECONDS = 60 * 60 * 24 * 1

def clean_old_message(channel_id):
    print('Start cleaning message at channel "{}".'.format(channel_id))
    current_ts = int(datetime.now().strftime('%s'))
    messages = get_message_history(channel_id)
    print('{} messages in "{}".'.format(len(messages), channel_id))
    for message in messages:
        message_ts = int(re.sub(r'\.\d+$', '', message['ts']))
        if current_ts - message_ts > TERM_SECONDS:
            delete_message(channel_id, message['ts'])
            # sleep(1.0)


def get_message_history(channel_id, limit=100):
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'Authorization': f"Bearer {API_TOKEN}"
    }

    message_history = []
    cursor = 'init'
    while cursor != '':
        params = {
            'channel': channel_id,
            'limit': str(limit)
        }

        if cursor != '' and cursor != 'init':
            params['cursor'] = cursor

        req_url = '{}?{}'.format(HISTORY_URL, urllib.parse.urlencode(params))
        req = urllib.request.Request(req_url, headers=headers)

        with urllib.request.urlopen(req) as res:
            data = json.loads(res.read().decode("utf-8"))
            # print(data)
            if 'ok' not in data or data['ok'] is not True:
                print('Failed to get message.')
                print(data)
                return message_history
            if 'messages' in data:
                message_history.extend(data['messages'])
                if 'response_metadata' in data and 'next_cursor' in data['response_metadata']:
                    cursor = data['response_metadata']['next_cursor']
                else:
                    cursor = ''

    return message_history


def delete_message(channel_id, message_ts):
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'Authorization': f"Bearer {API_TOKEN}"
    }
    params = {
        'channel': channel_id,
        'ts': message_ts
    }

    req_url = '{}?{}'.format(DELETE_URL, urllib.parse.urlencode(params))
    req = urllib.request.Request(req_url, headers=headers)
    with urllib.request.urlopen(req) as res:
        data = json.loads(res.read().decode("utf-8"))
        print(data)
        if 'ok' not in data or data['ok'] is not True:
            print('Failed to delete message. ts: {}'.format(message_ts))


def lambda_handler(event, context):
    if 'target_ch_ids' not in event:
        print('Target channel id is required.')
        return False

    for target_ch_id in event['target_ch_ids']:
        clean_old_message(target_ch_id)


if __name__ == "__main__":
    args = sys.argv
    if len(args) < 2:
        print("The first parameter for slack channel id is required.")
    else:
        clean_old_message(args[1])