bot.Bot._get_users_names() - Code Metrics - Inspection of "Update toctree." - dadyarri/ralph - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( ad3b9e...b48730 )

by Daniil

created 2020-02-24 12:47 UTC

bot.Bot._get_users_names() A

↳ Parent: bot

Complexity

Conditions

Size

Total Lines	12
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	7
nop	2
dl	0
loc	12
rs	10
c	0
b	0
f	0

"""
:project: ralph
:version: see VERSION.txt
:authors: dadyarri, 6a16ec
:contact: https://vk.me/dadyarri, https://vk.me/6a16ec
:license: MIT

:copyright: (c) 2019 - 2020 dadyarri, 6a16ec

Info about logging levels:

DEBUG: Detailed information, typically of interest only when diagnosing problems.

INFO: Confirmation that things are working as expected.

WARNING: An indication that something unexpected happened, or indicative of some
problem in the near future (e.g. ‘disk space low’). The software is still working
as expected.

ERROR: Due to a more serious problem, the software has not been able to perform
some function.

CRITICAL: A serious error, indicating that the program itself may be unable to
continue running.

"""


import os
import random
from typing import List
from typing import NoReturn

import requests
import vk_api
from vk_api.bot_longpoll import VkBotEventType

from database import Database
from keyboard import Keyboards
from logger import Logger
from vkbotlongpoll import RalphVkBotLongPoll


class Bot:
    """Класс, описывающий объект бота, включая авторизацию в API, и все методы бота.
    
    Attributes:
        log (Logger): Объект класса Logger, содержащий настройки модуля logging
        token (str): Токен доступа к сообществу ВКонтакте
        user_token (str): Токен доступа пользователя-администратора ВКонтакте (используется для обновления статуса сообщества)
        gid (str): Идентификатор сообщества ВКонтакте
        cid (str): Идентификатор активной беседы (используется в рассылке расписания)
        kbs (Keyboards): Объект класса Keyboards, содержащий генераторы клавиатур
        db (Database): Объект класса Database, инициирующий подключение к базе данных
        admins (List[str]): Список идентификаторов ВКонтакте пользователей, имеющих доступ администратора бота 
    
    Example:
        .. code-block:: python
        
            from bot import Bot
        
            bot = Bot()
        
    Todo:
        * Почистить атрибуты
    """

    def __init__(self) -> None:

        self.log = Logger()

        self.log.log.info("Инициализация...")

        self.token = os.environ["VK_TOKEN"]
        self.user_token = os.environ["VK_USER_TOKEN"]
        self.gid = os.environ["GID_ID"]
        self.cid = os.environ["CID_ID"]

        self.kbs = Keyboards()

        self.db = Database(os.environ["DATABASE_URL"])

        self.log.log.info("Авторизация ВКонтакте...")
        try:
            self.bot_session = vk_api.VkApi(token=self.token, api_version="5.103")
            self.user_session = vk_api.VkApi(token=self.user_token, api_version="5.103")
        except vk_api.exceptions.AuthError:
            self.log.log.error("Неудача. Ошибка авторизации.")
        else:
            try:
                self.bot_vk = self.bot_session.get_api()
                self.user_vk = self.user_session.get_api()
                self.longpoll = RalphVkBotLongPoll(
                    vk=self.bot_session, group_id=self.gid
                )
            except requests.exceptions.ConnectionError:
                self.log.log.error("Неудача. Превышен лимит попыток подключения.")
            except vk_api.exceptions.ApiError:
                self.log.log.error("Неудача. Ошибка доступа.")
            else:
                self.log.log.info("Успех.")
                self.log.log.debug(
                    f"Версия API ВКонтакте: {self.bot_session.api_version}."
                )

        # Инициализация дополнительных переменных
        self.event = {}
        self.admins = os.environ["ADMINS_IDS"].split(",")

        # Переименование обрабатываемых типов событий
        self.NEW_MESSAGE = VkBotEventType.MESSAGE_NEW

        self.log.log.info(
            f"Беседа... {'Тестовая' if self.cid.endswith('1') else 'Основная'}"
        )

        self.log.log.info("Инициализация завершена.")

    def send_message(
        self,
        msg: str,
        pid: int = None,
        keyboard: str = "",
        attachments: str = None,
        user_ids: str = None,
        forward: str = "",
    ) -> NoReturn:

        """Обёртка над API ВКонтакте, отправляющая сообщения
        
        Arguments:
            msg: Текст отправляемого сообщения
            pid: Идентификатор пользователя/беседы/сообщества получателя сообщения (*не нужен, если указан user_ids*)
            keyboard: JSON-подобная строка со встроенной клавиатурой
            attachments: Вложения к сообщению (**не работает**)
            user_ids: Перечень адресатов для отправки одного сообщения (*не нужен, если указан pid*)
            forward: Перечень идентификаторов сообщений для пересылки
        """

        try:
            self.bot_vk.messages.send(
                peer_id=pid,
                random_id=random.getrandbits(64),
                message=msg,
                keyboard=keyboard,
                attachments=attachments,
                user_ids=user_ids,
                forward_messages=forward,
            )

        except vk_api.exceptions.ApiError as e:
            self.log.log.error(msg=e.__str__())
        except FileNotFoundError as e:
            self.log.log.error(msg=e)

    def send_mailing(self, ids: str, msg: str = "") -> NoReturn:
        """Отправка рассылки
        
        Arguments:
            ids: Список идентификаторов пользователей-получателей рассылки
            msg: Сообщение рассылки
            
        Todo:
            Удалить и заменить на Bot.send_message
        
        """
        self.send_message(msg=msg, user_ids=ids)

    def get_users_names(self, ids: list) -> List[str]:
        """Получает имена пользователей  из базы данных по идентификаторам из списка
        
        Arguments:
            ids: Список идентификаторов для формирования списка имён
        
        Returns:
            List[str]: Список имён пользователей
        """
        user_ids = [
            self.db.query(f"SELECT id FROM users WHERE vk_id={i}")[0][0] for i in ids
        ]
        user_names = [
            self.db.query(f"SELECT first_name FROM users_info WHERE user_id={i}")[0][0]
            for i in user_ids
        ]
        return user_names

    def generate_mentions(self, ids: str, names: bool) -> str:
        """Генерирует строку с упоминаниями из списка идентификаторов
        
        Arguments:
            ids: Перечень идентификаторов пользователей
            names: Флаг, указывающий на необходимость использования имён
            
        Returns:
            str: Сообщение, упоминающее выбранных пользователей
        """
        ids = ids.split(",")
        if names:
            users_names = self.get_users_names(ids)
        else:
            users_names = ["!" for i in range(len(ids))]
        result = (", " if names else "").join(
            [f"@id{_id}({users_names[i]})" for (i, _id) in enumerate(ids)]
        )
        return result

    def current_is_admin(self) -> bool:
        """Проверяет, является ли текущий пользователь администратором бота
        
        Returns:
            bool: Флаг, указывающий на принадлежность текущего пользователя к касте Администраторов
            
        Todo:
            Сделать метод более общим
        """
        return str(self.event.object.from_id) in self.admins

    def send_gui(self, text: str = "Привет!") -> NoReturn:
        """Отправляет клавиатуру главного меню
        
        Arguments:
            text: Сообщение, вместе с которым будет отправлена клавиатура
        """
        self.send_message(
            msg=text,
            pid=self.event.object.from_id,
            keyboard=self.kbs.generate_main_menu(self.current_is_admin()),
        )

    def update_version(self):
        """
        Обновляет версию в статусе группы с ботом
        """
        self.log.log.info("Обновление версии в статусе группы...")
        try:
            with open("VERSION.txt", "r") as f:
                v = f"Версия: {f.read()}"
            self.user_vk.status.set(text=v, group_id=self.gid)
        except vk_api.exceptions.ApiError as e:
            self.log.log.error(f"Ошибка {e.__str__()}")
        else:
            self.log.log.info(f"Успех.")


1			"""
2			:project: ralph
3			:version: see VERSION.txt
4			:authors: dadyarri, 6a16ec
5			:contact: https://vk.me/dadyarri, https://vk.me/6a16ec
6			:license: MIT
7
8			:copyright: (c) 2019 - 2020 dadyarri, 6a16ec
9
10			Info about logging levels:
11
12			DEBUG: Detailed information, typically of interest only when diagnosing problems.
13
14			INFO: Confirmation that things are working as expected.
15
16			WARNING: An indication that something unexpected happened, or indicative of some
17			problem in the near future (e.g. ‘disk space low’). The software is still working
18			as expected.
19
20			ERROR: Due to a more serious problem, the software has not been able to perform
21			some function.
22
23			CRITICAL: A serious error, indicating that the program itself may be unable to
24			continue running.
25
26			"""
27
28
29			import os
30			import random
31			from typing import List
32			from typing import NoReturn
33
34			import requests
35			import vk_api
36			from vk_api.bot_longpoll import VkBotEventType
37
38			from database import Database
39			from keyboard import Keyboards
40			from logger import Logger
41			from vkbotlongpoll import RalphVkBotLongPoll
42
43
44			class Bot:
45			"""Класс, описывающий объект бота, включая авторизацию в API, и все методы бота.
46
47			Attributes:
48			log (Logger): Объект класса Logger, содержащий настройки модуля logging
49			token (str): Токен доступа к сообществу ВКонтакте
50			user_token (str): Токен доступа пользователя-администратора ВКонтакте (используется для обновления статуса сообщества)
51			gid (str): Идентификатор сообщества ВКонтакте
52			cid (str): Идентификатор активной беседы (используется в рассылке расписания)
53			kbs (Keyboards): Объект класса Keyboards, содержащий генераторы клавиатур
54			db (Database): Объект класса Database, инициирующий подключение к базе данных
55			admins (List[str]): Список идентификаторов ВКонтакте пользователей, имеющих доступ администратора бота
56
57			Example:
58			.. code-block:: python
59
60			from bot import Bot
61
62			bot = Bot()
63
64			Todo:
65			* Почистить атрибуты
66			"""
67
68			def __init__(self) -> None:
69
70			self.log = Logger()
71
72			self.log.log.info("Инициализация...")
73
74			self.token = os.environ["VK_TOKEN"]
75			self.user_token = os.environ["VK_USER_TOKEN"]
76			self.gid = os.environ["GID_ID"]
77			self.cid = os.environ["CID_ID"]
78
79			self.kbs = Keyboards()
80
81			self.db = Database(os.environ["DATABASE_URL"])
82
83			self.log.log.info("Авторизация ВКонтакте...")
84			try:
85			self.bot_session = vk_api.VkApi(token=self.token, api_version="5.103")
86			self.user_session = vk_api.VkApi(token=self.user_token, api_version="5.103")
87			except vk_api.exceptions.AuthError:
88			self.log.log.error("Неудача. Ошибка авторизации.")
89			else:
90			try:
91			self.bot_vk = self.bot_session.get_api()
92			self.user_vk = self.user_session.get_api()
93			self.longpoll = RalphVkBotLongPoll(
94			vk=self.bot_session, group_id=self.gid
95			)
96			except requests.exceptions.ConnectionError:
97			self.log.log.error("Неудача. Превышен лимит попыток подключения.")
98			except vk_api.exceptions.ApiError:
99			self.log.log.error("Неудача. Ошибка доступа.")
100			else:
101			self.log.log.info("Успех.")
102			self.log.log.debug(
103			f"Версия API ВКонтакте: {self.bot_session.api_version}."
104			)
105
106			# Инициализация дополнительных переменных
107			self.event = {}
108			self.admins = os.environ["ADMINS_IDS"].split(",")
109
110			# Переименование обрабатываемых типов событий
111			self.NEW_MESSAGE = VkBotEventType.MESSAGE_NEW
112
113			self.log.log.info(
114			f"Беседа... {'Тестовая' if self.cid.endswith('1') else 'Основная'}"
115			)
116
117			self.log.log.info("Инициализация завершена.")
118
119			def send_message(
120			self,
121			msg: str,
122			pid: int = None,
123			keyboard: str = "",
124			attachments: str = None,
125			user_ids: str = None,
126			forward: str = "",
127			) -> NoReturn:
128
129			"""Обёртка над API ВКонтакте, отправляющая сообщения
130
131			Arguments:
132			msg: Текст отправляемого сообщения
133			pid: Идентификатор пользователя/беседы/сообщества получателя сообщения (не нужен, если указан user_ids)
134			keyboard: JSON-подобная строка со встроенной клавиатурой
135			attachments: Вложения к сообщению (не работает)
136			user_ids: Перечень адресатов для отправки одного сообщения (не нужен, если указан pid)
137			forward: Перечень идентификаторов сообщений для пересылки
138			"""
139
140			try:
141			self.bot_vk.messages.send(
142			peer_id=pid,
143			random_id=random.getrandbits(64),
144			message=msg,
145			keyboard=keyboard,
146			attachments=attachments,
147			user_ids=user_ids,
148			forward_messages=forward,
149			)
150
151			except vk_api.exceptions.ApiError as e:
152			self.log.log.error(msg=e.__str__())
153			except FileNotFoundError as e:
154			self.log.log.error(msg=e)
155
156			def send_mailing(self, ids: str, msg: str = "") -> NoReturn:
157			"""Отправка рассылки
158
159			Arguments:
160			ids: Список идентификаторов пользователей-получателей рассылки
161			msg: Сообщение рассылки
162
163			Todo:
164			Удалить и заменить на Bot.send_message
165
166			"""
167			self.send_message(msg=msg, user_ids=ids)
168
169			def get_users_names(self, ids: list) -> List[str]:
170			"""Получает имена пользователей из базы данных по идентификаторам из списка
171
172			Arguments:
173			ids: Список идентификаторов для формирования списка имён
174
175			Returns:
176			List[str]: Список имён пользователей
177			"""
178			user_ids = [
179			self.db.query(f"SELECT id FROM users WHERE vk_id={i}")[0][0] for i in ids
180			]
181			user_names = [
182			self.db.query(f"SELECT first_name FROM users_info WHERE user_id={i}")[0][0]
183			for i in user_ids
184			]
185			return user_names
186
187			def generate_mentions(self, ids: str, names: bool) -> str:
188			"""Генерирует строку с упоминаниями из списка идентификаторов
189
190			Arguments:
191			ids: Перечень идентификаторов пользователей
192			names: Флаг, указывающий на необходимость использования имён
193
194			Returns:
195			str: Сообщение, упоминающее выбранных пользователей
196			"""
197			ids = ids.split(",")
198			if names:
199			users_names = self.get_users_names(ids)
200			else:
201			users_names = ["!" for i in range(len(ids))]
202			result = (", " if names else "").join(
203			[f"@id{_id}({users_names[i]})" for (i, _id) in enumerate(ids)]
204			)
205			return result
206
207			def current_is_admin(self) -> bool:
208			"""Проверяет, является ли текущий пользователь администратором бота
209
210			Returns:
211			bool: Флаг, указывающий на принадлежность текущего пользователя к касте Администраторов
212
213			Todo:
214			Сделать метод более общим
215			"""
216			return str(self.event.object.from_id) in self.admins
217
218			def send_gui(self, text: str = "Привет!") -> NoReturn:
219			"""Отправляет клавиатуру главного меню
220
221			Arguments:
222			text: Сообщение, вместе с которым будет отправлена клавиатура
223			"""
224			self.send_message(
225			msg=text,
226			pid=self.event.object.from_id,
227			keyboard=self.kbs.generate_main_menu(self.current_is_admin()),
228			)
229
230			def update_version(self):
231			"""
232			Обновляет версию в статусе группы с ботом
233			"""
234			self.log.log.info("Обновление версии в статусе группы...")
235			try:
236			with open("VERSION.txt", "r") as f:
237			v = f"Версия: {f.read()}"
238			self.user_vk.status.set(text=v, group_id=self.gid)
239			except vk_api.exceptions.ApiError as e:
240			self.log.log.error(f"Ошибка {e.__str__()}")
241			else:
242			self.log.log.info(f"Успех.")
243

dadyarri / ralph

Push — master ( ad3b9e...b48730 )

bot.Bot._get_users_names() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like