database.py - master - Soldier327/Telegram_Bot_Russian_English

database.py
312 строк11 KB
Soldier327
create: config.py, database.py, handlers.py, keyboards.py, Readme.md, requirements.txt, Shema.png, Telegram_Bot.py, vocabulary.py
20 апр 2026, 21:31
Верифицирован
20 апр 2026, 21:314b09736
import sqlite3
import logging
from vocabulary import VOCABULARY

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class UserDatabase:
    def __init__(self, db_path='data/user_vocabulary.db'):
        self.db_path = db_path
        self.init_database()

    def init_database(self):
        """Инициализация базы данных"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()

            # Таблица пользователей
            cursor.execute('''
            CREATE TABLE IF NOT EXISTS users (
                user_id INTEGER PRIMARY KEY,
                username TEXT,
                first_name TEXT,
                registered_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                total_quizzes INTEGER DEFAULT 0,
                correct_answers INTEGER DEFAULT 0
            )
            ''')

            # Таблица общих слов (общий словарь)
            cursor.execute('''
            CREATE TABLE IF NOT EXISTS common_words (
                id INTEGER PRIMARY KEY,
                russian TEXT NOT NULL UNIQUE,
                translation TEXT NOT NULL
            )
            ''')

            # Таблица слов пользователя
            cursor.execute('''
            CREATE TABLE IF NOT EXISTS user_words (
                id INTEGER PRIMARY KEY,
                user_id INTEGER NOT NULL,
                russian TEXT NOT NULL,
                translation TEXT NOT NULL,
                correct_count INTEGER DEFAULT 0,
                wrong_count INTEGER DEFAULT 0,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                UNIQUE(user_id, russian),
                FOREIGN KEY (user_id) REFERENCES users(user_id)
                    ON DELETE CASCADE
            )
            ''')

            # Индексы для производительности
            cursor.execute('''
            CREATE INDEX IF NOT EXISTS idx_user_words_uid
            ON user_words (user_id)
            ''')

            cursor.execute('''
            CREATE INDEX IF NOT EXISTS idx_user_words_ru
            ON user_words (russian)
            ''')

            cursor.execute('''
            CREATE INDEX IF NOT EXISTS idx_common_words_ru
            ON common_words (russian)
            ''')

            conn.commit()

            # Заполняем таблицу common_words, если пуста
            self._init_common_words(cursor)

            logger.info("База данных инициализирована")

        except sqlite3.Error as e:
            logger.error(f"Ошибка инициализации БД: {e}")
        finally:
            conn.close()

    def _init_common_words(self, cursor):
        """Заполнение таблицы общих слов"""
        cursor.execute('SELECT COUNT(*) FROM common_words')
        if cursor.fetchone()[0] == 0:
            for russian, translation in VOCABULARY.items():
                cursor.execute('''
                INSERT OR IGNORE INTO common_words (russian, translation)
                VALUES (?, ?)
                ''', (russian, translation))
            logger.info("Таблица common_words заполнена")

    def register_user(self, user_id, username=None, first_name=None):
        """Регистрация нового пользователя"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()

            cursor.execute('''
            INSERT OR IGNORE INTO users (user_id, username, first_name)
            VALUES (?, ?, ?)
            ''', (user_id, username, first_name))

            conn.commit()

            # Проверяем, есть ли у пользователя слова
            cursor.execute(
                'SELECT COUNT(*) FROM user_words WHERE user_id = ?',
                (user_id,)
            )
            if cursor.fetchone()[0] == 0:
                self.add_default_words(user_id)

            return True

        except sqlite3.Error as e:
            logger.error(f"Ошибка регистрации: {e}")
            return False
        finally:
            conn.close()

    def add_default_words(self, user_id):
        """Добавление стандартных слов новому пользователю"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()

            # Берём слова из общего словаря
            cursor.execute(
                'SELECT russian, translation FROM common_words LIMIT 10'
            )
            default_words = cursor.fetchall()

            for russian, translation in default_words:
                cursor.execute('''
                INSERT OR IGNORE INTO user_words (user_id, russian, translation)
                VALUES (?, ?, ?)
                ''', (user_id, russian, translation))

            conn.commit()
            logger.info(f"Добавлены стандартные слова для {user_id}")

        except sqlite3.Error as e:
            logger.error(f"Ошибка добавления стандартных слов: {e}")
        finally:
            conn.close()

    def add_word(self, user_id, russian, translation):
        """Добавление слова пользователю"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()

            cursor.execute('''
            INSERT OR IGNORE INTO user_words (user_id, russian, translation)
            VALUES (?, ?, ?)
            ''', (user_id, russian.strip().capitalize(),
                  translation.strip().lower()))

            conn.commit()
            return cursor.rowcount > 0

        except sqlite3.Error as e:
            logger.error(f"Ошибка добавления слова: {e}")
            return False
        finally:
            conn.close()

    def delete_word(self, user_id, russian):
        """Удаление слова пользователя"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()

            cursor.execute('''
            DELETE FROM user_words
            WHERE user_id = ? AND russian = ?
            ''', (user_id, russian.strip().capitalize()))

            conn.commit()
            return cursor.rowcount > 0

        except sqlite3.Error as e:
            logger.error(f"Ошибка удаления слова: {e}")
            return False
        finally:
            conn.close()
            def get_user_words(self, user_id):
                """Получение всех слов пользователя"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()

            cursor.execute('''
            SELECT russian, translation, correct_count, wrong_count
            FROM user_words
            WHERE user_id = ?
            ORDER BY russian
            ''', (user_id,))

            return cursor.fetchall()

        except sqlite3.Error as e:
            logger.error(f"Ошибка получения слов: {e}")
            return []
        finally:
            conn.close()

    def get_random_word(self, user_id):
        """Получение случайного слова пользователя"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()

            cursor.execute('''
            SELECT russian, translation FROM user_words
            WHERE user_id = ?
            ORDER BY RANDOM() LIMIT 1
            ''', (user_id,))

            result = cursor.fetchone()
            return result if result else (None, None)

        except sqlite3.Error as e:
            logger.error(f"Ошибка получения случайного слова: {e}")
            return (None, None)
        finally:
            conn.close()

    def update_stats(self, user_id, russian, is_correct):
        """Обновление статистики по слову"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()

            field = 'correct_count' if is_correct else 'wrong_count'
            cursor.execute(f'''
            UPDATE user_words
            SET {field} = {field} + 1
            WHERE user_id = ? AND russian = ?
            ''', (user_id, russian))

            # Обновляем общую статистику пользователя
            cursor.execute('''
            UPDATE users
            SET total_quizzes = total_quizzes + 1,
                correct_answers = correct_answers + ?
            WHERE user_id = ?
            ''', (1 if is_correct else 0, user_id))

            conn.commit()

        except sqlite3.Error as e:
            logger.error(f"Ошибка обновления статистики: {e}")
        finally:
            conn.close()

    def get_stats(self, user_id):
        """Получение статистики пользователя"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()

            cursor.execute('''
            SELECT total_quizzes, correct_answers,
                   (SELECT COUNT(*) FROM user_words WHERE user_id = ?) 
                   as total_words
            FROM users
            WHERE user_id = ?
            ''', (user_id, user_id))

            result = cursor.fetchone()
            if result:
                total_quizzes, correct_answers, total_words = result
                accuracy = (correct_answers / total_quizzes * 100) if total_quizzes > 0 else 0
                return {
                    'total_quizzes': total_quizzes,
                    'correct_answers': correct_answers,
                    'accuracy': round(accuracy, 1),
                    'total_words': total_words
                }
            return None

        except sqlite3.Error as e:
            logger.error(f"Ошибка получения статистики: {e}")
            return None
        finally:
            conn.close()

    def get_common_words(self, limit=None):
        """Получение общих слов из словаря"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()

            query = 'SELECT russian, translation FROM common_words'
            if limit:
                query += ' LIMIT ?'
                cursor.execute(query, (limit,))
            else:
                cursor.execute(query)

            return cursor.fetchall()

        except sqlite3.Error as e:
            logger.error(f"Ошибка получения общих слов: {e}")
            return []
        finally:
            conn.close()