Delete old slack messages at a specific channel.
$ API_TOKEN=xoxb-XXXXXXXXXXXXXXXXXXXXXXXXXXX python3 delete_messages.py CxxxxxxxxxYou can specify multiple channel ids with following JSON data.
{
"target_ch_ids": [
"CHANNEL_1",
"CHANNEL_2",
"CHANNEL_3"
]
}| """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]) |