You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
332 lines
11 KiB
332 lines
11 KiB
"""
|
|
Database connection manager for Tesla Coil Spark Course
|
|
Handles SQLite connections, schema creation, and queries
|
|
"""
|
|
|
|
import sqlite3
|
|
import os
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
|
|
class Database:
|
|
"""SQLite database manager"""
|
|
|
|
def __init__(self, db_path=None):
|
|
"""
|
|
Initialize database connection
|
|
|
|
Args:
|
|
db_path: Path to SQLite database file. If None, uses default location.
|
|
"""
|
|
if db_path is None:
|
|
# Default location: user's home directory
|
|
home = Path.home()
|
|
data_dir = home / '.tesla_spark_course'
|
|
data_dir.mkdir(exist_ok=True)
|
|
db_path = data_dir / 'progress.db'
|
|
|
|
self.db_path = db_path
|
|
self.connection = None
|
|
self._connect()
|
|
self._initialize_schema()
|
|
|
|
def _connect(self):
|
|
"""Establish database connection"""
|
|
try:
|
|
self.connection = sqlite3.connect(
|
|
self.db_path,
|
|
check_same_thread=False # Allow usage from multiple threads
|
|
)
|
|
self.connection.row_factory = sqlite3.Row # Access columns by name
|
|
print(f"[DB] Connected to database: {self.db_path}")
|
|
except sqlite3.Error as e:
|
|
print(f"[DB ERROR] Failed to connect: {e}")
|
|
raise
|
|
|
|
def _initialize_schema(self):
|
|
"""Create tables if they don't exist"""
|
|
schema_file = Path(__file__).parent.parent / 'resources' / 'database' / 'schema.sql'
|
|
|
|
if not schema_file.exists():
|
|
print(f"[DB WARNING] Schema file not found: {schema_file}")
|
|
return
|
|
|
|
try:
|
|
with open(schema_file, 'r') as f:
|
|
schema_sql = f.read()
|
|
|
|
cursor = self.connection.cursor()
|
|
cursor.executescript(schema_sql)
|
|
self.connection.commit()
|
|
print("[DB] Schema initialized successfully")
|
|
except sqlite3.Error as e:
|
|
print(f"[DB ERROR] Failed to initialize schema: {e}")
|
|
raise
|
|
|
|
def execute(self, query, params=None):
|
|
"""
|
|
Execute a query and return cursor
|
|
|
|
Args:
|
|
query: SQL query string
|
|
params: Query parameters (tuple or dict)
|
|
|
|
Returns:
|
|
sqlite3.Cursor
|
|
"""
|
|
try:
|
|
cursor = self.connection.cursor()
|
|
if params:
|
|
cursor.execute(query, params)
|
|
else:
|
|
cursor.execute(query)
|
|
return cursor
|
|
except sqlite3.Error as e:
|
|
print(f"[DB ERROR] Query failed: {e}")
|
|
print(f"[DB ERROR] Query: {query}")
|
|
raise
|
|
|
|
def fetch_one(self, query, params=None):
|
|
"""Execute query and fetch one result"""
|
|
cursor = self.execute(query, params)
|
|
return cursor.fetchone()
|
|
|
|
def fetch_all(self, query, params=None):
|
|
"""Execute query and fetch all results"""
|
|
cursor = self.execute(query, params)
|
|
return cursor.fetchall()
|
|
|
|
def commit(self):
|
|
"""Commit transaction"""
|
|
self.connection.commit()
|
|
|
|
def close(self):
|
|
"""Close database connection"""
|
|
if self.connection:
|
|
self.connection.close()
|
|
print("[DB] Connection closed")
|
|
|
|
# =========================================================================
|
|
# Convenience methods for common operations
|
|
# =========================================================================
|
|
|
|
def get_user(self, user_id=1):
|
|
"""Get user by ID (default user is ID 1)"""
|
|
return self.fetch_one(
|
|
"SELECT * FROM users WHERE user_id = ?",
|
|
(user_id,)
|
|
)
|
|
|
|
def get_lesson_progress(self, user_id, lesson_id):
|
|
"""Get progress for a specific lesson"""
|
|
return self.fetch_one(
|
|
"SELECT * FROM lesson_progress WHERE user_id = ? AND lesson_id = ?",
|
|
(user_id, lesson_id)
|
|
)
|
|
|
|
def update_lesson_progress(self, user_id, lesson_id, **kwargs):
|
|
"""
|
|
Update lesson progress
|
|
|
|
Args:
|
|
user_id: User ID
|
|
lesson_id: Lesson ID
|
|
**kwargs: Fields to update (status, scroll_position, time_spent, etc.)
|
|
"""
|
|
# First, ensure record exists
|
|
existing = self.get_lesson_progress(user_id, lesson_id)
|
|
|
|
if existing is None:
|
|
# Create new record
|
|
self.execute(
|
|
"""INSERT INTO lesson_progress
|
|
(user_id, lesson_id, first_opened, last_accessed)
|
|
VALUES (?, ?, ?, ?)""",
|
|
(user_id, lesson_id, datetime.now(), datetime.now())
|
|
)
|
|
|
|
# Update fields
|
|
if kwargs:
|
|
# Add last_accessed to every update
|
|
kwargs['last_accessed'] = datetime.now()
|
|
|
|
set_clause = ', '.join([f"{key} = ?" for key in kwargs.keys()])
|
|
values = list(kwargs.values()) + [user_id, lesson_id]
|
|
|
|
query = f"""UPDATE lesson_progress
|
|
SET {set_clause}
|
|
WHERE user_id = ? AND lesson_id = ?"""
|
|
self.execute(query, values)
|
|
self.commit()
|
|
|
|
def mark_lesson_complete(self, user_id, lesson_id):
|
|
"""Mark a lesson as completed"""
|
|
self.update_lesson_progress(
|
|
user_id, lesson_id,
|
|
status='completed',
|
|
completion_percentage=100,
|
|
completed_at=datetime.now()
|
|
)
|
|
|
|
def get_all_lesson_progress(self, user_id):
|
|
"""Get progress for all lessons"""
|
|
return self.fetch_all(
|
|
"SELECT * FROM lesson_progress WHERE user_id = ?",
|
|
(user_id,)
|
|
)
|
|
|
|
def record_exercise_attempt(self, user_id, exercise_id, user_answer,
|
|
is_correct, points_earned, points_possible,
|
|
hints_used=0, time_taken=0, lesson_id=None):
|
|
"""Record an exercise attempt"""
|
|
# Get attempt number
|
|
cursor = self.execute(
|
|
"""SELECT COALESCE(MAX(attempt_number), 0) + 1 as next_attempt
|
|
FROM exercise_attempts
|
|
WHERE user_id = ? AND exercise_id = ?""",
|
|
(user_id, exercise_id)
|
|
)
|
|
attempt_number = cursor.fetchone()['next_attempt']
|
|
|
|
# Insert attempt
|
|
self.execute(
|
|
"""INSERT INTO exercise_attempts
|
|
(user_id, exercise_id, lesson_id, attempt_number, user_answer,
|
|
is_correct, points_earned, points_possible, hints_used, time_taken)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(user_id, exercise_id, lesson_id, attempt_number, user_answer,
|
|
is_correct, points_earned, points_possible, hints_used, time_taken)
|
|
)
|
|
|
|
# Update or create completion record
|
|
existing = self.fetch_one(
|
|
"SELECT * FROM exercise_completion WHERE user_id = ? AND exercise_id = ?",
|
|
(user_id, exercise_id)
|
|
)
|
|
|
|
if existing is None:
|
|
# First attempt
|
|
self.execute(
|
|
"""INSERT INTO exercise_completion
|
|
(user_id, exercise_id, best_score, max_possible, total_attempts,
|
|
first_attempted, first_completed, last_attempted)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(user_id, exercise_id, points_earned, points_possible, 1,
|
|
datetime.now(), datetime.now() if is_correct else None, datetime.now())
|
|
)
|
|
else:
|
|
# Update existing
|
|
best_score = max(existing['best_score'], points_earned)
|
|
first_completed = existing['first_completed']
|
|
if is_correct and first_completed is None:
|
|
first_completed = datetime.now()
|
|
|
|
self.execute(
|
|
"""UPDATE exercise_completion
|
|
SET best_score = ?, total_attempts = total_attempts + 1,
|
|
first_completed = ?, last_attempted = ?
|
|
WHERE user_id = ? AND exercise_id = ?""",
|
|
(best_score, first_completed, datetime.now(), user_id, exercise_id)
|
|
)
|
|
|
|
self.commit()
|
|
|
|
def get_overall_progress(self, user_id):
|
|
"""Get overall progress statistics"""
|
|
# Total points earned
|
|
points_result = self.fetch_one(
|
|
"""SELECT SUM(best_score) as total_points
|
|
FROM exercise_completion
|
|
WHERE user_id = ?""",
|
|
(user_id,)
|
|
)
|
|
total_points = points_result['total_points'] or 0
|
|
|
|
# Lessons completed
|
|
lessons_result = self.fetch_one(
|
|
"""SELECT COUNT(*) as completed
|
|
FROM lesson_progress
|
|
WHERE user_id = ? AND status = 'completed'""",
|
|
(user_id,)
|
|
)
|
|
lessons_completed = lessons_result['completed'] or 0
|
|
|
|
# Total study time
|
|
time_result = self.fetch_one(
|
|
"""SELECT SUM(time_spent) as total_time
|
|
FROM lesson_progress
|
|
WHERE user_id = ?""",
|
|
(user_id,)
|
|
)
|
|
total_time = time_result['total_time'] or 0
|
|
|
|
return {
|
|
'total_points': total_points,
|
|
'lessons_completed': lessons_completed,
|
|
'total_time': total_time,
|
|
'percentage': (lessons_completed / 30.0) * 100 # 30 total lessons
|
|
}
|
|
|
|
def update_study_session(self, user_id):
|
|
"""Update or create today's study session"""
|
|
today = datetime.now().date()
|
|
|
|
existing = self.fetch_one(
|
|
"SELECT * FROM study_sessions WHERE user_id = ? AND session_date = ?",
|
|
(user_id, today)
|
|
)
|
|
|
|
if existing is None:
|
|
self.execute(
|
|
"""INSERT INTO study_sessions
|
|
(user_id, session_date, session_start)
|
|
VALUES (?, ?, ?)""",
|
|
(user_id, today, datetime.now())
|
|
)
|
|
else:
|
|
self.execute(
|
|
"""UPDATE study_sessions
|
|
SET session_end = ?
|
|
WHERE user_id = ? AND session_date = ?""",
|
|
(datetime.now(), user_id, today)
|
|
)
|
|
|
|
self.commit()
|
|
|
|
def get_study_streak(self, user_id):
|
|
"""Calculate current study streak (consecutive days)"""
|
|
sessions = self.fetch_all(
|
|
"""SELECT session_date FROM study_sessions
|
|
WHERE user_id = ?
|
|
ORDER BY session_date DESC""",
|
|
(user_id,)
|
|
)
|
|
|
|
if not sessions:
|
|
return 0
|
|
|
|
from datetime import timedelta
|
|
streak = 0
|
|
expected_date = datetime.now().date()
|
|
|
|
for session in sessions:
|
|
session_date = datetime.strptime(session['session_date'], '%Y-%m-%d').date()
|
|
if session_date == expected_date:
|
|
streak += 1
|
|
expected_date -= timedelta(days=1)
|
|
else:
|
|
break
|
|
|
|
return streak
|
|
|
|
|
|
# Global database instance
|
|
_db_instance = None
|
|
|
|
def get_database():
|
|
"""Get global database instance"""
|
|
global _db_instance
|
|
if _db_instance is None:
|
|
_db_instance = Database()
|
|
return _db_instance
|