mirror of
https://github.com/miloszowi/everyone-mention-telegram-bot.git
synced 2025-11-08 23:35:52 +00:00
PEP-8 coding standards, used dataclasses for User and UpdateData classes
This commit is contained in:
@@ -8,17 +8,14 @@ from handler.vo.updateData import UpdateData
|
||||
|
||||
|
||||
class AbstractHandler:
|
||||
def __init__(self) -> None:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def getBotHandler(self) -> Handler: raise Exception('getBotHandler method is not implemented')
|
||||
def get_bot_handler(self) -> Handler: raise Exception('get_bot_handler method is not implemented')
|
||||
|
||||
@abstractmethod
|
||||
def handle(self, update: Update, context: CallbackContext) -> None: raise Exception('handle method is not implemented')
|
||||
|
||||
def getUpdateData(self, update: Update) -> UpdateData:
|
||||
return UpdateData.createFromUpdate(update)
|
||||
def get_update_data(self, update: Update) -> UpdateData:
|
||||
return UpdateData.create_from_update(update)
|
||||
|
||||
def reply(self, update: Update, message: str) -> None:
|
||||
update.effective_message.reply_markdown_v2(text=message)
|
||||
|
||||
@@ -9,30 +9,30 @@ from handler.abstractHandler import AbstractHandler
|
||||
|
||||
|
||||
class InHandler(AbstractHandler):
|
||||
botHandler: CommandHandler
|
||||
userRepository: UserRepository
|
||||
bot_handler: CommandHandler
|
||||
user_repository: UserRepository
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.botHandler = CommandHandler('in', self.handle)
|
||||
self.userRepository = UserRepository()
|
||||
self.bot_handler = CommandHandler('in', self.handle)
|
||||
self.user_repository = UserRepository()
|
||||
|
||||
def handle(self, update: Update, context: CallbackContext) -> None:
|
||||
updateData = self.getUpdateData(update)
|
||||
update_data = self.get_update_data(update)
|
||||
|
||||
try:
|
||||
user = self.userRepository.getById(updateData.getUserId())
|
||||
user = self.user_repository.get_by_id(update_data.user_id)
|
||||
|
||||
if user.isInChat(updateData.getChatId()):
|
||||
if user.is_in_chat(update_data.chat_id):
|
||||
self.reply(update, opted_in_failed)
|
||||
return
|
||||
|
||||
user.addToChat(updateData.getChatId())
|
||||
self.userRepository.save(user)
|
||||
user.add_to_chat(update_data.chat_id)
|
||||
self.user_repository.save(user)
|
||||
|
||||
except NotFoundException:
|
||||
self.userRepository.saveByUpdateData(updateData)
|
||||
self.user_repository.save_by_update_data(update_data)
|
||||
|
||||
self.reply(update, opted_in)
|
||||
|
||||
def getBotHandler(self) -> CommandHandler:
|
||||
return self.botHandler
|
||||
def get_bot_handler(self) -> CommandHandler:
|
||||
return self.bot_handler
|
||||
|
||||
@@ -11,37 +11,37 @@ from handler.abstractHandler import AbstractHandler
|
||||
|
||||
|
||||
class MentionHandler(AbstractHandler):
|
||||
botHandler: CommandHandler
|
||||
userRepository: UserRepository
|
||||
bot_handler: CommandHandler
|
||||
user_repository: UserRepository
|
||||
silent: str = 'silent'
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.botHandler = CommandHandler('everyone', self.handle)
|
||||
self.userRepository = UserRepository()
|
||||
self.bot_handler = CommandHandler('everyone', self.handle)
|
||||
self.user_repository = UserRepository()
|
||||
|
||||
def handle(self, update: Update, context: CallbackContext) -> None:
|
||||
updateData = self.getUpdateData(update)
|
||||
users = self.userRepository.getAllForChat(updateData.getChatId())
|
||||
updateData = self.get_update_data(update)
|
||||
users = self.user_repository.get_all_for_chat(updateData.chat_id)
|
||||
|
||||
if users:
|
||||
self.reply(update, self.buildMentionMessage(users, self.isSilent(context)))
|
||||
self.reply(update, self.build_mention_message(users, self.isSilent(context)))
|
||||
return
|
||||
|
||||
self.reply(update, mention_failed)
|
||||
|
||||
def getBotHandler(self) -> CommandHandler:
|
||||
return self.botHandler
|
||||
def get_bot_handler(self) -> CommandHandler:
|
||||
return self.bot_handler
|
||||
|
||||
def buildMentionMessage(self, users: Iterable[User], silent: bool = False) -> str:
|
||||
def build_mention_message(self, users: Iterable[User], silent: bool = False) -> str:
|
||||
result = ''
|
||||
|
||||
for user in users:
|
||||
if not silent:
|
||||
result += f'*[{user.getUsername()}](tg://user?id={user.getUserId()})* '
|
||||
result += f'*[{user.username}](tg://user?id={user.user_id})* '
|
||||
else:
|
||||
result += f'*{user.getUsername()}\({user.getUserId()}\)*\n'
|
||||
result += f'*{user.username}\({user.user_id}\)*\n'
|
||||
|
||||
return result
|
||||
|
||||
def isSilent(self, context: CallbackContext) -> bool:
|
||||
return context.args and context.args[0] == self.silent
|
||||
return self.silent in context.args
|
||||
@@ -9,28 +9,29 @@ from handler.abstractHandler import AbstractHandler
|
||||
|
||||
|
||||
class OutHandler(AbstractHandler):
|
||||
botHandler: CommandHandler
|
||||
userRepository: UserRepository
|
||||
bot_handler: CommandHandler
|
||||
user_repository: UserRepository
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.botHandler = CommandHandler('out', self.handle)
|
||||
self.userRepository = UserRepository()
|
||||
self.bot_handler = CommandHandler('out', self.handle)
|
||||
self.user_repository = UserRepository()
|
||||
|
||||
def handle(self, update: Update, context: CallbackContext) -> None:
|
||||
updateData = self.getUpdateData(update)
|
||||
updateData = self.get_update_data(update)
|
||||
|
||||
try:
|
||||
user = self.userRepository.getById(updateData.getUserId())
|
||||
if not user.isInChat(updateData.getChatId()):
|
||||
user = self.user_repository.get_by_id(updateData.user_id)
|
||||
|
||||
if not user.is_in_chat(updateData.chat_id):
|
||||
raise NotFoundException()
|
||||
except NotFoundException:
|
||||
self.reply(update, opted_off_failed)
|
||||
return
|
||||
|
||||
user.removeFromChat(updateData.getChatId())
|
||||
self.userRepository.save(user)
|
||||
user.remove_from_chat(updateData.chat_id)
|
||||
self.user_repository.save(user)
|
||||
|
||||
self.reply(update, opted_off)
|
||||
|
||||
def getBotHandler(self) -> CommandHandler:
|
||||
return self.botHandler
|
||||
def get_bot_handler(self) -> CommandHandler:
|
||||
return self.bot_handler
|
||||
|
||||
@@ -1,35 +1,24 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
import names
|
||||
from telegram.update import Update
|
||||
|
||||
|
||||
@dataclass
|
||||
class UpdateData():
|
||||
userId: str
|
||||
chatId: str
|
||||
user_id: str
|
||||
chat_id: str
|
||||
username: str
|
||||
|
||||
def __init__(self, userId: str, chatId: str, username: str) -> None:
|
||||
self.userId = userId
|
||||
self.chatId = chatId
|
||||
self.username = username
|
||||
|
||||
def getUserId(self) -> str:
|
||||
return self.userId
|
||||
|
||||
def getChatId(self) -> str:
|
||||
return self.chatId
|
||||
|
||||
def getUsername(self) -> str:
|
||||
return self.username
|
||||
|
||||
@staticmethod
|
||||
def createFromUpdate(update: Update) -> UpdateData:
|
||||
userId = str(update.effective_user.id)
|
||||
chatId = str(update.effective_chat.id)
|
||||
def create_from_update(update: Update) -> UpdateData:
|
||||
user_id = str(update.effective_user.id)
|
||||
chat_id = str(update.effective_chat.id)
|
||||
username = update.effective_user.username or update.effective_user.first_name
|
||||
|
||||
if not username:
|
||||
username = names.get_first_name()
|
||||
|
||||
return UpdateData(userId, chatId, username)
|
||||
return UpdateData(user_id, chat_id, username)
|
||||
|
||||
Reference in New Issue
Block a user