Skip to content

Instantly share code, notes, and snippets.

@Tafkas Tafkas/codebase_cop.py Secret
Last active Dec 11, 2018

Embed
What would you like to do?
#!/usr/bin/env python
# encoding: utf-8
"""
codebase2slack.py
Created by Christian Stade-Schuldt on 2015-10-20.
Copyright (c) 2015 Project-A Ventures. All rights reserved.
"""
from datetime import datetime
import time
import requests
CODEBASE_API_URL = 'https://api3.codebasehq.com'
CODEBASE_USER = 'project-a/codebase-user-7'
CODEBASE_KEY = 'mysecretcodebasekey'
CODEBASE_PROJECT_URL = "https://project-a.codebasehq.com/projects/bi/tickets"
SLACK_CHANNEL = "#my-slack-channel"
SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/my/secret/webhook_url"
SLACK_USERNAME = "codebase cop"
SLACK_ICON_EMOJI = ":cop:"
assignee_slack_mapping = {'codebase-user-63': '@slack.user63',
'codebase-user-7': '@slack.user7'}
def get_overdue_qa_tickets():
qa_tickets = []
for status in ['QA', 'Deployed']:
r = requests.get(f'{CODEBASE_API_URL}/bi/tickets.json?query=status%3A{status}',
auth=(CODEBASE_USER, CODEBASE_KEY))
now = datetime.now()
for ticket in r.json():
if ticket['ticket']['milestone']['name'] == 'Current week/sprint' and \
(now - datetime.strptime(ticket['ticket']['updated_at'], '%Y-%m-%dT%H:%M:%SZ')).days > 7:
qa_tickets.append(ticket)
time.sleep(1)
return qa_tickets
def get_old_tickets():
old_tickets = []
now = datetime.now()
cnt = 1
while True:
r = requests.get(
f'{CODEBASE_API_URL}/bi/tickets.json?query=milestone:"Current week/sprint"&page={cnt}',
auth=(CODEBASE_USER, CODEBASE_KEY))
if r.status_code == 200:
for ticket in r.json():
if ticket['ticket']['status']['name'] != 'Done' and \
(now - datetime.strptime(ticket['ticket']['updated_at'],
'%Y-%m-%dT%H:%M:%SZ')).days > 60:
old_tickets.append(ticket)
cnt += 1
time.sleep(1)
else:
break
return old_tickets
def notify_assignees(tickets):
now = datetime.now()
with open('/var/log/codebase_cop.log', 'a') as w:
for t in tickets:
ticket_id = t['ticket']['ticket_id']
ticket_summary = t['ticket']['summary']
assignee = t['ticket']['reporter']
status = t['ticket']['status']['name']
if t['ticket']['assignee'] is not None:
assignee = t['ticket']['assignee']
ticket_assignee = assignee_slack_mapping[assignee]
ticket_overdue_time = (now - datetime.strptime(t['ticket']['updated_at'], '%Y-%m-%dT%H:%M:%SZ')).days
fallback_txt, pretext, text = "", "", ""
if status == 'QA':
fallback_txt = f"Please QA ticket #{ticket_id}: {ticket_summary} - {CODEBASE_PROJECT_URL}/{ticket_id}"
pretext = "QA is overdue"
text = f"Hey, {ticket_assignee}: The QA status for this ticket has not been updated for over {ticket_overdue_time} days - {CODEBASE_PROJECT_URL}/{ticket_id}"
if status == 'Deployed':
fallback_txt = f"Please mark ticket #{ticket_id} as Done: {ticket_summary} - {CODEBASE_PROJECT_URL}/{ticket_id}"
pretext = "Ticket is done"
text = f"Hey, {ticket_assignee}: The ticket has been deployed {ticket_overdue_time} days ago. Please review it - {CODEBASE_PROJECT_URL}/tickets/{ticket_id}"
if ticket_overdue_time > 60:
fallback_txt = f"Please review old Ticket #{ticket_id}: {ticket_summary} - {CODEBASE_PROJECT_URL}/{ticket_id}"
pretext = "Review old ticket"
text = f"Hey, {ticket_assignee}: The ticket has not been updated for over {ticket_overdue_time} days, please check if it is still valid - {CODEBASE_PROJECT_URL}/{ticket_id}"
payload = {"channel": SLACK_CHANNEL, "username": SLACK_USERNAME, "icon_emoji": SLACK_ICON_EMOJI,
"attachments": [{
"fallback": fallback_txt,
"pretext": pretext,
"title": f"Ticket #{ticket_id}: {ticket_summary}" % (ticket_id, ticket_summary),
"title_link": f"{CODEBASE_PROJECT_URL}/{ticket_id}",
"text": text,
"color": "danger"}]
}
w.write(
f'{now.isoformat()}, {ticket_id}, {ticket_summary}, {assignee}, {status}, {ticket_assignee}, {ticket_overdue_time}, {status}\n')
# notify channel every Monday
if now.weekday() == 0:
r = requests.post(SLACK_WEBHOOK_URL, json=payload)
payload['channel'] = ticket_assignee
# notify assignee directly
r = requests.post(SLACK_WEBHOOK_URL, json=payload)
time.sleep(1)
def main():
qa_tickets = get_overdue_qa_tickets()
notify_assignees(qa_tickets)
old_tickets = get_old_tickets()
notify_assignees(old_tickets)
if __name__ == '__main__':
main()
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.