add TO MUCH callbacks

This commit is contained in:
KatykhinAA
2024-07-05 01:27:35 +03:00
parent 1aed0c0c4f
commit aded540d82
2 changed files with 144 additions and 20 deletions

25
db.py
View File

@@ -313,13 +313,30 @@ class BotDB:
except sqlite3.Error as error: except sqlite3.Error as error:
print(error) print(error)
def get_last_users_from_bot(self): def get_last_users_from_db(self):
"""Возвращает список идентификаторов последних 10 пользователей обращавшихся в бот""" """Возвращает список идентификаторов последних 10 пользователей обращавшихся в бот"""
try: try:
result = self.cursor.execute("SELECT full_name, user_id FROM our_users ORDER BY date_changed DESC LIMIT 10") result = self.cursor.execute("SELECT full_name, user_id FROM our_users ORDER BY date_changed DESC")
users = result.fetchall() users = result.fetchall()
user_dict = {user[1]: user[0] for user in users} return users
return user_dict except sqlite3.Error as error:
print(error)
def get_banned_users_from_db(self):
"""Возвращает список идентификаторов последних 10 пользователей обращавшихся в бот"""
try:
result = self.cursor.execute("SELECT user_name, user_id, message_for_user, date_to_unban FROM blacklist", )
users = result.fetchall()
return users
except sqlite3.Error as error:
print(error)
def get_banned_users_from_db_with_limits(self, offset: int, limit: int):
"""Возвращает список идентификаторов последних 10 пользователей обращавшихся в бот"""
try:
result = self.cursor.execute("SELECT user_name, user_id, message_for_user, date_to_unban FROM blacklist LIMIT ?, ?", (offset, limit,) )
users = result.fetchall()
return users
except sqlite3.Error as error: except sqlite3.Error as error:
print(error) print(error)

139
main.py
View File

@@ -4,6 +4,8 @@ import sys
from pathlib import Path from pathlib import Path
from time import sleep from time import sleep
from enum import Enum from enum import Enum
from typing import Any
import db import db
from db import BotDB from db import BotDB
import telebot import telebot
@@ -35,8 +37,9 @@ TEST = config.getboolean('Settings', 'test')
BotDB = BotDB('tg-bot-database') BotDB = BotDB('tg-bot-database')
#TODO: state хранить в базе #TODO: Дописать функцию разбана и подключить ее к обработке коллбеков
#TODO: Перенести обработчик коллбэков #TODO: Подумать что сделать с процессом наступления даты разбана
#TODO: Сделать функционал игнора забаненных
class State(Enum): class State(Enum):
START = "START" START = "START"
@@ -121,7 +124,7 @@ class TelegramHelperBot:
self.bot.send_message(chat_id, message_from_admin, reply_markup=markup) self.bot.send_message(chat_id, message_from_admin, reply_markup=markup)
# Админка # Админка
@self.bot.callback_query_handler(func=lambda call: True) @self.bot.callback_query_handler(func=lambda call: call.data in ['publish', 'decline'])
def post_for_group(call): def post_for_group(call):
if call.data == 'publish' and call.message.content_type == 'text': if call.data == 'publish' and call.message.content_type == 'text':
try: try:
@@ -150,9 +153,29 @@ class TelegramHelperBot:
if LOGS: if LOGS:
self.bot.send_message(IMPORTANT_LOGS, self.bot.send_message(IMPORTANT_LOGS,
f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}") f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
elif call.data[:3] == 'ban':
@self.bot.callback_query_handler(func=lambda call: True)
def pagination(call):
if call.data[:3] == 'ban':
user_id = call.data[4:] user_id = call.data[4:]
self.ban_user(call.message, user_id) self.ban_user(call.message, user_id)
if call.data[:5] == 'unban':
#TODO: Тут обрабатываем события, ДОПИСАТЬ!!!!
print(f'UNBAN NAXUY , {call.data[6:]}')
elif call.data[:4] == 'page':
if call.message.text == 'Список пользователей которые последними обращались к боту':
list_users = BotDB.get_last_users_from_db()
keyboard = self.create_keyboard_with_pagination(int(call.data[5:]), 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():
#Готовим сообщения
message_user = self.get_banned_users_list(int(call.data[5:])*7-7)
self.bot.edit_message_text(chat_id=call.message.chat.id, message_id=call.message.message_id, text=message_user)
#Готовим клавиатуру
buttons = self.get_banned_users_buttons()
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)
def register_chat_handler(self, message): def register_chat_handler(self, message):
self.bot.register_next_step_handler(message, self.resend_message_in_group_for_message) self.bot.register_next_step_handler(message, self.resend_message_in_group_for_message)
@@ -171,7 +194,8 @@ class TelegramHelperBot:
item1 = types.KeyboardButton("Бан (Список)") item1 = types.KeyboardButton("Бан (Список)")
item2 = types.KeyboardButton("Добавить админа") item2 = types.KeyboardButton("Добавить админа")
item3 = types.KeyboardButton("Удалить админа") item3 = types.KeyboardButton("Удалить админа")
markup.add(item1, item2, item3) item4 = types.KeyboardButton("Разбан (список)")
markup.add(item1, item2, item3, item4)
self.bot.send_message(message.chat.id, "Добро пожаловать в админку. Выбери что хочешь:", self.bot.send_message(message.chat.id, "Добро пожаловать в админку. Выбери что хочешь:",
reply_markup=markup) reply_markup=markup)
self.bot.register_next_step_handler(message, self.handle_admin_message) self.bot.register_next_step_handler(message, self.handle_admin_message)
@@ -180,8 +204,10 @@ class TelegramHelperBot:
def handle_admin_message(self, message): def handle_admin_message(self, message):
try: try:
if message.text == "Забанить пользователя из списка": if message.text == "Бан (Список)":
self.get_last_users(message) self.get_last_users(message)
elif message.text == "Разбан (список)":
self.get_banned_users(message)
except Exception as e: except Exception as e:
self.bot.reply_to(message, f"Ошибка\n\n {e}") self.bot.reply_to(message, f"Ошибка\n\n {e}")
self.admin_panel(message) self.admin_panel(message)
@@ -243,20 +269,21 @@ class TelegramHelperBot:
self.bot.reply_to(message, f"Пользователь {ban_object['user_name']} успешно заблокирован.") self.bot.reply_to(message, f"Пользователь {ban_object['user_name']} успешно заблокирован.")
self.admin_panel(message) self.admin_panel(message)
else: else:
self.bot.reply_to(message, f"Произошла ошибка при блокировке пользователя.") #TODO: тут какой-то баг. Отвечает что ошибка, но ошибок None
self.bot.reply_to(message, f"Произошла ошибка при блокировке пользователя. ERROR: {result}")
self.admin_panel(message) self.admin_panel(message)
def get_last_users(self, message): def get_last_users(self, message):
list_users = BotDB.get_last_users_from_bot() list_users = BotDB.get_last_users_from_db()
markup = types.InlineKeyboardMarkup(row_width=1) keyboard = self.create_keyboard_with_pagination(1, len(list_users), list_users, 'ban')
for user_id, full_name in list_users.items():
button = types.InlineKeyboardButton(
f"{full_name}", callback_data=f'ban_{user_id}'
)
# Добавляем кнопки на клавиатуру
markup.add(button)
self.bot.send_message(chat_id=message.chat.id, text="Список пользователей которые последними обращались к боту", self.bot.send_message(chat_id=message.chat.id, text="Список пользователей которые последними обращались к боту",
reply_markup=markup) reply_markup=keyboard)
def get_banned_users(self, message):
message_text = self.get_banned_users_list(0)
buttons_list = self.get_banned_users_buttons()
k = self.create_keyboard_with_pagination(1, len(buttons_list), buttons_list, 'unban')
self.bot.send_message(message.chat.id, message_text, reply_markup=k)
def start_message(self, message): def start_message(self, message):
try: try:
@@ -481,6 +508,86 @@ class TelegramHelperBot:
formatted_date = future_date.strftime("%d-%m-%Y") formatted_date = future_date.strftime("%d-%m-%Y")
return formatted_date return formatted_date
@staticmethod
def create_keyboard_with_pagination(page: int, total_items: int, array_items: list[tuple[Any, Any]], callback: str):
"""
Создает клавиатуру с пагинацией для заданного набора элементов и устанавливает необходимый callback
Args:
page: Номер текущей страницы.
total_items: Общее количество элементов.
array_items: Лист кортежей. Содержит в себе user_name: user_id
callback: Действие в коллбеке. Вернет callback вида ({callback}_{user_id})
Returns:
InlineKeyboardMarkup: Клавиатура с кнопками пагинации.
"""
# Определяем общее количество страниц
total_pages = (total_items + 7 - 1) // 7
page = page
# Создаем список кнопок
buttons = []
# Вычисляем стартовый номер для текущей страницы
start_index = (page - 1) * 7 #тут было +1, убрал, потому что на текстовом массиве выходит за пределы
# Кнопки с номерами страниц
for i in range(start_index, min(start_index + 7, len(array_items))): #тут было len(array_items) +1, убрал, потому что на текстовом массиве выходит за пределы
buttons.append(types.InlineKeyboardButton(f"{array_items[i][0]}", callback_data=f"{callback}_{array_items[i][1]}"))
# Добавляем кнопки "Предыдущая" и "Следующая"
if int(page) > 1:
buttons.insert(6, types.InlineKeyboardButton("⬅️ Предыдущая", callback_data=f"page_{page-1}"))
if page < total_pages:
buttons.append(types.InlineKeyboardButton("➡️ Следующая", callback_data=f"page_{page+1}"))
# Формируем клавиатуру с 3 кнопками в ряд
keyboard = []
for i in range(0, len(buttons), 3):
keyboard.append(buttons[i:i+3])
return types.InlineKeyboardMarkup(keyboard)
@staticmethod
def get_banned_users_list(offset: int):
"""
Возвращает сообщение со списком пользователей и словарь с ником + идентификатором
Args:
offset: отступ для запроса в базу данных
Returns:
message - текст сообщения
user_ids - лист кортежей [(user_name: user_id)]
"""
users = BotDB.get_banned_users_from_db_with_limits(limit=7, offset=offset)
message = "Список заблокированных пользователей:\n"
for user in users:
message += f"Пользователь: {user[0]}\n"
message += f"Причина бана: {user[2]}\n"
message += f"Дата разбана: {user[3]}\n\n"
return message
@staticmethod
def get_banned_users_buttons():
"""
Возвращает сообщение со списком пользователей и словарь с ником + идентификатором
Args:
offset: отступ для запроса в базу данных
Returns:
message - текст сообщения
user_ids - лист кортежей [(user_name: user_id)]
"""
users = BotDB.get_banned_users_from_db()
user_ids = []
for user in users:
user_ids.append((user[0], user[1]))
return user_ids
bot = TelegramHelperBot(BOT_TOKEN) bot = TelegramHelperBot(BOT_TOKEN)