Как стать автором
Обновить

Фильтр для телеграм-бота

В первой статье на Хабре хочу поделиться своим решением задачи с которой столкнулся, разрабатывая бота для Телеграм. Суть проблемы, заключалась в следующем, когда добавил своему "питомцу" функцию наделения участников группы правами администратора, он должен был оперативно реагировать на эти изменения. Но получалось так, что при запуске бот получал список администраторов, при помощи метода:

[admin.user.id for admin in await bot.get_chat_administrators(GROUP_ID)] 

Но чтобы он увидел нововведения в этом списке его надо было перезапускать.

Не найдя решения, как заставить бота заново пересоздать список без перезапуска, написал такой фильтр, работающий с базой данных:

DEFAULT_LIST = [OWNER_ID]] 
ADMINS_LIST = [] 

class AdminFilter(BoundFilter):
    key = 'admin'
    def __init__(self, admin):
        self.admin_test = admin   

    async def check(self, message: types.Message):
       ADMINS_LIST.clear()
       ADMINS_LIST.extend(DEFAULT_LIST)
       curs = conn.cursor()
       curs.execute(f"SELECT user_id FROM users WHERE admin = 'True'")
       rez = cur.fetchall()# получаем id пользователей с правом доступа из базы    
       conn.commit() 
       for q in result:
          w = q[0]     
          ADMINS_LIST.append(w)
          for q in ADMINS_LIST:
             if  message.from_user.id == q:
                return admin

Для примера взял простую SQLite, но может работать и другими.
Учёт администраторов ведётся по их ID.

В начале объявляем переменную:

admin = 'admin'

Создаём список по умолчанию, куда вносим ID владельца группы (можно кого-то ещё по усмотрению):

DEFAULT_LIST = [OWNER_ID] 

Определяем список админов:

ADMINS_LIST = []

Создаём класс AdminFilter, который наследуем от класса BoundFilter:

class AdminFilter(BoundFilter):

Классу фильтра присваиваем переменную key:

key = 'admin'

Запускаем конструктор класса (метод_ _init__ с необходимыми переменными):

 def __init__(self, admin):
        self.admin_test = admin

Далее в функции check() обращаемся к базе откуда получаем список админов:

async def check(self, message: types.Message):

Удаляем старый список админов:

ADMINS_LIST.clear()

В пустой ADMINS_LIST и добавляем в список по умолчанию:

ADMINS_LIST.extend(DEFAULT_LIST)

Делаем запрос к базе данных и получаем ID пользователей у которых в столбце admin стоит 'True'.

  curs = conn.cursor()
  curs.execute(f"SELECT user_id FROM users WHERE admin = 'True'")
  rez = cur.fetchall() 
  conn.commit()    

Пробегаемся по полученному списку, вынимаем чистые значения ID:

 for q in result:
      w = q[0] 

И вносим их в ADMINS_LIST:

ADMINS_LIST.append(w)

Берём ID пользователя из message.from_user.id. Пробегаемся по списку по администраторов и если он там есть функция возвращает переменную admin.

  for q in ADMINS_LIST:
        if message.from_user.id ==  q:
           return admin

Далее на стороне бота делается проверка (admin=True). И при положительном результате код выполняется.

Идея в том, что при каждой проверке создаётся новый список администраторов, который подтягивается из базы. Это замедляет работу бота, но зато теперь он оперативно (без перезапуска) видит у кого в группе какие полномочия.

Теги:
Хабы:
Данная статья не подлежит комментированию, поскольку её автор ещё не является полноправным участником сообщества. Вы сможете связаться с автором только после того, как он получит приглашение от кого-либо из участников сообщества. До этого момента его username будет скрыт псевдонимом.