# database.py """ Модуль для работы с базой данных SQLite3 """ import sqlite3 import bcrypt from typing import Optional, Tuple from config import DATABASE_PATH, MESSAGES, DEBUG_MODE def debug_log(message: str) -> None: """Вывод отладочных сообщений""" if DEBUG_MODE: print(f"[DEBUG] {message}") def init_database() -> None: """Инициализация базы данных и создание таблицы пользователей""" debug_log(f"Инициализация базы данных: {DATABASE_PATH}") try: conn = sqlite3.connect(DATABASE_PATH) cursor = conn.cursor() # Создание таблицы пользователей cursor.execute(''' CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, password_hash TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') conn.commit() conn.close() debug_log("База данных инициализирована успешно") except sqlite3.Error as e: debug_log(f"Ошибка при инициализации базы данных: {e}") raise def user_exists(username: str) -> bool: """ Проверка существования пользователя в базе данных Args: username: Логин пользователя Returns: bool: True если пользователь существует, False в противном случае """ debug_log(f"Проверка существования пользователя: {username}") try: conn = sqlite3.connect(DATABASE_PATH) cursor = conn.cursor() cursor.execute('SELECT id FROM users WHERE username = ?', (username,)) result = cursor.fetchone() conn.close() exists = result is not None debug_log(f"Пользователь {'существует' if exists else 'не существует'}") return exists except sqlite3.Error as e: debug_log(f"Ошибка при проверке существования пользователя: {e}") return False def create_user(username: str, password: str) -> Tuple[bool, str]: """ Создание нового пользователя в базе данных Args: username: Логин пользователя password: Пароль пользователя (будет захеширован) Returns: Tuple[bool, str]: (успех, сообщение) """ debug_log(f"Создание пользователя: {username}") try: # Проверка на существование пользователя if user_exists(username): debug_log("Ошибка: пользователь уже существует") return False, MESSAGES['error_username_exists'] # Хеширование пароля password_bytes = password.encode('utf-8') salt = bcrypt.gensalt() password_hash = bcrypt.hashpw(password_bytes, salt) debug_log("Пароль захеширован") # Сохранение в базу данных conn = sqlite3.connect(DATABASE_PATH) cursor = conn.cursor() cursor.execute( 'INSERT INTO users (username, password_hash) VALUES (?, ?)', (username, password_hash) ) conn.commit() conn.close() debug_log("Пользователь успешно создан в базе данных") return True, MESSAGES['success_register'] except sqlite3.Error as e: debug_log(f"Ошибка базы данных: {e}") return False, MESSAGES['error_database'] def verify_user(username: str, password: str) -> Tuple[bool, str]: """ Проверка учетных данных пользователя Args: username: Логин пользователя password: Пароль пользователя Returns: Tuple[bool, str]: (успех, сообщение) """ debug_log(f"Проверка учетных данных для: {username}") try: conn = sqlite3.connect(DATABASE_PATH) cursor = conn.cursor() cursor.execute( 'SELECT password_hash FROM users WHERE username = ?', (username,) ) result = cursor.fetchone() conn.close() if result is None: debug_log("Ошибка: пользователь не найден") return False, MESSAGES['error_invalid_credentials'] stored_hash = result[0] password_bytes = password.encode('utf-8') # Проверка пароля if bcrypt.checkpw(password_bytes, stored_hash): debug_log("Аутентификация успешна") return True, MESSAGES['success_login'] else: debug_log("Ошибка: неверный пароль") return False, MESSAGES['error_invalid_credentials'] except sqlite3.Error as e: debug_log(f"Ошибка базы данных: {e}") return False, MESSAGES['error_database']