telegram-helper-bot/db.py

import sqlite3
import configparser
import os
import sys
import random

config_path = os.path.join(sys.path[0], 'settings.ini')
config = configparser.ConfigParser()
config.read(config_path)
LOGS = config.getboolean('Settings', 'logs')
IMPORTANT_LOGS = config.get('Telegram', 'important_logs')


class BotDB:

    def __init__(self, db_file):
        self.conn = sqlite3.connect(db_file, check_same_thread=False)
        self.cursor = self.conn.cursor()

    def get_message_from_db(self, type: str, username):
        """Функция для запроса к базе данных и получения сообщений для бота. В аргументы передаются:
        type - тип получаемой обратной связи, строковое значение, сохраненное в БД
        username - имя пользователя
        """
        # Подключаемся к базе
        try:
            cursor = self.conn.cursor()
            cursor.execute(f"SELECT * FROM messages WHERE type=?", (type,))
            # Забираем данные из таблицы, преобразуем в строку, заменяем поле username на имя пользователя,
            # и вместо амберсанда подставляем перенос строки
            if type == 'connect_with_admin' or type == 'del_message' or type == 'suggest_news' or type == 'start_message':
                response_from_database = str(cursor.fetchone()[1]).replace('username', username).replace('&', '\n')
            else:
                response_from_database = str(cursor.fetchone()[1]).replace('&', '\n')
            return response_from_database
        except sqlite3.Error as error:
            print(error)

    def get_error_message_from_db(self, id: int):
        """Функция для запроса к базе данных и получения сообщений ошибки. В аргументы передаются:
        id - идентификатор ошибки
        """
        # Подключаемся к базе
        try:
            cursor = self.conn.cursor()
            cursor.execute(f"SELECT * FROM error_messages WHERE id=?", (id,))
            response_from_database = str(cursor.fetchone()[1])
            return response_from_database
        except sqlite3.Error as error:
            print(error)

    def add_new_user_in_db(self, user_id, first_name, full_name, username, is_bot, language_code, date_added, date_changed):
        """Добавляем юзера в базу"""
        try:
            self.cursor.execute("INSERT INTO 'our_users' ('user_id', 'first_name', 'full_name', 'username', 'is_bot', "
                                "'language_code', 'date_added', 'date_changed') VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                                (user_id, first_name, full_name,
                                username,is_bot,language_code,date_added,date_changed))
            return self.conn.commit()
        except sqlite3.Error as error:
            print(error)


    def user_exists(self, user_id):
        """Проверяем, есть ли юзер в базе"""
        try:
            result = self.cursor.execute("SELECT `id` FROM `our_users` WHERE `user_id` = ?", (user_id,))
            return bool(len(result.fetchall()))
        except sqlite3.Error as error:
            print(error)


    def get_user_id(self, user_id):
        """Достаем id юзера в базе по его user_id"""
        try:
            result = self.cursor.execute("SELECT `id` FROM `our_users` WHERE `user_id` = ?", (user_id,))
            return result.fetchone()[0]
        except sqlite3.Error as error:
            print(error)


    def get_user_first_name(self, user_id):
        try:
            result = self.cursor.execute("SELECT `first_name` FROM `our_users` WHERE `user_id` = ?", (user_id,))
            return result.fetchone()[0]
        except sqlite3.Error as error:
            print(error)


    def change_name(self, user_id):
        #TODO: реализовать функцию изменения имени пользователя по которому к нему обращается бот. ОБновляем поля first_name, date_changed
        #result = self.cursor.execute("UPDATE 'our_users' SET (?) WHERE user_id = (?)", (new_user_name), )
        pass

    def add_audio_record(self, file_name, author_id, date_added, listen_count, file_id):
        """Добавляет информацию о войсе юзера в БД"""
        try:
            result = self.cursor.execute("INSERT INTO `audio_message_reference` (file_name, author_id, date_added, listen_count, file_id) VALUES (?, ?, ?, ?, ?)", (file_name, author_id, date_added, listen_count, file_id))
            return self.conn.commit()
        except sqlite3.Error as error:
            print(error)

    def get_last_user_audio_record(self, user_id):
        """Получает данные о количестве записей пользователя"""
        try:
            result = self.cursor.execute("SELECT `file_id` FROM `audio_message_reference` WHERE `author_id` = ?", (user_id,))
            return bool(len(result.fetchall()))
        except sqlite3.Error as error:
            print(error)

    def get_id_for_audio_record(self, user_id):
        """Получает ID аудио сообщения пользователя"""
        try:
            result = self.cursor.execute("SELECT `file_id` FROM `audio_message_reference` WHERE `author_id` = ? ORDER BY date_added DESC LIMIT 1",
                                         (user_id,))
            return result.fetchone()[0]
        except sqlite3.Error as error:
            print(error)

    def get_path_for_audio_record(self, user_id):
        """Получает данные о названии файла"""
        try:
            result = self.cursor.execute("SELECT `file_name` FROM `audio_message_reference` WHERE `author_id` = ? ORDER BY date_added DESC LIMIT 1", (user_id,))
            return result.fetchone()[0]
        except sqlite3.Error as error:
            print(error)

    def get_random_audio(self, user_id):
        """Получает данные о войсе юзера из БД"""
        try:
            file_name = self.cursor.execute(
                "SELECT `file_name` FROM `audio_message_reference` WHERE `author_id` <> ?", (user_id,))
            file_name_massive = file_name.fetchall()
            number_element = random.randint(0, len(file_name_massive)-1)
            audio_for_user = file_name_massive[number_element]
            return audio_for_user
        except sqlite3.Error as error:
            print(error)

    def close(self):
        """Закрываем соединение с БД"""
        self.conn.close()