Version 0.1.1

.github/workflows/release.yaml

@@ -2,8 +2,6 @@ name: Release
 
 "on":
   push:
-    branches:
-      - main
     tags:
       - "[0-9]+.[0-9]+.[0-9]+"
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "smsbot"
-version = "0.1.0"
+version = "0.1.1"
 description = "A simple Telegram bot to receive SMS messages."
 authors = [{ name = "Andrew Williams", email = "andy@tensixtyone.com" }]
 license = { text = "MIT" }

smsbot/utils.py

@@ -5,19 +5,14 @@ def get_smsbot_version():
     return version("smsbot")
 
 
-class TwilioMessage:
+class TwilioWebhookPayload:
-    """
+    @staticmethod
-    Parses a Twilio webhook message.
+    def parse(data: dict):
-    """
+        """Return the correct class for the incoming Twilio webhook payload"""
-
+        if "SmsMessageSid" in data:
-    def __init__(self, data: dict) -> None:
+            return TwilioMessage(data)
-        self.from_number: str = data.get("From", "Unknown")
+        if "CallSid" in data:
-        self.to_number: str = data.get("To", "Unknown")
+            return TwilioCall(data)
-        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 _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

uv.lock

@@ -567,7 +567,7 @@ wheels = [
 
 [[package]]
 name = "smsbot"
-version = "0.1.0"
+version = "0.1.1"
 source = { editable = "." }
 dependencies = [
     { name = "flask", extra = ["async"] },