import sqlite3 import uuid import os from datetime import datetime, timezone DB_FOLDER = os.path.join(os.path.dirname(os.path.abspath(__file__)), "scratch") DB_PATH = os.path.join(DB_FOLDER, "sandbox.db") def init_db(): os.makedirs(DB_FOLDER, exist_ok=True) conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() # Enable foreign keys cursor.execute("PRAGMA foreign_keys = ON;") # 1. Problems table cursor.execute(""" CREATE TABLE IF NOT EXISTS problems ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, name TEXT NOT NULL, problem_number INTEGER, pattern TEXT NOT NULL, difficulty TEXT NOT NULL, google_doc_url TEXT, google_doc_id TEXT, optimal_time TEXT DEFAULT 'O(N)', optimal_space TEXT DEFAULT 'O(N)', description TEXT, reference_code TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP, UNIQUE(user_id, name) ); """) # 2. User Reviews table cursor.execute(""" CREATE TABLE IF NOT EXISTS user_reviews ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, problem_id TEXT NOT NULL, box_level INTEGER DEFAULT 1, last_reviewed TEXT DEFAULT CURRENT_TIMESTAMP, next_review TEXT DEFAULT CURRENT_TIMESTAMP, times_correct INTEGER DEFAULT 0, total_attempts INTEGER DEFAULT 0, UNIQUE(user_id, problem_id), FOREIGN KEY(problem_id) REFERENCES problems(id) ON DELETE CASCADE ); """) # 3. User Streaks table cursor.execute(""" CREATE TABLE IF NOT EXISTS user_streaks ( user_id TEXT PRIMARY KEY, current_streak INTEGER DEFAULT 0, longest_streak INTEGER DEFAULT 0, last_active_date TEXT, updated_at TEXT DEFAULT CURRENT_TIMESTAMP ); """) # 4. User Journey Progress table cursor.execute(""" CREATE TABLE IF NOT EXISTS user_journey_progress ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, problem_number INTEGER NOT NULL, viewed_at TEXT DEFAULT CURRENT_TIMESTAMP, UNIQUE(user_id, problem_number) ); """) # 5. User Scores table cursor.execute(""" CREATE TABLE IF NOT EXISTS user_scores ( user_id TEXT PRIMARY KEY, total_score INTEGER DEFAULT 0, python_challenges_completed INTEGER DEFAULT 0, sql_challenges_completed INTEGER DEFAULT 0, updated_at TEXT DEFAULT CURRENT_TIMESTAMP ); """) # 6. Challenge History table cursor.execute(""" CREATE TABLE IF NOT EXISTS challenge_history ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, challenge_type TEXT NOT NULL, challenge_title TEXT NOT NULL, points_awarded INTEGER DEFAULT 0, completed_at TEXT DEFAULT CURRENT_TIMESTAMP ); """) conn.commit() # Seed default problems and reviews if empty cursor.execute("SELECT COUNT(*) FROM problems WHERE user_id = 'sandbox-user-id'") if cursor.fetchone()[0] == 0: seed_default_problems(cursor) conn.commit() conn.close() def seed_default_problems(cursor): default_problems = [ { "id": "prob-contains-duplicate", "user_id": "sandbox-user-id", "name": "Contains Duplicate", "problem_number": 217, "pattern": "Arrays & Hashing", "difficulty": "Easy", "google_doc_url": "https://leetcode.com/problems/contains-duplicate/", "google_doc_id": "doc-contains-duplicate", "optimal_time": "O(N)", "optimal_space": "O(N)", "description": "Given an integer array nums, return true if any value appears at least twice in the array, and return false if every element is distinct.", "reference_code": "def solve(nums: list[int]) -> bool:\n # Return True if any value appears at least twice\n return len(nums) != len(set(nums))" }, { "id": "prob-valid-anagram", "user_id": "sandbox-user-id", "name": "Valid Anagram", "problem_number": 242, "pattern": "Arrays & Hashing", "difficulty": "Easy", "google_doc_url": "https://leetcode.com/problems/valid-anagram/", "google_doc_id": "doc-valid-anagram", "optimal_time": "O(N)", "optimal_space": "O(1)", "description": "Given two strings s and t, return true if t is an anagram of s, and false otherwise.", "reference_code": "def solve(s: str, t: str) -> bool:\n # Return True if t is an anagram of s\n if len(s) != len(t):\n return False\n return sorted(s) == sorted(t)" }, { "id": "prob-two-sum", "user_id": "sandbox-user-id", "name": "Two Sum", "problem_number": 1, "pattern": "Arrays & Hashing", "difficulty": "Easy", "google_doc_url": "https://leetcode.com/problems/two-sum/", "google_doc_id": "doc-two-sum", "optimal_time": "O(N)", "optimal_space": "O(N)", "description": "Given an array of integers nums and an integer target, return indices of the two numbers such that they add up to target.", "reference_code": "def solve(nums: list[int], target: int) -> list[int]:\n # Return indices of the two numbers that add up to target\n seen = {}\n for i, n in enumerate(nums):\n diff = target - n\n if diff in seen:\n return [seen[diff], i]\n seen[n] = i\n return []" }, { "id": "prob-group-anagrams", "user_id": "sandbox-user-id", "name": "Group Anagrams", "problem_number": 49, "pattern": "Arrays & Hashing", "difficulty": "Medium", "google_doc_url": "https://leetcode.com/problems/group-anagrams/", "google_doc_id": "doc-group-anagrams", "optimal_time": "O(N * K log K)", "optimal_space": "O(N * K)", "description": "Given an array of strings strs, group the anagrams together. You can return the answer in any order.", "reference_code": "def solve(strs: list[str]) -> list[list[str]]:\n # Group anagrams together\n ans = {}\n for s in strs:\n key = ''.join(sorted(s))\n if key not in ans:\n ans[key] = []\n ans[key].append(s)\n return list(ans.values())" }, { "id": "prob-valid-parentheses", "user_id": "sandbox-user-id", "name": "Valid Parentheses", "problem_number": 20, "pattern": "Stack", "difficulty": "Easy", "google_doc_url": "https://leetcode.com/problems/valid-parentheses/", "google_doc_id": "doc-valid-parentheses", "optimal_time": "O(N)", "optimal_space": "O(N)", "description": "Given a string s containing just the characters '(', ')', '{', '}', '[' and ']', determine if the input string is valid.", "reference_code": "def solve(s: str) -> bool:\n # Determine if the parentheses input string is valid\n stack = []\n mapping = {')': '(', '}': '{', ']': '['}\n for char in s:\n if char in mapping:\n top_element = stack.pop() if stack else '#'\n if mapping[char] != top_element:\n return False\n else:\n stack.append(char)\n return not stack" } ] for p in default_problems: columns = list(p.keys()) placeholders = ", ".join(["?"] * len(columns)) query = f"INSERT OR IGNORE INTO problems ({', '.join(columns)}) VALUES ({placeholders})" cursor.execute(query, list(p.values())) now_iso = datetime.now(timezone.utc).isoformat() default_reviews = [ { "id": "rev-contains-duplicate", "user_id": "sandbox-user-id", "problem_id": "prob-contains-duplicate", "box_level": 1, "last_reviewed": now_iso, "next_review": now_iso, "times_correct": 0, "total_attempts": 0 }, { "id": "rev-valid-anagram", "user_id": "sandbox-user-id", "problem_id": "prob-valid-anagram", "box_level": 1, "last_reviewed": now_iso, "next_review": now_iso, "times_correct": 0, "total_attempts": 0 }, { "id": "rev-two-sum", "user_id": "sandbox-user-id", "problem_id": "prob-two-sum", "box_level": 1, "last_reviewed": now_iso, "next_review": now_iso, "times_correct": 0, "total_attempts": 0 } ] for r in default_reviews: cols = list(r.keys()) places = ", ".join(["?"] * len(cols)) query = f"INSERT OR IGNORE INTO user_reviews ({', '.join(cols)}) VALUES ({places})" cursor.execute(query, list(r.values())) class MockResponse: def __init__(self, data): self.data = data class QueryBuilder: def __init__(self, table_name, real_client=None): self.table_name = table_name self.real_client = real_client self.filters = [] self.order_by = [] self.limit_val = None self.op = None self.op_data = None self.on_conflict = None def select(self, columns="*"): self.op = 'select' self.op_data = columns return self def insert(self, data): self.op = 'insert' self.op_data = data return self def update(self, data): self.op = 'update' self.op_data = data return self def upsert(self, data, on_conflict=None): self.op = 'upsert' self.op_data = data self.on_conflict = on_conflict return self def delete(self): self.op = 'delete' return self def eq(self, column, value): self.filters.append(('eq', column, value)) return self def lte(self, column, value): self.filters.append(('lte', column, value)) return self def order(self, column, desc=False): self.order_by.append((column, desc)) return self def limit(self, count): self.limit_val = count return self def execute(self): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row cursor = conn.cursor() try: if self.op == 'select': # Build SELECT query # If table is 'problems' and real client is online, try fetching from Supabase first if self.table_name == 'problems' and self.real_client: try: # Build public Supabase query supabase_query = self.real_client.table('problems').select('*') for op, col, val in self.filters: if op == 'eq': supabase_query = supabase_query.eq(col, val) elif op == 'lte': supabase_query = supabase_query.lte(col, val) for col, desc in self.order_by: supabase_query = supabase_query.order(col, desc=desc) if self.limit_val is not None: supabase_query = supabase_query.limit(self.limit_val) res = supabase_query.execute() data = res.data if res.data else [] # Post-process for nested table relations problems -> user_reviews if self.op_data and 'user_reviews(' in self.op_data: for row in data: prob_id = row.get('id') if prob_id: cursor.execute("SELECT * FROM user_reviews WHERE problem_id = ? AND user_id = 'sandbox-user-id'", (prob_id,)) rev_rows = cursor.fetchall() row['user_reviews'] = [dict(r) for r in rev_rows] return MockResponse(data) except Exception as e: # Fallback silently to SQLite pass query = f"SELECT * FROM {self.table_name}" params = [] where_clauses = [] for op, col, val in self.filters: if op == 'eq': where_clauses.append(f"{col} = ?") params.append(val) elif op == 'lte': where_clauses.append(f"{col} <= ?") params.append(val) if where_clauses: query += " WHERE " + " AND ".join(where_clauses) # Order by if self.order_by: order_clauses = [] for col, desc in self.order_by: direction = "DESC" if desc else "ASC" order_clauses.append(f"{col} {direction}") query += " ORDER BY " + ", ".join(order_clauses) # Limit if self.limit_val is not None: query += f" LIMIT {self.limit_val}" cursor.execute(query, params) rows = cursor.fetchall() data = [dict(row) for row in rows] # Post-process for nested table relations # 1. user_reviews -> problems if self.table_name == 'user_reviews' and self.op_data and 'problems(*)' in self.op_data: for row in data: prob_id = row.get('problem_id') if prob_id: cursor.execute("SELECT * FROM problems WHERE id = ?", (prob_id,)) prob_row = cursor.fetchone() if prob_row: row['problems'] = dict(prob_row) else: # Fallback to real Supabase client for public problems if self.real_client: try: res_p = self.real_client.table('problems').select('*').eq('id', prob_id).execute() row['problems'] = res_p.data[0] if res_p.data else None except Exception: row['problems'] = None else: row['problems'] = None # 2. problems -> user_reviews elif self.table_name == 'problems' and self.op_data and 'user_reviews(' in self.op_data: for row in data: prob_id = row.get('id') if prob_id: cursor.execute("SELECT * FROM user_reviews WHERE problem_id = ? AND user_id = 'sandbox-user-id'", (prob_id,)) rev_rows = cursor.fetchall() row['user_reviews'] = [dict(r) for r in rev_rows] return MockResponse(data) elif self.op == 'insert': data_list = self.op_data if isinstance(self.op_data, list) else [self.op_data] inserted_rows = [] for row_data in data_list: # Auto generate UUID if not exists or None if 'id' in row_data and row_data['id'] is None: row_data['id'] = str(uuid.uuid4()) elif self.table_name in ('problems', 'user_reviews', 'user_journey_progress', 'challenge_history') and 'id' not in row_data: row_data['id'] = str(uuid.uuid4()) columns = list(row_data.keys()) placeholders = ", ".join(["?"] * len(columns)) query = f"INSERT INTO {self.table_name} ({', '.join(columns)}) VALUES ({placeholders})" cursor.execute(query, list(row_data.values())) inserted_rows.append(row_data) conn.commit() return MockResponse(inserted_rows) elif self.op == 'update': columns = list(self.op_data.keys()) set_clause = ", ".join([f"{col} = ?" for col in columns]) params = list(self.op_data.values()) query = f"UPDATE {self.table_name} SET {set_clause}" where_clauses = [] where_params = [] for op, col, val in self.filters: if op == 'eq': where_clauses.append(f"{col} = ?") where_params.append(val) elif op == 'lte': where_clauses.append(f"{col} <= ?") where_params.append(val) if where_clauses: query += " WHERE " + " AND ".join(where_clauses) cursor.execute(query, params + where_params) conn.commit() # Fetch and return updated rows select_query = f"SELECT * FROM {self.table_name}" if where_clauses: select_query += " WHERE " + " AND ".join(where_clauses) cursor.execute(select_query, where_params) rows = cursor.fetchall() return MockResponse([dict(row) for row in rows]) elif self.op == 'upsert': row_data = self.op_data if not isinstance(self.op_data, list) else self.op_data[0] if self.table_name == 'user_journey_progress': if 'id' not in row_data: row_data['id'] = str(uuid.uuid4()) columns = list(row_data.keys()) placeholders = ", ".join(["?"] * len(columns)) query = f""" INSERT INTO user_journey_progress ({', '.join(columns)}) VALUES ({placeholders}) ON CONFLICT(user_id, problem_number) DO UPDATE SET viewed_at = excluded.viewed_at """ cursor.execute(query, list(row_data.values())) else: columns = list(row_data.keys()) placeholders = ", ".join(["?"] * len(columns)) query = f"INSERT OR REPLACE INTO {self.table_name} ({', '.join(columns)}) VALUES ({placeholders})" cursor.execute(query, list(row_data.values())) conn.commit() return MockResponse([row_data]) elif self.op == 'delete': query = f"DELETE FROM {self.table_name}" params = [] where_clauses = [] for op, col, val in self.filters: if op == 'eq': where_clauses.append(f"{col} = ?") params.append(val) if where_clauses: query += " WHERE " + " AND ".join(where_clauses) cursor.execute(query, params) conn.commit() return MockResponse([]) finally: conn.close() class MockSupabaseClient: def __init__(self): self.current_user_id = 'sandbox-user-id' # Ensure database and tables exist init_db() # Initialize a real Supabase client for reading public data from supabase import create_client import os import toml url = os.environ.get("SUPABASE_URL") key = os.environ.get("SUPABASE_KEY") if not url or not key: try: secrets_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".streamlit", "secrets.toml") if os.path.exists(secrets_path): secrets = toml.load(secrets_path) url = url or secrets.get("SUPABASE_URL") key = key or secrets.get("SUPABASE_KEY") except Exception: pass self.real_client = None if url and key: try: self.real_client = create_client(url, key) except Exception as e: print(f"Error creating real client in sandbox: {e}") def table(self, table_name): return QueryBuilder(table_name, real_client=self.real_client) def get_sandbox_client(): return MockSupabaseClient()