163 lines
5.5 KiB
Python
163 lines
5.5 KiB
Python
# 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']
|
||
|