Trac-Slack Integration with Tracbot https://sixfeetup.com/blog/trac-slack-integration-with-tracbot

requirements.txt

requests==2.21.0
six==1.12.0
slackeventsapi==2.1.0
slackclient==2.5.0
websocket-client==0.54.0

tracbot.py

import json
import requests

from flask import Flask

from multiprocessing import Process
from slack import WebClient
from slackeventsapi import SlackEventAdapter
from urllib.parse import urlparse


with open('slack-creds.txt') as f:
    BOT_ID = f.readline().strip()
    SLACK_BOT_TOKEN = f.readline().strip()
    SLACK_CLIENT_ID = f.readline().strip()
    SLACK_CLIENT_SECRET = f.readline().strip()
    SLACK_SIGNING_SECRET = f.readline().strip()

# constants
AT_BOT = "<@" + BOT_ID + ">:"

# instantiate Slack client
app = Flask(__name__)
slack_client = WebClient(SLACK_BOT_TOKEN)
slack_events_adapter = SlackEventAdapter(
    SLACK_SIGNING_SECRET,
    "/slack/events/",
    app)


def handle_links(links, channel, ts, thread_ts, text):
    """
        Receives commands directed at the bot and determines if they
        are valid commands. If so, then acts on the commands. If not,
        returns back what it needs for clarification.
    """
    for link in links:
        url = urlparse(link)
        commentno = url.fragment.split(':')[-1]
        path = url.path[1:].split('/')
        if 'ticket' not in path or len(path) != 3:
            # we don't have a ticket link
            continue
        trac_url = url.scheme + "://" + url.netloc + '/' + path[0] + '/rpc'
        trac_ticket = int(path[2])
        r = requests.put(
            trac_url,
            json={'method': 'ticket.get', 'params': [trac_ticket]}
        )
        if r.json().get('error'):
            slack_client.chat_postMessage(
                channel=channel,
                text="Sorry, got a bad ticket #{}".format(trac_ticket),
                as_user=True)
            continue
        prio_colors = {
            'trivial': ['#beedef', ':prio-minor:'],
            'minor': ['#beedef', ':prio-minor:'],
            'major': ['#2fb421', ':prio-major:'],
            'critical': ['#f4ed3b', ':prio-critical:'],
            'blocker': ['#f32520', ':prio-blocker:']
        }
        prio = prio_colors.get(
            r.json()['result'][3]['priority'], ['#2fb421', ':prio-major:'])
        summary = r.json()['result'][3]['summary']
        if 'cat' in summary:
            if r.json()['result'][3]['priority'] == 'blocker':
                summary = summary + " :scream_cat:"
            else:
                summary = summary + " :cat:"
        attachments = [
          {
            "fallback": " :: ".join([
                path[0],
                summary,
                r.json()['result'][3]['status']]),
            "color": prio[0],
            "pretext": "{0} *<{1}|{2}>*".format(
                prio[1], link, summary
            ),
            "text": r.json()['result'][3]['description'],
            "fields": [
                {
                    "title": "Owner",
                    "value": r.json()['result'][3]['owner'],
                    "short": "true"
                },
                {
                    "title": "Status",
                    "value": r.json()['result'][3]['status'],
                    "short": "true"
                }
            ]
          }
        ]
        if commentno and commentno.isdigit():
            # don't show extra fields on comments
            attachments[0].pop('fields', None)
            # show the comment instead of description
            updates = requests.put(
                trac_url,
                json={'method': 'ticket.changeLog', 'params': [trac_ticket]})
            for update in updates.json().get('result'):
                if 'comment' and commentno in update:
                    attachments[0]['text'] = '*' + update[1] + ':* ' + update[4]

        for i, atch in enumerate(attachments):
            # limit text length by line breaks and characters
            lines = atch['text'].split('\n')
            lines.remove('\r')
            ctd = '...' if len(atch['text']) > 400 else ''
            attachments[i]['text'] = ' '.join(lines[:4])[:400] + ctd
        message_args = {
            'channel': channel,
            'text': "",
            'attachments': json.dumps(attachments),
            'as_user': "true"
        }
        if thread_ts:
            message_args['thread_ts'] = thread_ts
        slack_client.chat_postMessage(**message_args)


@slack_events_adapter.on("link_shared")
def link_added(event_data):
    # print(json.dumps(event_data, indent=4))
    data = event_data['event']
    channel = data['channel']
    ts = data['message_ts']
    thread_ts = data.get('thread_ts', 0)
    text = 'hi'
    links = list(l['url'] for l in data['links'])
    # print(f"handle_links({links}, {channel}, {ts}, {text})")
    handle_links(links, channel, ts, thread_ts, text)


# Error events
@slack_events_adapter.on("error")
def error_handler(err):
    print("ERROR: " + str(err))


if __name__ == "__main__":
    app.run(port=5000)