diff --git a/telegram.py b/telegram.py index 5f12374..f419e57 100644 --- a/telegram.py +++ b/telegram.py @@ -16,7 +16,7 @@ class CommandProcessor: self.bot: TeleBot = TeleBot(token, use_class_middlewares=True) self.bot.setup_middleware(UserAuthMiddleware(database)) self.bot.setup_middleware(ExceptionHandlerMiddleware(self.bot)) - self.db: Database = database + self.database: Database = database def run(self): """Run a bot and poll for new messages indefinitely.""" @@ -28,7 +28,7 @@ class CommandProcessor: self.bot.infinity_polling() - def __command_help(self, message: Message, data: dict): + def __command_help(self, message: Message): self.bot.reply_to( message, 'Supported commands:\n' @@ -47,12 +47,12 @@ class CommandProcessor: if not self.__is_url_valid(url): raise DisplayableException('Invalid feed URL') - self.db.subscribe_user_by_url(data['user_id'], url) + self.database.subscribe_user_by_url(data['user_id'], url) self.bot.reply_to(message, 'Successfully subscribed to feed.') def __list_feeds(self, message: Message, data: dict): - feeds = self.db.find_user_feeds(data['user_id']) + feeds = self.database.find_user_feeds(data['user_id']) feed_list = '' for feed in feeds: @@ -69,15 +69,16 @@ class CommandProcessor: if not self.__is_url_valid(url): raise DisplayableException('Invalid feed URL') - self.db.unsubscribe_user_by_url(data['user_id'], url) + self.database.unsubscribe_user_by_url(data['user_id'], url) self.bot.reply_to(message, 'Unsubscribed.') - def __is_url_valid(self, url: str) -> bool: + @staticmethod + def __is_url_valid(url: str) -> bool: if not validators.url(url): return False - """For security reasons we should not allow anything except HTTP/HTTPS.""" + # For security reasons we should not allow anything except HTTP/HTTPS. if not url.startswith(('http://', 'https://')): return False return True @@ -101,13 +102,8 @@ class Notifier: parse_mode='MarkdownV2' ) - def __format_message(self) -> str: - update = { - 'title': 'Item Title', - 'text': 'Short item text here...', - 'url': 'https://fediland.github.io/', - } - + @staticmethod + def __format_message(item) -> str: return ( f"**[{update['title']}]({update['url']})**\n\n" f"{update['text']}" @@ -117,25 +113,24 @@ class Notifier: class UserAuthMiddleware(BaseMiddleware): """Transparently authenticates and registers the user if needed.""" - def __init__(self, db: Database): + def __init__(self, database: Database): super().__init__() self.update_types = ['message'] - self.db: Database = db + self.database: Database = database def pre_process(self, message: Message, data: dict): - """Pre-process the message, find user and add it's ID to the handler data dictionary.""" + """Pre-process update, find user and add it's ID to the handler data dictionary.""" data['user_id'] = self.__find_or_register_user(message) def post_process(self, message: Message, data: dict, exception): - """Post-process the message.""" - pass + """Post-process update.""" def __find_or_register_user(self, message: Message) -> int: telegram_id = message.from_user.id - user_id = self.db.find_user(telegram_id) + user_id = self.database.find_user(telegram_id) if user_id is None: - return self.db.add_user(telegram_id) + return self.database.add_user(telegram_id) return user_id @@ -147,13 +142,15 @@ class ExceptionHandlerMiddleware(BaseMiddleware): self.update_types = ['message'] self.bot: TeleBot = bot - def pre_process(self, message, data): - pass + def pre_process(self, message: Message, data: dict): + """Pre-process update.""" + # pylint: disable=W0613 def post_process(self, message: Message, data: dict, exception: Exception | None): + """Post-process update. Send user an error notification.""" if exception is None: return - elif type(exception) is DisplayableException: + if isinstance(exception, DisplayableException): self.bot.reply_to(message, 'Error: ' + str(exception)) else: self.bot.reply_to(message, 'Something went wrong. Please try again (maybe later).')