Handle call webhook types correctly

smsbot/utils.py

@@ -5,19 +5,14 @@ def get_smsbot_version():
     return version("smsbot")
 
 
-class TwilioMessage:
-    """
-    Parses a Twilio webhook message.
-    """
-
-    def __init__(self, data: dict) -> None:
-        self.from_number: str = data.get("From", "Unknown")
-        self.to_number: str = data.get("To", "Unknown")
-        self.body: str = data.get("Body", "")
-
-        self.media = []
-        for i in range(0, int(data.get("NumMedia", "0"))):
-            self.media.append(data.get(f"MediaUrl{i}"))
+class TwilioWebhookPayload:
+    @staticmethod
+    def parse(data: dict):
+        """Return the correct class for the incoming Twilio webhook payload"""
+        if "SmsMessageSid" in data:
+            return TwilioMessage(data)
+        if "CallSid" in data:
+            return TwilioCall(data)
 
     def _escape(self, text: str) -> str:
         """Escape text for MarkdownV2"""
@@ -45,6 +40,19 @@ class TwilioMessage:
             text = text.replace(char, rf"\{char}")
         return text
 
+
+class TwilioMessage(TwilioWebhookPayload):
+    """Represents a Twilio SMS message"""
+
+    def __init__(self, data: dict) -> None:
+        self.from_number: str = data.get("From", "Unknown")
+        self.to_number: str = data.get("To", "Unknown")
+        self.body: str = data.get("Body", "")
+
+        self.media = []
+        for i in range(0, int(data.get("NumMedia", "0"))):
+            self.media.append(data.get(f"MediaUrl{i}"))
+
     def __repr__(self) -> str:
         return f"TwilioWebhookMessage(from={self.from_number}, to={self.to_number})"
 
@@ -54,6 +62,29 @@ class TwilioMessage:
         return msg
 
     def to_markdownv2(self):
-        media_str = "\n".join([f"{self._escape(url)}" for url in self.media]) if self.media else ""
+        media_str = (
+            "\n".join([f"{self._escape(url)}" for url in self.media])
+            if self.media
+            else ""
+        )
         msg = f"**From**: {self._escape(self.from_number)}\n**To**: {self._escape(self.to_number)}\n\n{self._escape(self.body)}\n\n{media_str}"
         return msg
+
+
+class TwilioCall(TwilioWebhookPayload):
+    """Represents a Twilio voice call"""
+
+    def __init__(self, data: dict) -> None:
+        self.from_number: str = data.get("From", "Unknown")
+        self.to_number: str = data.get("To", "Unknown")
+
+    def __repr__(self) -> str:
+        return f"TwilioCall(from={self.from_number}, to={self.to_number})"
+
+    def to_str(self) -> str:
+        msg = f"Call from {self.from_number}, rejected."
+        return msg
+
+    def to_markdownv2(self):
+        msg = f"Call from {self._escape(self.from_number)}, rejected\."
+        return msg

smsbot/webhook_handler.py

@@ -7,7 +7,7 @@ from prometheus_client import Counter, Summary, make_wsgi_app
 from twilio.request_validator import RequestValidator
 from werkzeug.middleware.dispatcher import DispatcherMiddleware
 
-from smsbot.utils import TwilioMessage, get_smsbot_version
+from smsbot.utils import TwilioWebhookPayload, get_smsbot_version
 
 REQUEST_TIME = Summary(
     "webhook_request_processing_seconds", "Time spent processing request"
@@ -88,7 +88,7 @@ class TwilioWebhookHandler(object):
         )
 
         await self.bot.send_subscribers(
-            TwilioMessage(request.values.to_dict()).to_markdownv2()
+            TwilioWebhookPayload.parse(request.values.to_dict()).to_markdownv2()
         )
 
         # Return a blank response
@@ -103,7 +103,7 @@ class TwilioWebhookHandler(object):
             "Received Call from {From}".format(**request.values.to_dict())
         )
         await self.bot.send_subscribers(
-            "Received Call from {From}, rejecting.".format(**request.values.to_dict())
+            TwilioWebhookPayload.parse(request.values.to_dict()).to_markdownv2()
         )
 
         # Always reject calls