Skip to content

Instantly share code, notes, and snippets.

@cdw9 cdw9/requirements.txt
Last active Mar 4, 2020

Embed
What would you like to do?
requests==2.21.0
six==1.12.0
slackeventsapi==2.1.0
slackclient==2.5.0
websocket-client==0.54.0
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.