diff --git a/smsbot/telegram.py b/smsbot/telegram.py
index 11bb11b..7489685 100644
--- a/smsbot/telegram.py
+++ b/smsbot/telegram.py
@@ -12,9 +12,7 @@ from telegram.ext import (
from smsbot.utils import get_smsbot_version
-REQUEST_TIME = Summary(
- "telegram_request_processing_seconds", "Time spent processing request"
-)
+REQUEST_TIME = Summary("telegram_request_processing_seconds", "Time spent processing request")
COMMAND_COUNT = Counter("telegram_command_count", "Total number of commands processed")
@@ -35,21 +33,18 @@ class TelegramSmsBot:
async def callback(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle the update"""
- if update.effective_user.id in self.owners:
- self.logger.info(
- f"{update.effective_user.username} sent {update.message.text}"
- )
- COMMAND_COUNT.inc()
- else:
- self.logger.debug(f"Ignoring message from user {update.effective_user.id}")
- raise ApplicationHandlerStop
+ if update.effective_user and update.message:
+ if update.effective_user.id in self.owners:
+ self.logger.info(f"{update.effective_user.username} sent {update.message.text}")
+ COMMAND_COUNT.inc()
+ else:
+ self.logger.debug(f"Ignoring message from user {update.effective_user.username}")
+ raise ApplicationHandlerStop
async def send_message(self, chat_id: int, text: str):
"""Send a message to a specific chat"""
self.logger.info(f"Sending message to chat {chat_id}: {text}")
- await self.app.bot.send_message(
- chat_id=chat_id, text=text, parse_mode="MarkdownV2"
- )
+ await self.app.bot.send_message(chat_id=chat_id, text=text, parse_mode="MarkdownV2")
async def send_subscribers(self, text: str):
"""Send a message to all subscribers"""
@@ -64,48 +59,41 @@ class TelegramSmsBot:
await self.send_message(owner, text)
@REQUEST_TIME.time()
- async def handler_help(self, update, context):
+ async def handler_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Send a message when the command /help is issued."""
- self.logger.info("/help command received in chat: %s", update.message.chat)
+ if update.message:
+ self.logger.info("/help command received in chat: %s", update.message.chat)
- commands = []
- for command in self.app.handlers[0]:
- if isinstance(command, CommandHandler):
- commands.extend(["/{0}".format(cmd) for cmd in command.commands])
+ commands = []
+ for command in self.app.handlers[0]:
+ if isinstance(command, CommandHandler):
+ commands.extend(["/{0}".format(cmd) for cmd in command.commands])
- await update.message.reply_markdown(
- "Smsbot v{0}\n\n{1}".format(get_smsbot_version(), "\n".join(commands))
- )
- COMMAND_COUNT.inc()
+ await update.message.reply_markdown("Smsbot v{0}\n\n{1}".format(get_smsbot_version(), "\n".join(commands)))
+ COMMAND_COUNT.inc()
@REQUEST_TIME.time()
- async def handler_subscribe(
- self, update: Update, context: ContextTypes.DEFAULT_TYPE
- ):
+ async def handler_subscribe(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle subscription requests"""
- user_id = update.effective_user.id
- if user_id not in self.subscribers:
- self.subscribers.append(user_id)
- self.logger.info(f"User {user_id} subscribed.")
- self.logger.info(f"Current subscribers: {self.subscribers}")
- await update.message.reply_markdown(
- "You have successfully subscribed to updates."
- )
- else:
- self.logger.info(f"User {user_id} is already subscribed.")
+ if update.effective_user and update.message:
+ user_id = update.effective_user.id
+ if user_id not in self.subscribers:
+ self.subscribers.append(user_id)
+ self.logger.info(f"User {user_id} subscribed.")
+ self.logger.info(f"Current subscribers: {self.subscribers}")
+ await update.message.reply_markdown("You have successfully subscribed to updates.")
+ else:
+ self.logger.info(f"User {user_id} is already subscribed.")
@REQUEST_TIME.time()
- async def handler_unsubscribe(
- self, update: Update, context: ContextTypes.DEFAULT_TYPE
- ):
+ async def handler_unsubscribe(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle unsubscription requests"""
- user_id = update.effective_user.id
- if user_id in self.subscribers:
- self.subscribers.remove(user_id)
- self.logger.info(f"User {user_id} unsubscribed.")
- self.logger.info(f"Current subscribers: {self.subscribers}")
- await update.message.reply_markdown(
- "You have successfully unsubscribed from updates."
- )
- else:
- self.logger.info(f"User {user_id} is not subscribed.")
+ if update.effective_user and update.message:
+ user_id = update.effective_user.id
+ if user_id in self.subscribers:
+ self.subscribers.remove(user_id)
+ self.logger.info(f"User {user_id} unsubscribed.")
+ self.logger.info(f"Current subscribers: {self.subscribers}")
+ await update.message.reply_markdown("You have successfully unsubscribed from updates.")
+ else:
+ self.logger.info(f"User {user_id} is not subscribed.")
diff --git a/smsbot/utils.py b/smsbot/utils.py
index 89800b0..7d4bf2b 100644
--- a/smsbot/utils.py
+++ b/smsbot/utils.py
@@ -1,13 +1,13 @@
from importlib.metadata import version
-def get_smsbot_version():
+def get_smsbot_version() -> str:
return version("smsbot")
class TwilioWebhookPayload:
@staticmethod
- def parse(data: dict):
+ def parse(data: dict[str, str]) -> "TwilioCall | TwilioMessage | None":
"""Return the correct class for the incoming Twilio webhook payload"""
if "SmsMessageSid" in data:
return TwilioMessage(data)
@@ -62,11 +62,7 @@ class TwilioMessage(TwilioWebhookPayload):
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
@@ -85,6 +81,6 @@ class TwilioCall(TwilioWebhookPayload):
msg = f"Call from {self.from_number}, rejected."
return msg
- def to_markdownv2(self):
+ def to_markdownv2(self) -> str:
msg = f"Call from {self._escape(self.from_number)}, rejected\\."
return msg
diff --git a/smsbot/webhook.py b/smsbot/webhook.py
index 9477585..427238d 100644
--- a/smsbot/webhook.py
+++ b/smsbot/webhook.py
@@ -8,18 +8,16 @@ from werkzeug.middleware.dispatcher import DispatcherMiddleware
from smsbot.utils import TwilioWebhookPayload, get_smsbot_version
-REQUEST_TIME = Summary(
- "webhook_request_processing_seconds", "Time spent processing request"
-)
+REQUEST_TIME = Summary("webhook_request_processing_seconds", "Time spent processing request")
MESSAGE_COUNT = Counter("webhook_message_count", "Total number of messages processed")
CALL_COUNT = Counter("webhook_call_count", "Total number of calls processed")
-
class TwilioWebhookHandler(object):
"""
A wrapped Flask app handling webhooks received from Twilio
"""
+
def __init__(self, account_sid: str | None = None, auth_token: str | None = None):
self.app = Flask(self.__class__.__name__)
self.app.add_url_rule("/", "index", self.index, methods=["GET"])
@@ -50,9 +48,7 @@ class TwilioWebhookHandler(object):
async def decorated_function(*args, **kwargs):
# Create an instance of the RequestValidator class
if not self.auth_token:
- current_app.logger.warning(
- "Twilio request validation skipped due to Twilio Auth Token missing"
- )
+ current_app.logger.warning("Twilio request validation skipped due to Twilio Auth Token missing")
return await func(*args, **kwargs)
validator = RequestValidator(self.auth_token)
@@ -75,10 +71,10 @@ class TwilioWebhookHandler(object):
def set_bot(self, bot):
self.bot = bot
- async def index(self):
+ async def index(self) -> str:
return f'smsbot v{get_smsbot_version()} - GitHub'
- async def health(self):
+ async def health(self) -> dict[str, str | int]:
"""Return basic health information"""
return {
"version": get_smsbot_version(),
@@ -87,29 +83,24 @@ class TwilioWebhookHandler(object):
}
@time(REQUEST_TIME)
- async def message(self):
+ async def message(self) -> str:
"""Handle incoming SMS messages from Twilio"""
- current_app.logger.info(
- "Received SMS from {From}: {Body}".format(**request.values.to_dict())
- )
-
- await self.bot.send_subscribers(
- TwilioWebhookPayload.parse(request.values.to_dict()).to_markdownv2()
- )
+ current_app.logger.info("Received SMS from {From}: {Body}".format(**request.values.to_dict()))
+ hook_data = TwilioWebhookPayload.parse(request.values.to_dict())
+ if hook_data:
+ await self.bot.send_subscribers(hook_data.to_markdownv2())
# Return a blank response
MESSAGE_COUNT.inc()
return ''
@time(REQUEST_TIME)
- async def call(self):
+ async def call(self) -> str:
"""Handle incoming calls from Twilio"""
- current_app.logger.info(
- "Received Call from {From}".format(**request.values.to_dict())
- )
- await self.bot.send_subscribers(
- TwilioWebhookPayload.parse(request.values.to_dict()).to_markdownv2()
- )
+ current_app.logger.info("Received Call from {From}".format(**request.values.to_dict()))
+ hook_data = TwilioWebhookPayload.parse(request.values.to_dict())
+ if hook_data:
+ await self.bot.send_subscribers(hook_data.to_markdownv2())
# Always reject calls
CALL_COUNT.inc()