kerrad/telegram-helper-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add logging, refactor

Browse Source
This commit is contained in:
KatykhinAA
2024-07-11 22:59:58 +03:00
parent 66f3a870c4
commit 47050fd599
4 changed files with 169 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
helper_bot/helper_bot.py
View File

@@ -8,7 +8,10 @@ import telebot
from telebot import types
from telebot.apihelper import ApiTelegramException
import messages
from logs.custom_logger import Logger
#Инициализируем логгер
bot_logger = Logger(name='bot')
class State(Enum):
START = "START"
@@ -33,14 +36,17 @@ class TelegramHelperBot:
self.LOGS = self.settings['Settings']['logs']
self.TEST = self.settings['Settings']['test']
self.bot = telebot.TeleBot(token)
self.logger = bot_logger.get_logger()
# Router for user
@self.bot.message_handler(func=lambda message: True, chat_types=['private'])
def handle_message(message):
self.logger.info(f'Получено сообщение: {message.text} от пользователя: {message.from_user.full_name} id юзера: {message.chat.id}')
if self.BotDB.check_user_in_blacklist(message.from_user.id):
attribute = self.BotDB.get_blacklist_users_by_id(message.from_user.id)
self.bot.send_message(message.chat.id,
f'<b>Ты заблокирован\nПричина блокировки:</b> {attribute[2]}\n<b>Дата разблокировки:</b> {attribute[3]}', parse_mode='HTML')
self.logger.info(f'Пользователь {message.from_user.full_name} (ID: {message.chat.id}) заблокирован')
return
if self.state == State.START:
if message.text == '/start':
@@ -64,8 +70,10 @@ class TelegramHelperBot:
if access:
self.admin_panel(message)
self.state = State.ADMIN
self.logger.info(f'Пользователь {message.from_user.full_name} (ID: {message.chat.id}) вошел в админ-панель')
else:
self.bot.send_message(message.chat.id, 'Доступ запрещен, досвидания!')
self.logger.info(f'Пользователю {message.from_user.full_name} (ID: {message.chat.id}) отказано в доступе к админ-панели')
elif message.text == '/state':
self.bot.send_message(message.chat.id,
f'Твой state == {self.state.value}')
@@ -73,6 +81,7 @@ class TelegramHelperBot:
self.bot.send_message(message.chat.id,
#TODO: Здесь раньше был /state
"Не понимаю где ты находишься. Нажми /start, и я перезапущусь")
self.logger.info(f'Пользователь {message.from_user.full_name} (ID: {message.chat.id}) отправил непонятное сообщение: {message.text}')
if self.state == State.SUGGEST:
self.bot.register_next_step_handler(message, self.send_to_suggest)
@@ -80,20 +89,24 @@ class TelegramHelperBot:
if message.text == '/start':
self.state = State.START
self.start_message(message)
self.logger.info(f'Пользователь {message.from_user.full_name} (ID: {message.chat.id}) вернулся в главное меню')
if self.state == State.PRE_CHAT:
self.bot.register_next_step_handler(message, self.resend_message_in_group_for_message)
self.state = State.START
if message.text == '/start':
self.state = State.START
self.start_message(message)
self.logger.info(f'Пользователь {message.from_user.full_name} (ID: {message.chat.id}) вернулся в главное меню')
if self.state == State.CHAT:
if message.text == 'Выйти из чата':
self.state = State.START
self.end_message(message)
self.logger.info(f'Пользователь {message.from_user.full_name} (ID: {message.chat.id}) вышел из чата')
elif message.text == '/start':
self.state = State.START
self.start_message(message)
self.logger.info(f'Пользователь {message.from_user.full_name} (ID: {message.chat.id}) вернулся в главное меню')
else:
self.resend_message_in_group_for_message(message)
@@ -102,15 +115,19 @@ class TelegramHelperBot:
access = self.check_access(message.from_user.id)
if access:
self.admin_panel(message)
self.logger.info(f'Пользователь {message.from_user.full_name} (ID: {message.chat.id}) вошел в админ-панель')
else:
self.bot.send_message(message.chat.id, 'Доступ запрещен, досвидания!')
self.logger.info(f'Пользователю {message.from_user.full_name} (ID: {message.chat.id}) отказано в доступе к админ-панели')
if message.text == '/start':
self.state = State.START
self.start_message(message)
self.logger.info(f'Пользователь {message.from_user.full_name} (ID: {message.chat.id}) вернулся в главное меню')
@self.bot.message_handler(func=lambda message: True, chat_types=['group'])
def handle_message(message):
"""Функция ответа админа пользователю через закрытый чат"""
self.logger.info(f'Получено сообщение в группе {message.chat.title} (ID: {message.chat.id}) от пользователя {message.from_user.full_name} (ID: {message.from_user.id}): "{message.text}"')
self.state = State.CHAT
markup = types.ReplyKeyboardMarkup(resize_keyboard=True, one_time_keyboard=True)
item1 = types.KeyboardButton("Выйти из чата")
@@ -120,24 +137,30 @@ class TelegramHelperBot:
message_id = message.reply_to_message.id
except AttributeError:
self.bot.send_message(message.chat.id, f'Блять, выдели сообщение!')
self.logger.warning(f'В группе {message.chat.title} (ID: {message.chat.id}) админ не выделил сообщение для ответа.')
message_from_admin = message.text
try:
chat_id = self.BotDB.get_user_by_message_id(message_id)
self.bot.send_message(chat_id, message_from_admin, reply_markup=markup)
self.logger.info(f'Ответ админа "{message.text}" отправлен пользователю с ID: {chat_id}.')
except TypeError:
self.bot.send_message(message.chat.id, f'Не могу найти кому ответить в базе, проебали сообщение.')
self.logger.error(f'Ошибка при поиске пользователя в базе для ответа: {message.text} в группе {message.chat.title} (ID: {message.chat.id})')
# Админка
@self.bot.callback_query_handler(func=lambda call: call.data in ['publish', 'decline'])
def post_for_group(call):
self.logger.info(f'Получен callback-запрос с данными: {call.data} от пользователя {call.from_user.full_name} (ID: {call.from_user.id})')
if call.data == 'publish' and call.message.content_type == 'text':
try:
self.bot.send_message(chat_id=self.MAIN_PUBLIC, text=call.message.text)
self.bot.delete_message(chat_id=self.GROUP_FOR_POST, message_id=call.message.message_id)
self.logger.info(f'Текст сообщения опубликован в канале {self.MAIN_PUBLIC}.')
except Exception as e:
if self.LOGS:
self.bot.send_message(chat_id=self.IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
self.logger.error(f'Ошибка при публикации текста в канал {self.MAIN_PUBLIC}: {str(e)}')
elif call.data == 'publish' and call.message.content_type == 'photo':
try:
self.bot.send_photo(
@@ -146,41 +169,50 @@ class TelegramHelperBot:
photo=call.message.photo[-1].file_id,
)
self.bot.delete_message(chat_id=self.GROUP_FOR_POST, message_id=call.message.message_id)
self.logger.info(f'Пост с фото опубликован в канале {self.MAIN_PUBLIC}.')
except Exception as e:
if self.LOGS:
self.bot.send_message(chat_id=self.IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
self.logger.error(f'Ошибка при публикации фотографии в канал {self.MAIN_PUBLIC}: {str(e)}')
elif call.data == 'decline':
try:
self.bot.delete_message(chat_id=self.GROUP_FOR_POST, message_id=call.message.message_id)
self.logger.info(f'Сообщение отклонено админом {call.from_user.full_name} (ID: {call.from_user.id}).')
except Exception as e:
if self.LOGS:
self.bot.send_message(self.IMPORTANT_LOGS,
f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
self.logger.error(f'Ошибка при удалении сообщения в группе {self.GROUP_FOR_POST}: {str(e)}')
@self.bot.callback_query_handler(func=lambda call: True)
def pagination(call):
if call.data[:3] == 'ban':
user_id = call.data[4:]
self.logger.info(f"Бан пользователя с ID: {user_id}")
self.ban_user(call.message, user_id)
if call.data == 'return':
self.logger.info(f"Возврат в админ панель")
self.bot.delete_message(call.message.chat.id, call.message.message_id)
self.admin_panel(call.message)
if call.data[:5] == 'unban':
self.delete_user_blacklist(call.data[6:])
user_id = call.data[6:]
self.delete_user_blacklist(user_id)
msg = f'Успешно удалено.'
self.bot.send_message(chat_id=call.message.chat.id, text=msg)
self.logger.info(f"Разблокирован пользователь с ID: {user_id}")
elif call.data[:4] == 'page':
page_number = int(call.data[5:])
self.logger.info(f"Переход на страницу {page_number}")
if call.message.text == 'Список пользователей которые последними обращались к боту':
list_users = self.BotDB.get_last_users_from_db()
keyboard = self.create_keyboard_with_pagination(int(call.data[5:]), len(list_users), list_users,
keyboard = self.create_keyboard_with_pagination(int(page_number), len(list_users), list_users,
'ban')
self.bot.edit_message_reply_markup(call.message.chat.id, call.message.message_id,
reply_markup=keyboard)
if "Список заблокированных пользователей".lower() in call.message.text.lower():
elif "Список заблокированных пользователей".lower() in call.message.text.lower():
#Готовим сообщения
message_user = self.get_banned_users_list(int(call.data[5:]) * 7 - 7)
message_user = self.get_banned_users_list(int(page_number) * 7 - 7)
self.bot.edit_message_text(chat_id=call.message.chat.id, message_id=call.message.message_id,
text=message_user)
@@ -189,13 +221,17 @@ class TelegramHelperBot:
keyboard = self.create_keyboard_with_pagination(int(call.data[5:]), len(buttons), buttons, 'unban')
self.bot.edit_message_reply_markup(call.message.chat.id, call.message.message_id,
reply_markup=keyboard)
else:
self.logger.warning(f"Неизвестный callback data: {call.data}")
def start(self):
while True:
try:
self.bot.polling(none_stop=True)
except (ConnectionError, Exception):
print(f"Произошла ошибка: {str(Exception)}\n\nTraceback:\n{traceback.format_exc()}")
except ConnectionError as e:
self.logger.error(f"Произошла ошибка (потеря коннекта): {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
except Exception as e:
self.logger.error(f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
def unban_notifier(self):
# Получение сегодняшней даты в формате DD-MM-YYYY
@@ -214,6 +250,7 @@ class TelegramHelperBot:
# Черный список
def admin_panel(self, message):
try:
self.logger.info(f"Запуск админ панели для пользователя: {message.from_user.id}")
markup = types.ReplyKeyboardMarkup(resize_keyboard=True, one_time_keyboard=True)
item1 = types.KeyboardButton("Бан (Список)")
#item2 = types.KeyboardButton("Добавить админа") #TODO: Когда-нибудь потом доделаю
@@ -225,9 +262,11 @@ class TelegramHelperBot:
reply_markup=markup)
self.bot.register_next_step_handler(message, self.handle_admin_message)
except Exception as e:
self.logger.error(f"Ошибка при запуске админ панели: {e}")
self.bot.register_next_step_handler(message, self.admin_panel)
def handle_admin_message(self, message):
self.logger.info(f"Получено сообщение от админа: {message.text} (пользователь: {message.from_user.id})")
try:
if message.text == "Бан (Список)":
self.get_last_users(message)
@@ -235,11 +274,16 @@ class TelegramHelperBot:
self.get_banned_users(message)
elif message.text == "Вернуться в бота":
self.start_message(message)
else:
self.logger.warning(f"Неизвестное сообщение от админа: {message.text}")
self.bot.reply_to(message, "Неизвестная команда.")
except Exception as e:
self.logger.error(f"Ошибка при обработке сообщения админа: {e}")
self.bot.reply_to(message, f"Ошибка\n\n {e}")
self.admin_panel(message)
def ban_user(self, message, user_id: int):
self.logger.info(f"Получена команда от админа на бан пользователя: {message.text} (пользователь: {message.from_user.id})")
user_name = self.BotDB.get_username(user_id=user_id)
ban_object = {'user_id': user_id, 'user_name': user_name, 'message_for_user': None, 'date_to_unban': None}
markup = types.ReplyKeyboardMarkup(resize_keyboard=True, one_time_keyboard=True)
@@ -252,6 +296,7 @@ class TelegramHelperBot:
self.bot.register_next_step_handler(message, self.ban_user_step_2, ban_object)
def ban_user_step_2(self, message, ban_object: dict):
self.logger.info(f"Переход на шаг 2 бана пользователя. Словарь с данными для бана: {ban_object})")
ban_object['message_for_user'] = message.text
markup = types.ReplyKeyboardMarkup(resize_keyboard=True, one_time_keyboard=True)
item1 = types.KeyboardButton("1")
@@ -265,6 +310,7 @@ class TelegramHelperBot:
self.bot.register_next_step_handler(message, self.ban_user_step_3, ban_object)
def ban_user_step_3(self, message, ban_object: dict):
self.logger.info(f"Переход на шаг 3 бана пользователя. Словарь с данными для бана: {ban_object})")
date_to_unban = None
if message.text != 'Навсегда':
date_to_unban = self.add_days_to_date(message.text)
@@ -282,10 +328,12 @@ class TelegramHelperBot:
self.bot.register_next_step_handler(message, self.ban_user_final_step, ban_object)
def ban_user_final_step(self, message, ban_object: dict):
self.logger.info(f"Переход на финальный шаг бана пользователя. Словарь с данными для бана: {ban_object})")
if message.text == 'Подтвердить':
exists = self.BotDB.check_user_in_blacklist(ban_object['user_id'])
if exists:
self.bot.reply_to(message, f"Пользователь уже был заблокирован ранее.")
self.logger.info(f"Пользователь: {ban_object['user_id']} был заблокирован ранее)")
self.admin_panel(message)
else:
self.BotDB.set_user_blacklist(ban_object['user_id'],
@@ -293,15 +341,18 @@ class TelegramHelperBot:
ban_object['message_for_user'],
ban_object['date_to_unban'])
self.bot.reply_to(message, f"Пользователь {ban_object['user_name']} успешно заблокирован.")
self.logger.info(f"Пользователь: {ban_object['user_id']} успешно заблокирован)")
self.admin_panel(message)
def get_last_users(self, message):
self.logger.info(f"Попытка получения списка последних пользователей. Текст сообщения: {message.text} Имя автора сообщения: {message.from_user.full_name})")
list_users = self.BotDB.get_last_users_from_db()
keyboard = self.create_keyboard_with_pagination(1, len(list_users), list_users, 'ban')
self.bot.send_message(chat_id=message.chat.id, text="Список пользователей которые последними обращались к боту",
reply_markup=keyboard)
def get_banned_users(self, message):
self.logger.info(f"Попытка получения списка заблокированных пользователей. Текст сообщения: {message.text} Имя автора сообщения: {message.from_user.full_name})")
message_text = self.get_banned_users_list(0)
buttons_list = self.get_banned_users_buttons()
if buttons_list:
@@ -313,8 +364,10 @@ class TelegramHelperBot:
def start_message(self, message):
try:
self.logger.info(f"Формирование приветственного сообщения для пользователя. Сообщение: {message.text} Имя автора сообщения: {message.from_user.full_name})")
name_stick_hello = list(Path('Stick').rglob('Hello_*'))
random_stick_hello = open(random.choice(name_stick_hello), 'rb')
self.logger.info(f"Стикер успешно получен из БД")
# logging
if self.LOGS:
self.bot.forward_message(chat_id=self.GROUP_FOR_LOGS,
@@ -323,12 +376,13 @@ class TelegramHelperBot:
self.bot.send_sticker(message.chat.id, random_stick_hello)
sleep(0.3)
except Exception as e:
print(f'{str(e)}')
self.logger.error(f"Произошла ошибка при получении стикера. Ошибка: {str(e)}")
if self.LOGS:
self.bot.send_message(self.IMPORTANT_LOGS,
f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
try:
self.logger.info(f"Получение данных для приветственного сообщения пользователю. Сообщение: {message.text} Имя автора сообщения: {message.from_user.full_name})")
user_id = message.from_user.id
first_name = message.from_user.first_name
full_name = message.from_user.full_name
@@ -346,12 +400,13 @@ class TelegramHelperBot:
self.bot.send_message(message.chat.id, hello_message, parse_mode='html', reply_markup=markup,
disable_web_page_preview=not self.PREVIEW_LINK)
except Exception as e:
print(f'{str(e)}')
self.logger.error(f"Произошла ошибка при отправке приветственного сообщения для пользователя {message.from_user.id} Имя: {message.from_user.full_name}. Ошибка: {str(e)}")
if self.LOGS:
self.bot.send_message(self.IMPORTANT_LOGS,
f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
def resend_message_in_group_for_message(self, message):
self.logger.info(f"Попытка пересылки сообщения в связь с админами. Сообщение: {message.text} Имя автора сообщения: {message.from_user.full_name} Идентификатор сообщения: {message.id})")
self.bot.forward_message(chat_id=self.GROUP_FOR_MESSAGE,
from_chat_id=message.chat.id,
message_id=message.message_id
@@ -366,6 +421,7 @@ class TelegramHelperBot:
def suggest_post(self, message):
try:
self.logger.info(f"Вызов функции suggest_post. Сообщение: {message.text} Имя автора сообщения: {message.from_user.full_name} Идентификатор сообщения: {message.id})")
markup = types.ReplyKeyboardRemove()
suggest_news = messages.get_message(self.__get_first_name(message), 'SUGGEST_NEWS')
self.bot.send_message(message.chat.id, suggest_news, parse_mode='html')
@@ -375,13 +431,9 @@ class TelegramHelperBot:
except Exception as e:
self.bot.send_message(self.IMPORTANT_LOGS,
f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
# logging
if self.LOGS:
self.bot.forward_message(chat_id=self.GROUP_FOR_LOGS,
from_chat_id=message.chat.id,
message_id=message.message_id)
def stickers(self, message):
self.logger.info(f"Вызов функции stickers. Пользователь: {message.from_user.id} Имя автора сообщения: {message.from_user.full_name}")
self.BotDB.update_info_about_stickers(user_id=message.from_user.id)
markup = self.get_reply_keyboard(message)
try:
@@ -394,8 +446,10 @@ class TelegramHelperBot:
except ApiTelegramException as e:
self.bot.send_message(chat_id=self.IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
self.logger.error(f"Ошибка функции stickers. Ошибка: {str(e)} Пользователь: {message.from_user.id} Имя автора сообщения: {message.from_user.full_name}")
def connect_with_admin(self, message):
self.logger.info(f"Вызов функции connect_with_admin. Пользователь: {message.from_user.id} Имя автора сообщения: {message.from_user.full_name}")
connect_with_admin = messages.get_message(self.__get_first_name(message), 'CONNECT_WITH_ADMIN')
self.bot.send_message(message.chat.id, connect_with_admin, parse_mode="html")
# logging
@@ -406,10 +460,12 @@ class TelegramHelperBot:
def end_message(self, message):
try:
self.logger.info(f"Вызов функции end_message. Пользователь: {message.from_user.id} Имя автора сообщения: {message.from_user.full_name}")
name_stick_bye = list(Path('Stick').rglob('Universal_*'))
random_stick_bye = open(random.choice(name_stick_bye), 'rb')
self.bot.send_sticker(message.chat.id, random_stick_bye)
except ApiTelegramException as e:
self.logger.error(f"Ошибка в функции stickers при получении стикера: {str(e)} Пользователь: {message.from_user.id} Имя автора сообщения: {message.from_user.full_name}")
if self.LOGS:
self.bot.send_message(chat_id=self.IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
@@ -420,13 +476,9 @@ class TelegramHelperBot:
parse_mode='html', reply_markup=markup, disable_web_page_preview=not self.PREVIEW_LINK)
except Exception as e:
if self.LOGS:
self.logger.error(f"Ошибка в функции stickers при получении сообщения: {str(e)} Пользователь: {message.from_user.id} Имя автора сообщения: {message.from_user.full_name}")
self.bot.send_message(chat_id=self.IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
if self.LOGS:
# logging
self.bot.forward_message(chat_id=self.GROUP_FOR_LOGS,
from_chat_id=message.chat.id,
message_id=message.message_id)
def send_to_suggest(self, message):
markup = types.InlineKeyboardMarkup(row_width=1)
@@ -483,7 +535,7 @@ class TelegramHelperBot:
photo=message.photo[-1].file_id,
reply_markup=markup
)
# TODO: Не понятна реализация с альбомами от слова совсем
# TODO: Не понятна реализация с альбомами от слова совсем. 11.07 Реализация понятна. Нужно отправлять альбомы + сообщение в одном сообщении. Следующее сообщение с пробелом и кнопками
# elif message.content_type == 'photo' and message.media_group_id != None:
# bot.forward_message(chat_id=IMPORTANT_LOGS, from_chat_id=message.chat.id, message_id=message.message_id )
else:
@@ -503,7 +555,6 @@ class TelegramHelperBot:
item1 = types.KeyboardButton("📢Предложить свой пост")
item2 = types.KeyboardButton("📩Связаться с админами")
item3 = types.KeyboardButton("👋🏼Сказать пока!")
#TODO: Есть ощущение что не совсем так работает как надо
item4 = types.KeyboardButton("🤪Хочу стикеры") if not self.BotDB.get_info_about_stickers(
user_id=message.from_user.id) else None