Загрузить файлы в «/»

database.py

@@ -0,0 +1,162 @@
+# 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']
+