trevor/quiz_test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

devops
Some checks reported errors
continuous-integration/drone/push Build encountered an error
Details

Browse Source
This commit is contained in:
Andrey K. Choi
2025-09-11 08:02:35 +09:00
parent 398729a4a0
commit fcf27c1639
13 changed files with 585 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files

234
src/database/database.py
View File

@@ -1,17 +1,20 @@
import aiosqlite
import logging
from typing import List, Dict, Optional, Tuple, Union
import json
import logging
from typing import Dict, List, Optional, Tuple, Union
import aiosqlite
class DatabaseManager:
def __init__(self, db_path: str):
self.db_path = db_path
async def init_database(self):
"""Инициализация базы данных и создание таблиц"""
async with aiosqlite.connect(self.db_path) as db:
# Таблица пользователей
await db.execute("""
await db.execute(
"""
CREATE TABLE IF NOT EXISTS users (
user_id INTEGER PRIMARY KEY,
username TEXT,
@@ -23,10 +26,12 @@ class DatabaseManager:
total_questions INTEGER DEFAULT 0,
correct_answers INTEGER DEFAULT 0
)
""")
"""
)
# Таблица тестов
await db.execute("""
await db.execute(
"""
CREATE TABLE IF NOT EXISTS tests (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
@@ -37,10 +42,12 @@ class DatabaseManager:
created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
is_active BOOLEAN DEFAULT TRUE
)
""")
"""
)
# Таблица вопросов
await db.execute("""
await db.execute(
"""
CREATE TABLE IF NOT EXISTS questions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
test_id INTEGER,
@@ -52,10 +59,12 @@ class DatabaseManager:
correct_answer INTEGER NOT NULL,
FOREIGN KEY (test_id) REFERENCES tests (id)
)
""")
"""
)
# Таблица результатов
await db.execute("""
await db.execute(
"""
CREATE TABLE IF NOT EXISTS results (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER,
@@ -70,10 +79,12 @@ class DatabaseManager:
FOREIGN KEY (user_id) REFERENCES users (user_id),
FOREIGN KEY (test_id) REFERENCES tests (id)
)
""")
"""
)
# Таблица активных сессий
await db.execute("""
await db.execute(
"""
CREATE TABLE IF NOT EXISTS active_sessions (
user_id INTEGER PRIMARY KEY,
test_id INTEGER,
@@ -85,28 +96,38 @@ class DatabaseManager:
FOREIGN KEY (user_id) REFERENCES users (user_id),
FOREIGN KEY (test_id) REFERENCES tests (id)
)
""")
"""
)
await db.commit()
logging.info("Database initialized successfully")
async def register_user(self, user_id: int, username: Optional[str] = None,
first_name: Optional[str] = None, last_name: Optional[str] = None,
language_code: str = 'ru', is_guest: bool = True) -> bool:
async def register_user(
self,
user_id: int,
username: Optional[str] = None,
first_name: Optional[str] = None,
last_name: Optional[str] = None,
language_code: str = "ru",
is_guest: bool = True,
) -> bool:
"""Регистрация нового пользователя"""
try:
async with aiosqlite.connect(self.db_path) as db:
await db.execute("""
await db.execute(
"""
INSERT OR REPLACE INTO users
(user_id, username, first_name, last_name, language_code, is_guest)
VALUES (?, ?, ?, ?, ?, ?)
""", (user_id, username, first_name, last_name, language_code, is_guest))
""",
(user_id, username, first_name, last_name, language_code, is_guest),
)
await db.commit()
return True
except Exception as e:
logging.error(f"Error registering user {user_id}: {e}")
return False
async def get_user(self, user_id: int) -> Optional[Dict]:
"""Получение данных пользователя"""
try:
@@ -122,30 +143,34 @@ class DatabaseManager:
except Exception as e:
logging.error(f"Error getting user {user_id}: {e}")
return None
async def add_test(self, name: str, description: str, level: int,
category: str, csv_file: str) -> Optional[int]:
async def add_test(
self, name: str, description: str, level: int, category: str, csv_file: str
) -> Optional[int]:
"""Добавление нового теста"""
try:
async with aiosqlite.connect(self.db_path) as db:
cursor = await db.execute("""
cursor = await db.execute(
"""
INSERT INTO tests (name, description, level, category, csv_file)
VALUES (?, ?, ?, ?, ?)
""", (name, description, level, category, csv_file))
""",
(name, description, level, category, csv_file),
)
await db.commit()
return cursor.lastrowid
except Exception as e:
logging.error(f"Error adding test: {e}")
return None
async def get_tests_by_category(self, category: Optional[str] = None) -> List[Dict]:
"""Получение тестов по категории"""
try:
async with aiosqlite.connect(self.db_path) as db:
if category:
cursor = await db.execute(
"SELECT * FROM tests WHERE category = ? AND is_active = TRUE ORDER BY level",
(category,)
"SELECT * FROM tests WHERE category = ? AND is_active = TRUE ORDER BY level",
(category,),
)
else:
cursor = await db.execute(
@@ -157,56 +182,73 @@ class DatabaseManager:
except Exception as e:
logging.error(f"Error getting tests: {e}")
return []
async def add_questions_to_test(self, test_id: int, questions: List[Dict]) -> bool:
"""Добавление вопросов к тесту"""
try:
async with aiosqlite.connect(self.db_path) as db:
for q in questions:
await db.execute("""
await db.execute(
"""
INSERT INTO questions
(test_id, question, option1, option2, option3, option4, correct_answer)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (test_id, q['question'], q['option1'], q['option2'],
q['option3'], q['option4'], q['correct_answer']))
""",
(
test_id,
q["question"],
q["option1"],
q["option2"],
q["option3"],
q["option4"],
q["correct_answer"],
),
)
await db.commit()
return True
except Exception as e:
logging.error(f"Error adding questions to test {test_id}: {e}")
return False
async def get_random_questions(self, test_id: int, count: int = 10) -> List[Dict]:
"""Получение случайных вопросов из теста"""
try:
async with aiosqlite.connect(self.db_path) as db:
cursor = await db.execute("""
cursor = await db.execute(
"""
SELECT * FROM questions WHERE test_id = ?
ORDER BY RANDOM() LIMIT ?
""", (test_id, count))
""",
(test_id, count),
)
rows = await cursor.fetchall()
columns = [description[0] for description in cursor.description]
return [dict(zip(columns, row)) for row in rows]
except Exception as e:
logging.error(f"Error getting random questions: {e}")
return []
async def start_session(self, user_id: int, test_id: int,
questions: List[Dict], mode: str) -> bool:
async def start_session(
self, user_id: int, test_id: int, questions: List[Dict], mode: str
) -> bool:
"""Начало новой сессии викторины"""
try:
async with aiosqlite.connect(self.db_path) as db:
questions_json = json.dumps(questions)
await db.execute("""
await db.execute(
"""
INSERT OR REPLACE INTO active_sessions
(user_id, test_id, questions_data, mode)
VALUES (?, ?, ?, ?)
""", (user_id, test_id, questions_json, mode))
""",
(user_id, test_id, questions_json, mode),
)
await db.commit()
return True
except Exception as e:
logging.error(f"Error starting session: {e}")
return False
async def get_active_session(self, user_id: int) -> Optional[Dict]:
"""Получение активной сессии пользователя"""
try:
@@ -218,45 +260,54 @@ class DatabaseManager:
if row:
columns = [description[0] for description in cursor.description]
session = dict(zip(columns, row))
session['questions_data'] = json.loads(session['questions_data'])
session["questions_data"] = json.loads(session["questions_data"])
return session
return None
except Exception as e:
logging.error(f"Error getting active session: {e}")
return None
async def update_session_progress(self, user_id: int, question_num: int,
correct_count: int) -> bool:
async def update_session_progress(
self, user_id: int, question_num: int, correct_count: int
) -> bool:
"""Обновление прогресса сессии"""
try:
async with aiosqlite.connect(self.db_path) as db:
await db.execute("""
await db.execute(
"""
UPDATE active_sessions
SET current_question = ?, correct_count = ?
WHERE user_id = ?
""", (question_num, correct_count, user_id))
""",
(question_num, correct_count, user_id),
)
await db.commit()
return True
except Exception as e:
logging.error(f"Error updating session progress: {e}")
return False
async def update_session_questions(self, user_id: int, questions_data: list) -> bool:
async def update_session_questions(
self, user_id: int, questions_data: list
) -> bool:
"""Обновление данных вопросов в сессии (например, после перемешивания)"""
try:
async with aiosqlite.connect(self.db_path) as db:
questions_json = json.dumps(questions_data, ensure_ascii=False)
await db.execute("""
await db.execute(
"""
UPDATE active_sessions
SET questions_data = ?
WHERE user_id = ?
""", (questions_json, user_id))
""",
(questions_json, user_id),
)
await db.commit()
return True
except Exception as e:
logging.error(f"Error updating session questions: {e}")
return False
async def finish_session(self, user_id: int, score: float) -> bool:
"""Завершение сессии и сохранение результатов"""
try:
@@ -265,37 +316,52 @@ class DatabaseManager:
session = await self.get_active_session(user_id)
if not session:
return False
# Сохраняем результат
await db.execute("""
await db.execute(
"""
INSERT INTO results
(user_id, test_id, mode, questions_asked, correct_answers, score)
VALUES (?, ?, ?, ?, ?, ?)
""", (user_id, session['test_id'], session['mode'],
len(session['questions_data']), session['correct_count'], score))
""",
(
user_id,
session["test_id"],
session["mode"],
len(session["questions_data"]),
session["correct_count"],
score,
),
)
# Обновляем статистику пользователя
await db.execute("""
await db.execute(
"""
UPDATE users
SET total_questions = total_questions + ?,
correct_answers = correct_answers + ?
WHERE user_id = ?
""", (len(session['questions_data']), session['correct_count'], user_id))
""",
(len(session["questions_data"]), session["correct_count"], user_id),
)
# Удаляем активную сессию
await db.execute("DELETE FROM active_sessions WHERE user_id = ?", (user_id,))
await db.execute(
"DELETE FROM active_sessions WHERE user_id = ?", (user_id,)
)
await db.commit()
return True
except Exception as e:
logging.error(f"Error finishing session: {e}")
return False
async def get_user_stats(self, user_id: int) -> Optional[Dict]:
"""Получение статистики пользователя"""
try:
async with aiosqlite.connect(self.db_path) as db:
cursor = await db.execute("""
cursor = await db.execute(
"""
SELECT
u.total_questions,
u.correct_answers,
@@ -308,8 +374,10 @@ class DatabaseManager:
LEFT JOIN results r ON u.user_id = r.user_id
WHERE u.user_id = ?
GROUP BY u.user_id
""", (user_id,))
""",
(user_id,),
)
row = await cursor.fetchone()
if row:
columns = [description[0] for description in cursor.description]
@@ -319,12 +387,13 @@ class DatabaseManager:
except Exception as e:
logging.error(f"Error getting user stats: {e}")
return None
async def get_recent_results(self, user_id: int, limit: int = 5) -> List[Dict]:
"""Получение последних результатов пользователя"""
try:
async with aiosqlite.connect(self.db_path) as db:
cursor = await db.execute("""
cursor = await db.execute(
"""
SELECT
r.mode,
r.questions_asked,
@@ -338,20 +407,23 @@ class DatabaseManager:
WHERE r.user_id = ?
ORDER BY r.end_time DESC
LIMIT ?
""", (user_id, limit))
""",
(user_id, limit),
)
rows = await cursor.fetchall()
columns = [description[0] for description in cursor.description]
return [dict(zip(columns, row)) for row in rows]
except Exception as e:
logging.error(f"Error getting recent results: {e}")
return []
async def get_category_stats(self, user_id: int) -> List[Dict]:
"""Получение статистики по категориям"""
try:
async with aiosqlite.connect(self.db_path) as db:
cursor = await db.execute("""
cursor = await db.execute(
"""
SELECT
t.category,
COUNT(r.id) as attempts,
@@ -364,8 +436,10 @@ class DatabaseManager:
WHERE r.user_id = ?
GROUP BY t.category
ORDER BY attempts DESC
""", (user_id,))
""",
(user_id,),
)
rows = await cursor.fetchall()
columns = [description[0] for description in cursor.description]
return [dict(zip(columns, row)) for row in rows]