Add support for Microsoft Teams notifications

.env.template

@@ -3,6 +3,8 @@
 
 # DISCORD_WEBHOOK_URL=
 
+# TEAMS_WEBHOOK_URL=
+
 # TWITTER_CONSUMER_KEY=
 # TWITTER_CONSUMER_SECRET=
 # TWITTER_ACCESS_TOKEN_KEY=

README.md

@@ -2,7 +2,7 @@
 Bot to notify when vaccine appointments are available.
 
 Supports checking Hy-Vee, Walgreens, CVS, Walmart, Cosentino's stores (KC), and Ball's stores (KC).
-Supports sending notifications to Slack, Discord, Twilio, and [Twitter](https://twitter.com/kcvaccinewatch).
+Supports sending notifications to Slack, Discord, Microsoft Teams, Twilio, and [Twitter](https://twitter.com/kcvaccinewatch).
 
 Notifications are sent when a location has appointments. No more notifications are sent for that location until it becomes unavailable again.
 
@@ -41,6 +41,13 @@ Twitter
 1. Select the `Integrations` tab and click `Create Webhook`
 1. Enter a `Name` and `Channel` you want the webhook to notify and copy the Webhook URL.
 
+### Microsoft Teams
+1.  In the channel where you want to add the incoming webhook, click ... and then Connectors.
+1.  Search for Incoming Webhook and click Add.
+1.  Give the webhook a name (e.g. Vaccine Watch).
+1.  Click Create.
+1.  A unique webhook URL will be provided for the channel.
+
 ### Twitter
 1. Apply for a [Twitter Developer account](https://developer.twitter.com/en/portal/petition/use-case)
 1. Once you have the account, go to the [Developer Portal](https://developer.twitter.com/en/portal/dashboard)
@@ -94,6 +101,9 @@ Optional Environment Variables:
 - Discord:
   - `DISCORD_WEBHOOK_URL`: Discord webhook url for channel.
     - example: `https://discordapp.com/api/webhooks/1234567890/abc123`
+- Microsoft Teams:
+  - `TEAMS_WEBHOOK_URL`: Teams webhook url for channel.
+    - example: `https://company.webhook.office.com/webhookb2/abc123@def456/IncomingWebhook/aaa111/bbb222`
 - Twilio:
   - `TWILIO_ACCOUNT_SID`: Account SID for your twilio account
   - `TWILIO_AUTH_TOKEN`: Auth token for your twilio account

notify/teams.py

@@ -0,0 +1,86 @@
+import json
+import logging
+import os
+
+import requests
+
+from utils import timeout_amount
+
+from . import NotificationMethod
+
+
+class Teams(NotificationMethod):
+    def notify_available_locations(self, locations):
+        send_message_to_teams(format_available_message(locations))
+
+    def notify_unavailable_locations(self, locations):
+        send_message_to_teams(format_unavailable_message(locations))
+
+
+states = json.loads(os.environ["STATES"])
+
+
+def format_available_message(locations):
+    message = "\u2705 Vaccine appointments available at {} location{}:".format(
+        "these" if len(locations) > 1 else "this",
+        "s" if len(locations) > 1 else "",
+    )
+    for location in locations:
+        if "earliest_appointment_day" in location:
+            if (
+                location["earliest_appointment_day"]
+                == location["latest_appointment_day"]
+            ):
+                day_string = " on **{}**".format(location["earliest_appointment_day"])
+            else:
+                day_string = " from **{}** to **{}**".format(
+                    location["earliest_appointment_day"],
+                    location["latest_appointment_day"],
+                )
+        else:
+            day_string = ""
+
+        message += "\n\n* {}{}{}. Sign up **[here]({})**{}{}".format(
+            "**{}**: ".format(location["state"])
+            if (len(states) > 1 and "state" in location)
+            else "",
+            location["name"],
+            day_string,
+            location["link"],
+            ", zip code {}".format(location["zip"]) if "zip" in location else "",
+            " (as of {})".format(location["appointments_last_fetched"])
+            if location.get("appointments_last_fetched", None)
+            else "",
+        )
+    return message
+
+
+def format_unavailable_message(locations):
+    message = (
+        "\u274C Vaccine appointments no longer available at {} location{}:".format(
+            "these" if len(locations) > 1 else "this",
+            "s" if len(locations) > 1 else "",
+        )
+    )
+    for location in locations:
+        message += "\n\n* {}{}{}".format(
+            "**{}**: ".format(location["state"])
+            if (len(states) > 1 and "state" in location)
+            else "",
+            location["name"],
+            " (as of {})".format(location["appointments_last_fetched"])
+            if location.get("appointments_last_fetched", None)
+            else "",
+        )
+    return message
+
+
+def send_message_to_teams(message):
+    webhook_url = os.environ["TEAMS_WEBHOOK_URL"]
+    data = {"text": message}
+    try:
+        response = requests.post(webhook_url, json=data, timeout=timeout_amount)
+        response.raise_for_status()
+        logging.debug("Message to teams sent successfully")
+    except requests.exceptions.RequestException:
+        logging.exception("Error sending message to teams")

vaccine.py

@@ -12,6 +12,7 @@
 from notify.console import Console
 from notify.discord import Discord
 from notify.slack import Slack
+from notify.teams import Teams
 from notify.twilio import Twilio
 from notify.twitter import Twitter
 from utils import env_var_is_true
@@ -41,6 +42,8 @@
     enabled_notification_methods.append(Discord())
 if "SLACK_BOT_TOKEN" in os.environ:
     enabled_notification_methods.append(Slack())
+if "TEAMS_WEBHOOK_URL" in os.environ:
+    enabled_notification_methods.append(Teams())
 if "TWILIO_AUTH_TOKEN" in os.environ:
     enabled_notification_methods.append(Twilio())
 if "TWITTER_CONSUMER_KEY" in os.environ: