|
"""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 |
|
|
|
DELETE_URL = "https://slack.com/api/chat.delete" |
|
HISTORY_URL = "https://slack.com/api/channels.history" |
|
API_TOKEN = '******' |
|
TERM = 60 * 60 * 24 * 7 # 1 week |
|
|
|
|
|
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: |
|
delete_message(channel_id, message['ts']) |
|
sleep(1) |
|
|
|
|
|
def get_message_history(channel_id, limit=100): |
|
headers = { |
|
'Content-Type': 'application/x-www-form-urlencoded' |
|
} |
|
|
|
message_history = [] |
|
cursor = 'init' |
|
while cursor != '': |
|
params = { |
|
'token': API_TOKEN, |
|
'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' |
|
} |
|
params = { |
|
'token': API_TOKEN, |
|
'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]) |
Thanks for the nice script !
I just noticed that the equation at line 21 is inversed.