""" Supabase data storage for SpeechKid evaluations. Stores: - Recording WAV files → private 'recordings' bucket (per-user folders) - Evaluation results → 'evaluations' table (RLS-protected per user) - Server logs → captured during scoring, saved per evaluation Writes use the service_role key (server-to-server). Reads (dashboard) use the user's token + anon key so RLS filters automatically. """ import os import threading from datetime import date, datetime, timedelta, timezone # Bump this whenever the scoring logic/thresholds change. Stored with every # evaluation so we can later tell which model version produced each label — # essential for measuring accuracy improvements as the model evolves. MODEL_VERSION = "2026.06.14-shk-wetness-dormant" _client = None def _get_client(): """Lazy-init Supabase client (service_role). Returns None if not configured.""" global _client if _client is not None: return _client url = os.environ.get("SUPABASE_URL") key = os.environ.get("SUPABASE_KEY") if not url or not key: print("[DATA] Supabase not configured — SUPABASE_URL or SUPABASE_KEY missing") return None from supabase import create_client _client = create_client(url, key) print("[DATA] Supabase client initialized") return _client def _get_user_client(token: str): """ Create a Supabase client scoped to a specific user. Uses the anon key as the API key and the user's access token for Authorization. This way RLS policies filter by auth.uid() automatically — the user can only see their own data without any server-side filtering. """ url = os.environ.get("SUPABASE_URL") anon_key = os.environ.get("SUPABASE_ANON_KEY") if not url or not anon_key: print("[DATA] SUPABASE_ANON_KEY not configured — cannot create user client") return None from supabase import create_client client = create_client(url, anon_key) client.postgrest.auth(token) return client def extract_user_id(token: str) -> str: """ Verify a Supabase access token and extract the user_id. Uses the Supabase Auth API (via service_role client) to cryptographically verify the token — not just base64-decoding it. This ensures the token was actually issued by Supabase and hasn't been tampered with. Returns None if no token provided or verification fails. """ if not token: return None try: client = _get_client() if client is None: return None response = client.auth.get_user(token) user_id = response.user.id print(f"[AUTH] User verified: {user_id[:8]}...") return user_id except Exception as e: print(f"[AUTH] Token verification failed: {e}") return None def get_user_dashboard(token: str) -> dict: """ Query the user's evaluations and compute dashboard stats. Uses the user's token + anon key so RLS filters by user_id automatically. Returns dict with total_sessions, total_words, avg_score, streak_days, and recent_evaluations. """ client = _get_user_client(token) if client is None: return None # Fetch all evaluations for this user (RLS filters automatically) response = ( client.table("evaluations") .select("score, created_at, word, status") .order("created_at", desc=True) .execute() ) rows = response.data or [] if not rows: return { "total_sessions": 0, "total_words": 0, "avg_score": 0, "streak_days": 0, "recent_evaluations": [], } # --- Total words (total evaluations) --- total_words = len(rows) # --- Average score --- scores = [r["score"] for r in rows if r.get("score") is not None] avg_score = round(sum(scores) / len(scores)) if scores else 0 # --- Distinct practice dates --- practice_dates = set() for r in rows: try: dt = datetime.fromisoformat(r["created_at"].replace("Z", "+00:00")) practice_dates.add(dt.date()) except (ValueError, TypeError): pass total_sessions = len(practice_dates) # --- Streak: consecutive days counting back from today --- today = date.today() streak = 0 check_date = today while check_date in practice_dates: streak += 1 check_date -= timedelta(days=1) # --- Recent evaluations (last 20) --- recent = [] for r in rows[:20]: score_val = r.get("score", 0) recent.append({ "word": r.get("word", ""), "score": score_val, "status": "החלצה" if score_val >= 70 else "בוש הסנ", "created_at": r.get("created_at", ""), }) return { "total_sessions": total_sessions, "total_words": total_words, "avg_score": avg_score, "streak_days": streak, "recent_evaluations": recent, } def get_user_progress(token: str) -> dict: """ Return the user's level progress. unlocked_level = highest level where passed=true, plus 1 (minimum 1). level_stars = all rows for this user ordered by level. """ client = _get_user_client(token) if client is None: return None response = ( client.table("user_progress") .select("level, stars, passed, word_count") .order("level") .execute() ) rows = response.data or [] passed_levels = [r["level"] for r in rows if r.get("passed")] unlocked_level = max(passed_levels) + 1 if passed_levels else 1 level_stars = [ { "level": r["level"], "stars": r.get("stars", 0) or 0, "passed": bool(r.get("passed", False)), "word_count": r.get("word_count", 0) or 0, } for r in rows ] return {"unlocked_level": unlocked_level, "level_stars": level_stars} def save_user_progress(token: str, level: int, stars: int, passed: bool, word_count: int) -> bool: """ Upsert a user_progress row with merge semantics: - stars = max(old, new) - passed = old OR new (once passed, stays passed) - word_count = max(old, new) Uses the user's token so RLS scopes the write to their own rows and auth.uid() populates user_id via the column default. Returns True on success, False on failure. """ client = _get_user_client(token) if client is None: return False existing = ( client.table("user_progress") .select("stars, passed, word_count") .eq("level", level) .limit(1) .execute() ) rows = existing.data or [] if rows: old = rows[0] merged = { "stars": max(old.get("stars", 0) or 0, stars), "passed": bool(old.get("passed", False)) or bool(passed), "word_count": max(old.get("word_count", 0) or 0, word_count), "updated_at": datetime.now(timezone.utc).isoformat(), } client.table("user_progress").update(merged).eq("level", level).execute() else: client.table("user_progress").insert({ "level": level, "stars": stars, "passed": bool(passed), "word_count": word_count, }).execute() return True def save_evaluation( evaluation_id: str, word: str, result: dict, wav_data: bytes, server_logs: str, processing_ms: int, user_id: str = None, ): """ Save a complete evaluation to Supabase in a background thread. Uploads the recording WAV to the private 'recordings' bucket (organized by user_id for RLS), then inserts the evaluation record. Runs in a daemon thread so the API response is not delayed. Failures are logged but never propagate to the caller. """ def _save(): try: client = _get_client() if client is None: return # --- Upload recording to private storage bucket --- folder = user_id or "anonymous" storage_path = f"{folder}/{evaluation_id}.wav" try: client.storage.from_("recordings").upload( storage_path, wav_data, file_options={"content-type": "audio/wav"}, ) print(f"[DATA] Recording uploaded: {storage_path}") except Exception as e: print(f"[DATA] Recording upload failed: {e}") storage_path = None # --- Insert evaluation record --- # Stamp the model version into the details JSON (schema-safe — no # DB migration needed) so every recording is traceable to the # pipeline that scored it. details = dict(result.get("details", {})) details["model_version"] = MODEL_VERSION record = { "id": evaluation_id, "word": word, "score": result.get("score", 0), "status": result.get("status", "ERROR"), "diagnosis": result.get("diagnosis", "UNKNOWN"), "feedback": result.get("feedback", ""), "details": details, "evidence": result.get("evidence", {}), "recording_path": storage_path, "server_logs": server_logs, "processing_ms": processing_ms, "user_id": user_id, } client.table("evaluations").insert(record).execute() print(f"[DATA] Evaluation saved: {evaluation_id} (user={folder[:8]}...)") except Exception as e: print(f"[DATA] Failed to save evaluation: {e}") thread = threading.Thread(target=_save, daemon=True) thread.start()