Miroslavsckaya/tg_rss_bot
1
1
Fork 0
Code Issues 7 Pull requests Projects 1 Releases Activity

#7 Code style changes recommended by pylint.
Some checks failed
continuous-integration/drone/push Build is passing
Details
continuous-integration/drone/pr Build is failing
Details

Browse source
This commit is contained in:
Alexey Skobkin 2022-05-30 03:17:11 +03:00
parent 93610e4084
commit 9e4488b204
1 changed files with 17 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
telegram.py
View file

@ -16,7 +16,7 @@ class CommandProcessor:
self.bot: TeleBot = TeleBot(token, use_class_middlewares=True) self.bot: TeleBot = TeleBot(token, use_class_middlewares=True)
self.bot.setup_middleware(UserAuthMiddleware(database)) self.bot.setup_middleware(UserAuthMiddleware(database))
self.bot.setup_middleware(ExceptionHandlerMiddleware(self.bot)) self.bot.setup_middleware(ExceptionHandlerMiddleware(self.bot))
self.db: Database = database self.database: Database = database
def run(self): def run(self):
"""Run a bot and poll for new messages indefinitely.""" """Run a bot and poll for new messages indefinitely."""
@ -28,7 +28,7 @@ class CommandProcessor:
self.bot.infinity_polling() self.bot.infinity_polling()
def __command_help(self, message: Message, data: dict): def __command_help(self, message: Message):
self.bot.reply_to( self.bot.reply_to(
message, message,
'Supported commands:\n' 'Supported commands:\n'
@ -47,12 +47,12 @@ class CommandProcessor:
if not self.__is_url_valid(url): if not self.__is_url_valid(url):
raise DisplayableException('Invalid feed 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.') self.bot.reply_to(message, 'Successfully subscribed to feed.')
def __list_feeds(self, message: Message, data: dict): 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 = '' feed_list = ''
for feed in feeds: for feed in feeds:
@ -69,15 +69,16 @@ class CommandProcessor:
if not self.__is_url_valid(url): if not self.__is_url_valid(url):
raise DisplayableException('Invalid feed 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.') self.bot.reply_to(message, 'Unsubscribed.')
# pylint: disable=R0201
def __is_url_valid(self, url: str) -> bool: def __is_url_valid(self, url: str) -> bool:
if not validators.url(url): if not validators.url(url):
return False 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://')): if not url.startswith(('http://', 'https://')):
return False return False
return True return True
@ -101,6 +102,7 @@ class Notifier:
parse_mode='MarkdownV2' parse_mode='MarkdownV2'
) )
# pylint: disable=R0201
def __format_message(self) -> str: def __format_message(self) -> str:
update = { update = {
'title': 'Item Title', 'title': 'Item Title',
@ -120,22 +122,21 @@ class UserAuthMiddleware(BaseMiddleware):
def __init__(self, db: Database): def __init__(self, db: Database):
super().__init__() super().__init__()
self.update_types = ['message'] self.update_types = ['message']
self.db: Database = db self.database: Database = db
def pre_process(self, message: Message, data: dict): 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) data['user_id'] = self.__find_or_register_user(message)
def post_process(self, message: Message, data: dict, exception): def post_process(self, message: Message, data: dict, exception):
"""Post-process the message.""" """Post-process update."""
pass
def __find_or_register_user(self, message: Message) -> int: def __find_or_register_user(self, message: Message) -> int:
telegram_id = message.from_user.id 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: if user_id is None:
return self.db.add_user(telegram_id) return self.database.add_user(telegram_id)
return user_id return user_id
@ -147,13 +148,15 @@ class ExceptionHandlerMiddleware(BaseMiddleware):
self.update_types = ['message'] self.update_types = ['message']
self.bot: TeleBot = bot self.bot: TeleBot = bot
def pre_process(self, message, data): def pre_process(self, message: Message, data: dict):
"""Pre-process update."""
pass pass
def post_process(self, message: Message, data: dict, exception: Exception | None): def post_process(self, message: Message, data: dict, exception: Exception | None):
"""Post-process update. Send user an error notification."""
if exception is None: if exception is None:
return return
elif type(exception) is DisplayableException: if isinstance(exception, DisplayableException):
self.bot.reply_to(message, 'Error: ' + str(exception)) self.bot.reply_to(message, 'Error: ' + str(exception))
else: else:
self.bot.reply_to(message, 'Something went wrong. Please try again (maybe later).') self.bot.reply_to(message, 'Something went wrong. Please try again (maybe later).')