From 8ba995bc5f099b5cc1a0de55775487cc723f5d2a Mon Sep 17 00:00:00 2001 From: Andrew Williams Date: Sun, 17 Aug 2025 07:30:34 +0100 Subject: [PATCH] Cleanup types --- smsbot/telegram.py | 88 ++++++++++++++++++++-------------------------- smsbot/utils.py | 12 +++---- smsbot/webhook.py | 39 ++++++++------------ 3 files changed, 57 insertions(+), 82 deletions(-) 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()