slack-tail

#!/usr/bin/env python3
#
# How to get a slack API token
#
# From https://github.com/yuya373/emacs-slack/tree/1f6a40faec0d8d9c9de51c444508d05a3e995ccd#how-to-get-token
# - Using chrome or firefox, open and sign into the slack customization page, e.g. https://my.slack.com/customize
# - Right click anywhere on the page and choose "inspect" from the context menu. This will open the developer tools.
# - Find the console (it's one of the tabs in the developer tools window)
# - At the prompt ("> ") type the following: window.prompt("your api token is: ", TS.boot_data.api_token)
# - Copy the displayed token elsewhere, and close the window.

from ansi.colour import *
import ansi.cursor
import ansi.osc
import ansi.iterm
import argparse
import atexit
import emoji
import fcntl
import html
import os
from pprint import pprint
import re
import requests
import struct
import sys
import termios
import textwrap
import traceback
import unicodedata

is_iterm = os.environ['TERM_PROGRAM'] == "iTerm.app"

width=80
try:
    height, width = struct.unpack('hh', fcntl.ioctl(sys.stdin, termios.TIOCGWINSZ, '1234'))
except:
    pass

from slack_sdk.rtm import RTMClient

parser = argparse.ArgumentParser(description="Send messages to slack channels")
parser.add_argument('--until', metavar='REGEX')
parser.add_argument("channel", metavar="CHANNEL", nargs='?')
parser.add_argument("users", metavar="USER", nargs='*')
parser.add_argument("--debug", default=False, action='store_true')
parser.add_argument("--send", metavar="MESSAGE", default="")
args = parser.parse_args()

if args.channel and args.channel.startswith('#'):
    args.channel = args.channel[1:]

@RTMClient.run_on(event="hello")
def send_message(rtm_client, web_client, **payload):
    if args.send and args.channel:
        msg = args.send
        try:
            response = web_client.chat_postMessage(channel=args.channel, text=msg)
        except SlackApiError as e:
            print("Got an error: " + e.response['error'], file=sys.stderr)

@RTMClient.run_on(event="message")
def show_message(rtm_client, web_client, **payload):
    try:
        show_message_(rtm_client, web_client, **payload)
    except Exception as e:
        pprint(payload, width=width)
        traceback.print_exc()

def show_message_(rtm_client, web_client, **payload):
    data = payload['data']
    if data.get('subtype') in ('message_changed', 'message_replied', 'message_deleted'):
        return
    team = data.get('team', data.get('user_team', data.get('source_team', "")))
    user = bot = username = display_name = None
    channel = get_channel(web_client, data['channel'])
    link = 'slack://channel?team=%s&id=%s' % (team, data['channel'])
    avatars = {}
    if 'bot_id' in data:
        bot = get_bot(web_client, data['bot_id'])
        if bot:
            avatars = bot['icons']
    if 'bot_profile' in data and 'name' in data['bot_profile']:
        username = data['bot_profile']['name']
        avatars = data['bot_profile']['icons']
    if 'user' in data:
        user = get_user(web_client, data['user'])
        if 'profile' in user and 'display_name' in user['profile']:
            display_name = user['profile']['display_name']
        if 'real_name' in user:
            username = user['real_name']
        elif 'profile' in user and 'real_name' in user['profile']:
            username = user['profile']['real_name']
        avatars = user['profile']
    if 'username' in data:
        username = data['username']
    if 'icons' in data:
        avatars = data['icons']
    if channel['is_im'] or channel['is_mpim']:
        channel = ""
    else:
        channel = channel['name']

    if args.channel and args.channel != channel:
        return

    if username is None:
        print("Username not known!")
        pprint(payload, width=width)
    avatar = None
    if avatars:
        smallest = 100000
        for k in sorted(avatars.keys()):
            if k == 'emoji':
                avatar = find_emoji(avatars[k][1:-1], custom_emoji=web_client.emoji)
            if k.startswith('image_'):
                size = k[6:]
                if size.isdigit() and int(size) < smallest:
                    smallest = int(size)
                    avatar = avatars[k]

    if args.users and (username not in args.users) and (display_name not in args.users):
        return

    if data.get('text', ''):
        print_(web_client, channel, username, data['text'], args.until, link, avatar)
    for a in data.get('attachments', []) + data.get('blocks', []):
        if a.get('title', ''):
            print_(web_client, channel, username, a['title'] + ' ' + a.get('title_link', ''), args.until, link, avatar)
        if a.get('text', '') and isinstance(a['text'], str):
            print_(web_client, channel, username, a['text'], args.until, link, avatar)
        if a.get('image_url', ''):
            msg = a['fallback'][1:-1]
            imgt = None
            if a['image_bytes'] < 1024 * 1024:
                img = requests.get(a['image_url'])
                if img.ok:
                    msg = ''
                    imgt = ansi.iterm.image(img.content)
            print_(web_client, channel, username, msg, args.until, link, avatar)
            if imgt:
                print(imgt)
    for a in data.get('files', []):
        msg = "%s: %s" % (a['title'], a['url_private'])
        imgt = None
        if a['mimetype'].startswith('image/') and a['size'] < 1024 * 1024:
            img = requests.get(a['url_private'], headers={'Authorization': 'Bearer %s' % web_client.token})
            if img.ok:
                msg = ''
                imgt = ansi.iterm.image(img.content)
        print_(web_client, channel, username, msg, args.until, link, avatar)
        if imgt:
            print(imgt)
    if args.debug:
        pprint(payload, width=width)

def print_(w, c, u, m, r, l, a):
    ou = u
    if len(c) > 20:
        c = c[:17] + '...'
    if len(u) > 20:
        u = u[:17] + '...'

    c = "%-20s" % c
    u = "%-20s" % u

    if is_iterm:
        c = ansi.osc.anchor(l) + c + ansi.osc.anchor("")

    c = fx.faint(c)
    u = fx.bright(u)

    indent = 42
    if is_iterm:
        u = get_avatar(ou, a) + ' ' + u
        indent += 3

    links = []
    m = re.sub('<@([^>]+)>', lambda m_: replace_user(w, m_), m)
    m = re.sub('<#([^>]+?)(?:\|([^>]+))?>', lambda m_: replace_channel(w, m_), m)
    m = re.sub('<(https?://[^| >]+)\|([^>]+)>', lambda m_: replace_url(m_, links), m)
    m = re.sub('<(https?://[^| >]+)>', r'\1', m)
    m = re.sub(':([^ :]+):', lambda m_: find_emoji(m_.group(1)), m)
    m = html.unescape(m)
    frags = []
    for line in m.strip().splitlines():
        line = line.strip()
        if not line:
            continue
        frags += textwrap.wrap(line, width-indent, break_on_hyphens=False)
    if is_iterm:
        frags = [re.sub(':([^ :]+):', lambda m_: find_emoji(m_.group(1), custom_emoji=w.emoji), m) for m in frags]
    frags += [fx.faint("[%d] %s") % (pos+1, link) for (pos, link) in enumerate(links)]
    if not frags:
        frags = ['']

    print("\n%s %s %s" % (c, u, frags[0]),flush=True, end="")
    for frag in frags[1:]:
        print("\n" + " " * indent + frag, flush=True, end="")

    if r and re.search(r, m):
        sys.exit(0)

def replace_user(w, m):
    try:
        return '@' + get_user(w, m.group(1))['name']
    except:
        return m.group(0)

def replace_channel(w, m):
    if m.group(2):
        return '#' + m.group(2)
    try:
        return '@' + get_channel(w, m.group(1))['name']
    except:
        return m.group(0)

def replace_url(m, links):
    link = html.unescape(m.group(1))
    try:
        pos = links.index(link)+1
    except ValueError:
        pos = len(links)+1
        links.append(link)
    return "%s[%d]" % (m.group(2), pos)

modifiers = {
    'skin-tone-1': '\U0001F3FB',
    'skin-tone-2': '\U0001F3FC',
    'skin-tone-3': '\U0001F3FD',
    'skin-tone-4': '\U0001F3FE',
    'skin-tone-5': '\U0001F3FF',
}

def find_emoji(name, custom_emoji={}, fallback=None):
    if not fallback:
        fallback = ':' + name + ':'
    if name in modifiers:
        return modifiers[name]
    if name in custom_emoji:
        val = custom_emoji[name]
        if val.startswith('alias:'):
            return find_emoji(val[6:], custom_emoji, fallback)
        if val.startswith('https://'):
            img = requests.get(val)
            if not img.ok:
                return fallback
            val = custom_emoji[name] = ansi.iterm.image(img.content, name=fallback, height=1)
        return val
    text = ':' + name + ':'
    val = emoji.emojize(text, language='alias')
    if val != text:
        return val
    try:
        return unicodedata.lookup(name)
    except KeyError:
        try:
            return unicodedata.lookup(name.replace('_', ' '))
        except KeyError:
            return fallback

channels = {}
def get_channel(web_client, channel):
    if channel not in channels:
        channels[channel] = web_client.conversations_info(channel=channel).data['channel']
    return channels[channel]

users = {}
def get_user(web_client, user):
    if user not in users:
        users[user] = web_client.users_info(user=user).data['user']
    return users[user]

avatars = {}
def get_avatar(name, url):
    if url not in avatars:
        if url.startswith('https://'):
            img = requests.get(url)
            if not img.ok:
                return None
            avatars[url] = ansi.iterm.image(img.content, name=name, height=1)
        else:
            avatars[url] = url
    return avatars[url]

bots = {}
def get_bot(web_client, bot):
    if bot not in bots:
        try:
            bots[bot] = web_client.bots_info(bot=bot).data['bot']
        except:
            bots[bot] = None
    return bots[bot]

if is_iterm:
    print(ansi.iterm.badge("Slack"), end='', flush=True)
    atexit.register(lambda: print(ansi.iterm.badge(""), end='', flush=True))
atexit.register(lambda: print(ansi.cursor.show()))

print("Waiting for slack messages..." + ansi.cursor.hide(), end="", flush=True)
slack_token = os.environ["SLACK_TOKEN"]
rtm_client = RTMClient(token=slack_token)
rtm_client._web_client.emoji = rtm_client._web_client.emoji_list()['emoji']
rtm_client.start()