{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Emailed Digest of Clubhouse.IO story statuses\n", "**This is provided as a base; it is suspected that you'll need to alter for your needs.**\n", "\n", "Provided without warranty of any kind and released under MIT License by Junction Applications.\n", "\n", "[Clubhouse.IO]((https://clubhouse.io/)) is in their words \n", ">\"Where software teams do their best work. Clubhouse is the first project management platform for software development that brings everyone on every team together to build better products.\" \n", "\n", "I've come to depend on it for a number of projects, and general task managmement. I find it a joy to use and easy to integrate with; as such providing this Notebook as a short demo of how one might use it to send status updates for particular projects to interested parties. \n", "\n", "This notebook outlines how to connect to your [Clubhouse.IO](https://clubhouse.io/) organization using their [API](https://clubhouse.io/api/rest/v3/), run a query to get the stories needed, build an email and send it. You'll probably want to either extract out the bits needed and run as a proper scheduled script, or by using something like [PaperMill](https://github.com/nteract/papermill) to parameterize and schedule.\n", "\n", "This is a lengthy Notebook as it contains all the Clubhouse connection code which would normally be in it's own file and imported. There is also a function to send email via smtp here which may not suit your needs; you'll need to supply your own mail server if you are using it.\n", "\n", "## Requirements\n", "You may require some libraries to be installed to make this work. Some known ones:\n", "\n", "- [requests](https://pypi.org/project/requests/)\n", "- [parse](https://pypi.org/project/parse/)\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Configurable setttings" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "report_to_run = 'recent' # see the report_params dictionary id's in the next section for options\n", "default_send_to_address = \"your.email@yourcompany.com\"\n", "smtp_mail_server = \"smtp.yourmailserver.law\"\n", "# Enter your own token below\n", "# https://help.clubhouse.io/hc/en-us/articles/205701199-Clubhouse-API-Tokens\n", "clubhouse_api_token = \"12345678-9012-3456-7890-123456789012\" # os.getenv('CLUBHOUSE_TOKEN')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Report parameters\n", "This is where you'd set up any reports you want to run. Create new dictionary items as necessary in the report_params dictionary. \n", "\n", "The idea here is that we have a number of people who may want a status update. The parameters are stored in a dictionary, with ability to add a new report as time goes on. If this got out of hand, I'd throw all of this in a database somewhere, but for illustration purposes, it is stored here.\n", "\n", "`reporting_states` is a tuple of story workflow states you want to report on, in the order you want them to appear in the email. This allows you to restrict something like \"Backlog\" from appearing in certain reports. You'll need to alter this in the default_params to suit your organization." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# let's manage some imports and set a date format Clubhouse likes\n", "from datetime import datetime, timezone, timedelta\n", "now = datetime.now(timezone.utc)\n", "ch_date_format = '%Y-%m-%d'\n", "\n", "# complete_days is the number of days to go back to include in the \"Complete\" status\n", "# clubhouse_query is any valid query string outlined here:\n", "# https://help.clubhouse.io/hc/en-us/articles/360018875792-Searching-in-Clubhouse-Text-Strings\n", "report_params = {\"hold\": {\"clubhouse_query\":\"label:onhold\",\n", " \"complete_days\": 10,\n", " \"email_subject\": \"Items on Hold\"},\n", " # recent in this case is anything updated in the last 14 days\n", " \"recent\": {\"clubhouse_query\":f\"updated:{now - timedelta(days=14):{ch_date_format}}..*\",\n", " \"complete_days\": 10,\n", " \"email_subject\": \"Work Status with Recent Updates\"},\n", " # add additional reports here. \n", " \n", " }\n", "\n", "default_params = {\"send_to_addrs\": [default_send_to_address, ],\n", " \"reporting_states\": ('Backlog', 'Specify/Breakdown', 'Implementing', 'Implemented', 'Validating', 'Completed'),\n", " \"complete_days\": 10,\n", " \"send_from_addr\": '\"Change this Friendly Name and Email\" ',\n", " \"email_subject\": \"Clubhouse Story Status Digest\" \n", " }\n", "\n", "# shallow merge the two dictionaries (Python 3.5+)\n", "# https://www.python.org/dev/peps/pep-0448/\n", "run_params = {**default_params, **report_params[report_to_run]} " ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Create a little mailer function \n", "Suggested this goes in its own file, and imported, but we're trying to build this notebook with everything needed. Use whatever method you like for sending mail, this one uses an smtp server that requires no authentication." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# following has been conflated from several StackOverflow posts. Apologies to original authors \n", "# for lack of credit, but this has been evolving for a few years. Use whatever you need to send\n", "# an email (sendgrid or other for example), this is if you have a smtp server available.\n", "import smtplib\n", "\n", "from email.mime.application import MIMEApplication\n", "from email.mime.image import MIMEImage\n", "from email.mime.multipart import MIMEMultipart\n", "from email.mime.text import MIMEText\n", "from email.utils import COMMASPACE, formatdate\n", "\n", "\n", "def send_mail(send_from, \n", " send_to, \n", " subject, \n", " plain_text, \n", " html_text, \n", " server=smtp_mail_server):\n", " assert isinstance(send_to, list)\n", "\n", " msg = MIMEMultipart('related')\n", " msg['From'] = send_from\n", " msg['To'] = COMMASPACE.join(send_to)\n", " msg['Date'] = formatdate(localtime=True)\n", " msg['Subject'] = subject\n", " msg.preamble = 'This is a multi-part message in MIME format.'\n", " \n", " # alternative will contain plain text and html\n", " msg_alt = MIMEMultipart('alternative')\n", " msg.attach(msg_alt)\n", " \n", " msg_alt.attach(MIMEText(plain_text, 'plain'))\n", " msg_alt.attach(MIMEText(html_text, 'html', 'utf-8'))\n", "\n", " smtp = smtplib.SMTP(server)\n", " smtp.sendmail(send_from, send_to, msg.as_string())\n", " smtp.close()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Clubhouse API Hooks \n", "This would also most likely be imported normally as a class but reproduced here and stripped down to the essenstials for this demo. It provides a number of Python wrappers to the api calls." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Some Clubhouse.IO tools:\n", "import os\n", "import sys\n", "import time\n", "import requests\n", "\n", "# CH_TOKEN = os.getenv('CLUBHOUSE_TOKEN') # the normal way I'd store the token\n", "CH_TOKEN = clubhouse_api_token\n", "CH_BASE = 'https://api.clubhouse.io/'\n", "CH_API_VER = 'api/v3/'\n", "\n", "\n", "def url(entity, ver=None):\n", " \"\"\"\n", " Returns a properly formatted url given module's constants\n", " entity is one of the Clubhouse entities (epics, labels, stories etc.)\n", " :param entity: the clubhouse entity (can be combined like search/stories)\n", " :param ver: allows override of the api version\n", " \"\"\"\n", " if ver:\n", " ch_ver = ver\n", " else:\n", " ch_ver = CH_API_VER\n", " return f'{CH_BASE}{ch_ver}{entity}?token={CH_TOKEN}'\n", "\n", "\n", "def entity_list(entity):\n", " \"\"\"\n", " :return: a JSON packet listing of entities from Clubhouse\n", " \"\"\"\n", " response = requests.get(url(entity))\n", " return response.json()\n", "\n", "\n", "def project_list():\n", " \"\"\"\n", " :return: a JSON packet of projects\n", " \"\"\"\n", " return entity_list(entity='projects')\n", "\n", "\n", "def epic_list():\n", " \"\"\"\n", " :return: a JSON packet of epics\n", " \"\"\"\n", " return entity_list(entity='epics')\n", "\n", "\n", "def workflow_list():\n", " \"\"\"\n", " :return: a JSON packet of workflow states\n", " \"\"\"\n", " return entity_list(entity='workflows')\n", "\n", "\n", "def search_stories(query, page_size=25):\n", " # query in the form of a proper Clubhouse query string defined:\n", " # https://help.clubhouse.io/hc/en-us/articles/360000046646-Search-Operators\n", " body_params = {'page_size': page_size, 'query': query}\n", " entity = 'search/stories'\n", "\n", " data = dict()\n", "\n", " first_page = requests.get(url(entity=entity), body_params).json()\n", " stories = first_page.get('data', None)\n", " next_page_token = first_page.get('next', None)\n", " total_stories = first_page.get('total', None)\n", " while_count = 0\n", "\n", " while next_page_token:\n", " while_count += 1\n", " # if we're calling more than this we've probably made a mistake\n", " # picking arbitrary 1000 limit hopefully keeping us under the Clubhouse 200/min rate\n", " if while_count * page_size > 1000:\n", " data.update({'warning': 'Excessive api calls resulted in truncated data set. '\n", " f'About {page_size*while_count} of {total_stories} returned'})\n", " raise ValueError(data)\n", " break\n", " # so we strip off that last / because the next page token 'next' url starts with a /\n", " np_url = f'{CH_BASE[:-1]}{next_page_token}&token={CH_TOKEN}'\n", " next_page = requests.get(np_url).json()\n", " stories.extend(next_page['data'])\n", " next_page_token = next_page.get('next', None)\n", "\n", " data.update({'data': stories, 'total': total_stories})\n", " return data\n", "\n", "\n", "def get_story(id):\n", " entity = 'stories/{id}'.format(id=id)\n", " response = requests.get(url(entity))\n", " return response.json()\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## The email generator\n", "These next bits manage getting all the pieces, creating the email and sending it.\n", "\n", "The cells below are of no particular split points. This code was converted from another script and was pasted in in bitesized chunks simply to make testing functionality a bit easier in Jupyter Notebook." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import dateutil.parser\n", "from parse import search\n", "\n", "workflow_states = dict()\n", "for wf in workflow_list()[0][\"states\"]:\n", " workflow_states[wf[\"id\"]] = wf\n", " \n", "# set colours for story types (couldn't find a way to get these other than to hardwire like this)\n", "story_types = {'feature': {'colour': '#F7F4E8',},\n", " 'chore': {'colour': '#F1F6FE',},\n", " 'bug': {'colour': '#FCE8E8',}}\n", "\n", "epics = dict()\n", "for epic in epic_list():\n", " epics[epic['id']] = epic\n", "\n", "projects = dict()\n", "for project in project_list():\n", " projects[project['id']] = project" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Place the stories into dictionary with key of the epic name\n", "# this is later thought to be not quite needed as refactoring\n", "# happened, but we use this dict to loop on later.\n", "stories = dict()\n", "for story in search_stories(run_params[\"clubhouse_query\"])['data']:\n", " stories[story['id']] = story" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def remove_clipboard_links(text_with_links):\n", " # the description text has some links to the embedded images\n", " # this removes those for our email purposes.\n", " t = text_with_links\n", " str_sentinel = {'begin':\"![Clipboard \", 'end':\")\"}\n", " clipboard_strings = search(str_sentinel[\"begin\"]+'{}'+str_sentinel[\"end\"], t)\n", " if clipboard_strings:\n", " for clipboard_str in clipboard_strings:\n", " t = t.replace(f'{str_sentinel[\"begin\"]}{clipboard_str}{str_sentinel[\"end\"]}',\"\")\n", " return t\n", "\n", " \n", "def nl2br(text_with_br):\n", " return text_with_br.replace('\\n','
')\n", "\n", "\n", "def clean_story(text_to_clean):\n", " return nl2br(remove_clipboard_links(text_to_clean))\n", "\n", "\n", "def get_story_type_color(story_type):\n", " return story_types[story_type]['colour']\n", "\n", "\n", "def epic_phrase(epic_id):\n", " if epic_id:\n", " return f\"on {epics[story['epic_id']]['name']}\"\n", " else:\n", " return ''\n", "\n", " \n", "def tag_format(name, colour):\n", " return (f\"{name}  \")\n", "\n", "\n", "def email_format(story):\n", " # NOTE TO READER\n", " # this was put together by a non css, non html email loving person and needed to \n", " # work quickly and look good in Gmail. The following produces a terrible looking \n", " # output in Outlook, but fine in O365 Outlook, and fine on the mobile app.\n", " \n", " rstr = list()\n", " \n", " rstr.append(f\"
\")\n", " \n", " rstr.append(f\"
{story['story_type']} {story['id']} {epic_phrase(story['epic_id'])} in {projects[story['project_id']]['name']}
\")\n", "\n", " if story['labels']:\n", " rstr.append(\"
\")\n", " \n", " if story['blocker']:\n", " rstr.append(tag_format(\"BLOCKING\", \"#cc5856\"))\n", " \n", " for tag in story['labels']:\n", " rstr.append(tag_format(tag[\"name\"], tag[\"color\"]))\n", "\n", " if story['labels']:\n", " rstr.append(\"
\")\n", " rstr.append(f\"
{story['name']}
\")\n", " rstr.append(f\"
{clean_story(story['description'])}
\")\n", " if story['completed_at']:\n", " c_date = dateutil.parser.parse(story['completed_at'])\n", " rstr.append(f\"
Marked as completed {c_date:{ch_date_format}}
\")\n", " rstr.append(\"
\")\n", " return ''.join(rstr) " ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "reports = dict()\n", "include_story = True\n", "for story_id, story in stories.items():\n", " state = workflow_states[story[\"workflow_state_id\"]][\"name\"]\n", " if (state == 'Completed'): \n", " # we only want recently closed stories\n", " c_date = dateutil.parser.parse(story['completed_at'])\n", " include_story = now - c_date < timedelta(days=run_params[\"complete_days\"])\n", " else:\n", " include_story = True\n", " if include_story:\n", " story['email_html'] = email_format(story)\n", " try:\n", " reports[state].append(story)\n", " except KeyError:\n", " reports[state] = [story,] " ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "formatted_html_email = list()\n", "\n", "formatted_html_email.append(\"\")\n", "formatted_html_email.append(f\"
For Query:
\")\n", "formatted_html_email.append(f\"
\")\n", "formatted_html_email.append(f\"{run_params['clubhouse_query']}
\")\n", "formatted_html_email.append(f\"
\")\n", "for rs in run_params[\"reporting_states\"]:\n", " \n", " if reports.get(rs, None):\n", " formatted_html_email.append(f\"
{rs}\")\n", "\n", " if rs == \"Completed\":\n", " formatted_html_email.append(f\" - Last {run_params['complete_days']} days\")\n", " formatted_html_email.append(\"
\")\n", "\n", " for r in sorted(reports[rs], key=lambda k: k['position']):\n", " formatted_html_email.append(r[\"email_html\"])\n", "\n", "formatted_html_email.append(\"\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "send_mail(send_from=run_params['send_from_addr'],\n", " send_to=run_params['send_to_addrs'],\n", " subject=run_params['email_subject'],\n", " html_text=''.join(formatted_html_email),\n", " plain_text=\"Update from development.\",\n", " )" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "**MIT License**\n", "\n", "Copyright (c) 2020 Junction Applications\n", "\n", "Permission is hereby granted, free of charge, to any person obtaining a copy\n", "of this software and associated documentation files (the \"Software\"), to deal\n", "in the Software without restriction, including without limitation the rights\n", "to use, copy, modify, merge, publish, distribute, sublicense, and/or sell\n", "copies of the Software, and to permit persons to whom the Software is\n", "furnished to do so, subject to the following conditions:\n", "\n", "The above copyright notice and this permission notice shall be included in all\n", "copies or substantial portions of the Software.\n", "\n", "THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR\n", "IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,\n", "FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE\n", "AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER\n", "LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\n", "OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE\n", "SOFTWARE." ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.7.3" } }, "nbformat": 4, "nbformat_minor": 2 }