archive_slack.py

import requests
from creds import LEGACY_TOKEN
from datetime import datetime
from exceptions import (InvalidAuthToken,
                        APIRequestError)

channel_history_url = "https://slack.com/api/channels.history"
channel_list_url = 'https://slack.com/api/channels.list'
channel_archive_url = 'https://slack.com/api/channels.archive'
post_message_url = 'https://slack.com/api/chat.postMessage'


def channels(api_token, now):
    payload = {'exclude_archived': True, 'token': api_token}

    r = requests.post(channel_list_url, data=payload)

    channels = _process_response(response=r)
    for channel in channels['channels']:
        print(channel)
        # process_channel(channel, api_token, now)


def process_channel(channel, api_token, now):
    """
    {'id': 'C28587C3T', 'name': 'github-universe', 'is_channel': True, 'created': 1473016778, 'is_archived': False, 'is_general': False, 'unlinked': 0, 'creator': 'U0S1GKKFH', 'name_normalized': 'github-universe', 'is_shared': False, 'is_org_shared': False, 'is_member': False, 'is_private': False, 'is_mpim': False, 'members': [], 'topic': {'value': '', 'creator': '', 'last_set': 0}, 'purpose': {'value': '', 'creator': '', 'last_set': 0}, 'previous_names': [], 'num_members': 25}
    :param channel:
    :type channel:
    :return:
    :rtype:
    """

    payload = {'channel': channel['id'], 'token': api_token}

    r = requests.post(channel_history_url, data=payload)

    history = _process_response(response=r)

    expired, newest_message = is_channel_expired(history, now, channel["name"])
    if expired:
        print(f'channel is expired: {channel["name"]};')
        print(f'---> newest message: {newest_message}')
        print(f'---> newest date: {datetime.fromtimestamp(float(newest_message["ts"]))}')

        with open('stale_channels.txt', 'a') as f:
            f.write(f'channel is expired: {channel["name"]};')
            f.write(f'---> newest message: {newest_message}')
            f.write(f'---> newest date: {datetime.fromtimestamp(float(newest_message["ts"]))}')


def is_channel_expired(channel_history, now_timestamp, name):
    min_date = float('inf')
    min_message = None
    for item in channel_history['messages']:
        dt = now_timestamp - datetime.fromtimestamp(float(item['ts']))
        if dt.days < min_date and item.get('subtype', 'message') != 'channel_leave':
            min_message = item
            min_date = dt.days

    stale = min_date > 30
    if not stale:
        with open('not_stale.txt', 'a') as f:
            f.write(f'channel is expired: {name};')
            f.write(f'---> days ago: {min_date}')
            f.write(f'---> newest message: {min_message}')

    return min_date > 30, min_message


def message_archive(channel, api_token):
    text = "> Hello, very quiet channel. We tend the channel garden by archiving channels that have been quiet for 30 straight days and this channel is one of them.\n\n> Archiving a channel removes it from the list of active channels and prevents new comments from being made there. That makes it easier for newer folks to find the right place to get their questions, answers and comments heard without wading through extraneous channels or worse, posting a comment only to hear nothing in response.\n> All existing comments in the channel are retained for easy browsing and searching and can be read like any other channel. If need for this particular channel arises again, it can be unarchived by any user here.\n\n> If you'd like to keep this channel active, SAY ANYTHING and we'll reset the timer. It's that simple. If we don't hear from anyone in the next 1 day, this channel will be archived. Please direct all concerns to #community-feedback, this account's messages are not monitored"
    some_icon_url = ''
    payload = {'token': api_token, 'channel': channel, 'text': text, 'as_user': False,
               'link_names': 1,
               'icon_url': some_icon_url, 'username': 'hollomancer'}

    r = requests.post(post_message_url, data=payload)

    print(r)


def archive_channel(channel, api_token):
    payload = {'token': api_token, 'channel': channel}

    r = requests.post(channel_archive_url, data=payload)
    print(r)
    response = _process_response(response=r)
    print(response)


def _process_response(response):
    response_data = response.json()

    if not response.status_code == requests.codes.ok:
        raise APIRequestError('api_error')

    if not response_data.get('ok'):
        try:
            _check_error(response_data['error'])
        except Exception as e:
            print(e)

    return response_data


def _check_error(error):
    if error == 'invalid_auth':
        raise InvalidAuthToken(error)


if __name__ == '__main__':
    OC_TEAM = ''

    now = datetime.now()
    # channels(LEGACY_TOKEN, now)
    channel = ''
    message_archive(channel, LEGACY_TOKEN)
    # archive_channel(channel, LEGACY_TOKEN)

exceptions.py

class AlreadyInTeam(Exception):
    pass


class InvalidInviteeEmail(Exception):
    pass


class InvalidAuthToken(Exception):
    pass


class AlreadyInvited(Exception):
    pass


class APIRequestError(Exception):
    pass