import toml from fastapi import FastAPI, Header, HTTPException, Depends from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from supabase import create_client, Client, ClientOptions import db import srs_logic import llm_reviewer import gdrive_sync import sandbox import sql_playground from datetime import datetime, timezone DAILY_PYTHON_CHALLENGE_CACHE = {} # maps YYYY-MM-DD to challenge dict ACTIVE_SQL_CHALLENGES = {} # maps user_id to active challenge dict def extract_starter_code(reference_code: str) -> str: if not reference_code: return "" lines = reference_code.splitlines() starter_lines = [] found_class = False found_def = False in_def = False for line in lines: stripped = line.strip() # If we see a class, add it if stripped.startswith("class ") and not in_def: starter_lines.append(line) found_class = True continue # If we see def, start adding lines if stripped.startswith("def ") and not in_def: in_def = True starter_lines.append(line) check_line = stripped.split('#')[0].strip() if check_line.endswith(':'): in_def = False indent = len(line) - len(line.lstrip()) starter_lines.append(" " * (indent + 4) + "pass") found_def = True break continue # If we are inside a multi-line def, keep adding lines if in_def: starter_lines.append(line) check_line = stripped.split('#')[0].strip() if check_line.endswith(':'): in_def = False def_indent = 4 for l in reversed(starter_lines): if l.strip().startswith("def "): def_indent = len(l) - len(l.lstrip()) break starter_lines.append(" " * (def_indent + 4) + "pass") found_def = True break if not starter_lines: return reference_code return "\n".join(starter_lines) app = FastAPI(title="AlgoSpaced API") # Configure CORS app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:5173", "http://127.0.0.1:5173"], # Vite React Dev Server allow_origin_regex=r"http://(localhost|127\.0\.0\.1|192\.168\.\d+\.\d+|10\.\d+\.\d+\.\d+|172\.(1[6-9]|2\d|3[0-1])\.\d+\.\d+):5173", allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) import os # Load secrets for Supabase configuration SUPABASE_URL = os.environ.get("SUPABASE_URL") SUPABASE_KEY = os.environ.get("SUPABASE_KEY") if not SUPABASE_URL or not SUPABASE_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) SUPABASE_URL = SUPABASE_URL or secrets.get("SUPABASE_URL") SUPABASE_KEY = SUPABASE_KEY or secrets.get("SUPABASE_KEY") except Exception as e: print(f"Error loading secrets.toml: {e}") # Dependency to resolve a Supabase client configured for the specific request user session def get_supabase_client(authorization: str = Header(None)) -> Client: if not authorization or not authorization.startswith("Bearer "): raise HTTPException(status_code=401, detail="Missing or invalid Authorization header") access_token = authorization.split(" ")[1] if access_token == "sandbox": import sandbox_db return sandbox_db.get_sandbox_client() try: # Initialize a Supabase client authenticated as this specific user client = create_client( SUPABASE_URL, SUPABASE_KEY, options=ClientOptions(headers={"Authorization": f"Bearer {access_token}"}) ) # Validate session actively by querying user details user_res = client.auth.get_user(access_token) if not user_res or not user_res.user: raise HTTPException(status_code=401, detail="Invalid Supabase authentication session") client.current_user_id = user_res.user.id return client except Exception as e: raise HTTPException(status_code=401, detail=f"Authentication failed: {str(e)}") @app.get("/api/config") def get_public_config(): return { "supabaseUrl": SUPABASE_URL, "supabaseAnonKey": SUPABASE_KEY } @app.get("/api/streak") def get_streak(client: Client = Depends(get_supabase_client)): streak_data = db.get_active_streak(client) if streak_data: streak_data = srs_logic.lazy_check_streak(streak_data) db.update_streak(streak_data, client) return streak_data @app.get("/api/reviews/due") def get_due_reviews(client: Client = Depends(get_supabase_client)): reviews = db.get_due_reviews(client) for review in reviews: if "problems" in review and review["problems"]: prob = review["problems"] prob["starter_code"] = extract_starter_code(prob.get("reference_code", "")) return reviews @app.get("/api/reviews/problems/{problem_number}") def get_problems_by_number(problem_number: int, client: Client = Depends(get_supabase_client)): problems = db.get_problems_by_number(problem_number, client) for prob in problems: prob["starter_code"] = extract_starter_code(prob.get("reference_code", "")) return problems @app.get("/api/reviews/problems/id/{problem_id}") def get_reviews_by_problem_id(problem_id: str, client: Client = Depends(get_supabase_client)): reviews = db.get_reviews_by_problem_id(problem_id, client) for review in reviews: if "problems" in review and review["problems"]: prob = review["problems"] prob["starter_code"] = extract_starter_code(prob.get("reference_code", "")) return reviews class CompleteReviewRequest(BaseModel): review_id: str box_level: int times_correct: int total_attempts: int rating: str # "Correct", "Mixed", "Incorrect" @app.post("/api/reviews/complete") def complete_review(req: CompleteReviewRequest, client: Client = Depends(get_supabase_client)): new_box = srs_logic.evaluate_new_box(req.box_level, req.rating) next_review_date = srs_logic.calculate_next_review(new_box).isoformat() new_times_correct = req.times_correct + (1 if req.rating == "Correct" else 0) new_total_attempts = req.total_attempts + 1 db.update_review(req.review_id, { "box_level": new_box, "next_review": next_review_date, "times_correct": new_times_correct, "total_attempts": new_total_attempts }, client) streak_data = db.get_active_streak(client) if streak_data: new_streak = srs_logic.increment_streak(streak_data) db.update_streak(new_streak, client) return { "success": True, "new_box": new_box, "next_review": next_review_date } class EvaluateCodeRequest(BaseModel): problem_name: str pattern: str optimal_time: str optimal_space: str code: str @app.post("/api/reviews/evaluate") def evaluate_code( req: EvaluateCodeRequest, client: Client = Depends(get_supabase_client), x_gemini_api_key: str = Header(None) ): eval_res = llm_reviewer.evaluate_code( req.problem_name, req.pattern, req.optimal_time, req.optimal_space, req.code, custom_api_key=x_gemini_api_key ) if "error" in eval_res: raise HTTPException(status_code=500, detail=eval_res["error"]) return eval_res class ExplainCodeRequest(BaseModel): problem_number: int method: str @app.post("/api/reviews/explain") def explain_review_code( req: ExplainCodeRequest, client: Client = Depends(get_supabase_client), x_gemini_api_key: str = Header(None) ): try: user_id = db.get_current_user_id(client) if not user_id: raise HTTPException(status_code=401, detail="Authentication failed") res = client.table('problems').select('*').eq('user_id', user_id).eq('problem_number', req.problem_number).execute() if not res.data: raise HTTPException(status_code=404, detail=f"No local entry found for LeetCode {req.problem_number}") problem = None for p in res.data: pattern = p.get('pattern', '') name = p.get('name', '') if pattern.lower() == req.method.lower() or name.lower() == req.method.lower(): problem = p break if not problem: problem = res.data[0] reference_code = problem.get('reference_code') if not reference_code: raise HTTPException(status_code=400, detail="No reference code stored for this problem") optimal_time = problem.get('optimal_time', 'O(N)') optimal_space = problem.get('optimal_space', 'O(N)') problem_name = problem.get('name', '') eval_res = llm_reviewer.explain_code( problem_name=problem_name, code=reference_code, optimal_time=optimal_time, optimal_space=optimal_space, custom_api_key=x_gemini_api_key ) if "error" in eval_res: raise HTTPException(status_code=500, detail=eval_res["error"]) return eval_res except HTTPException: raise except Exception as e: import traceback db.log_trace(f"Explain endpoint error: {str(e)}\n{traceback.format_exc()}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/journey-progress") def get_journey_progress(client: Client = Depends(get_supabase_client)): return list(db.get_user_journey_progress(client)) class ToggleJourneyRequest(BaseModel): problem_number: int @app.post("/api/journey-progress/toggle") def toggle_journey(req: ToggleJourneyRequest, client: Client = Depends(get_supabase_client)): db.toggle_journey_progress(req.problem_number, client) return {"success": True} @app.get("/api/journey-progress/titles") def get_journey_titles(client: Client = Depends(get_supabase_client)): return db.get_problem_titles_by_number(client) @app.post("/api/sync") def trigger_sync(client: Client = Depends(get_supabase_client)): user_id = db.get_current_user_id(client) if user_id == "sandbox-user-id": return {"message": "Google Drive Sync is not available in Sandbox Mode. Please log in to sync your notes."} try: res = gdrive_sync.sync_gdrive(client) return {"message": res} except Exception as e: import traceback err_msg = f"Sync endpoint exception: {str(e)}\n{traceback.format_exc()}" db.log_trace(err_msg) raise HTTPException(status_code=500, detail=err_msg) class SubmitPythonRequest(BaseModel): code: str @app.get("/api/challenges/python/daily") def get_daily_python_challenge( client: Client = Depends(get_supabase_client), x_gemini_api_key: str = Header(None) ): today_str = datetime.now(timezone.utc).strftime("%Y-%m-%d") if today_str not in DAILY_PYTHON_CHALLENGE_CACHE: challenge = llm_reviewer.generate_daily_python_challenge(custom_api_key=x_gemini_api_key) if "error" in challenge: raise HTTPException(status_code=500, detail=f"Failed to generate challenge: {challenge['error']}") DAILY_PYTHON_CHALLENGE_CACHE[today_str] = challenge challenge = DAILY_PYTHON_CHALLENGE_CACHE[today_str] user_id = db.get_current_user_id(client) completed = False if user_id: try: res = client.table('challenge_history')\ .select('id')\ .eq('user_id', user_id)\ .eq('challenge_type', 'python')\ .eq('challenge_title', challenge.get("title", ""))\ .execute() completed = len(res.data) > 0 if res.data else False except Exception: pass score_record = db.get_user_score(client) total_score = score_record["total_score"] if score_record else 0 return { "title": challenge.get("title"), "description": challenge.get("description"), "difficulty": challenge.get("difficulty"), "points": challenge.get("points"), "starter_code": challenge.get("starter_code"), "completed": completed, "total_score": total_score } @app.post("/api/challenges/python/submit") def submit_python_challenge(req: SubmitPythonRequest, client: Client = Depends(get_supabase_client)): today_str = datetime.now(timezone.utc).strftime("%Y-%m-%d") if today_str not in DAILY_PYTHON_CHALLENGE_CACHE: challenge = llm_reviewer.generate_daily_python_challenge() if "error" in challenge: raise HTTPException(status_code=500, detail=f"No challenge available: {challenge['error']}") DAILY_PYTHON_CHALLENGE_CACHE[today_str] = challenge challenge = DAILY_PYTHON_CHALLENGE_CACHE[today_str] test_cases = challenge.get("test_cases", []) run_res = sandbox.run_python_sandbox(req.code, test_cases) if not run_res.get("success", False): return { "success": False, "error": run_res.get("error", "Unknown execution error"), "sandbox_type": run_res.get("sandbox_type") } results = run_res.get("results", []) all_passed = len(results) > 0 and all(r.get("status") == "passed" for r in results) points_awarded = 0 completed_before = False total_score = 0 user_id = db.get_current_user_id(client) if all_passed and user_id: title = challenge.get("title", "Daily Python Challenge") points = challenge.get("points", 100) try: hist_res = client.table('challenge_history')\ .select('id')\ .eq('user_id', user_id)\ .eq('challenge_type', 'python')\ .eq('challenge_title', title)\ .execute() completed_before = len(hist_res.data) > 0 if hist_res.data else False except Exception: pass if not completed_before: db.increment_user_score(points, "python", client) db.add_challenge_history("python", title, points, client) points_awarded = points score_record = db.get_user_score(client) if score_record: total_score = score_record["total_score"] return { "success": True, "all_passed": all_passed, "results": results, "points_awarded": points_awarded, "completed_before": completed_before, "total_score": total_score, "sandbox_type": run_res.get("sandbox_type"), "sandbox_warning": run_res.get("sandbox_warning") } class SQLQueryRequest(BaseModel): query: str @app.post("/api/challenges/mysql/init") def init_mysql_challenge( client: Client = Depends(get_supabase_client), x_gemini_api_key: str = Header(None) ): import os user_id = db.get_current_user_id(client) if not user_id: raise HTTPException(status_code=401, detail="Authentication failed") challenge = llm_reviewer.generate_sql_challenge(custom_api_key=x_gemini_api_key) if "error" in challenge: raise HTTPException(status_code=500, detail=f"Failed to generate SQL challenge: {challenge['error']}") init_res = sql_playground.init_user_db(user_id, challenge) if not init_res.get("success", False): raise HTTPException(status_code=500, detail=f"Failed to initialize database: {init_res.get('error')}") ACTIVE_SQL_CHALLENGES[user_id] = challenge score_record = db.get_user_score(client) total_score = score_record["total_score"] if score_record else 0 return { "success": True, "title": challenge.get("title"), "objective": challenge.get("objective"), "points": challenge.get("points"), "schema": init_res.get("schema"), "challenge_type": challenge.get("challenge_type"), "total_score": total_score } @app.get("/api/challenges/mysql/status") def get_mysql_status(client: Client = Depends(get_supabase_client)): import os user_id = db.get_current_user_id(client) if not user_id: raise HTTPException(status_code=401, detail="Authentication failed") if user_id not in ACTIVE_SQL_CHALLENGES: return {"active": False, "message": "No active SQL session. Please initialize with 'db init'."} challenge = ACTIVE_SQL_CHALLENGES[user_id] db_path = sql_playground.get_db_path(user_id) schema = {} if os.path.exists(db_path): import sqlite3 try: conn = sqlite3.connect(db_path) schema = sql_playground.get_db_schema(conn) conn.close() except Exception: pass completed = False try: res = client.table('challenge_history')\ .select('id')\ .eq('user_id', user_id)\ .eq('challenge_type', 'mysql')\ .eq('challenge_title', challenge.get("title", ""))\ .execute() completed = len(res.data) > 0 if res.data else False except Exception: pass score_record = db.get_user_score(client) total_score = score_record["total_score"] if score_record else 0 return { "active": True, "title": challenge.get("title"), "objective": challenge.get("objective"), "points": challenge.get("points"), "challenge_type": challenge.get("challenge_type"), "schema": schema, "completed": completed, "total_score": total_score } @app.post("/api/challenges/mysql/query") def run_mysql_query(req: SQLQueryRequest, client: Client = Depends(get_supabase_client)): user_id = db.get_current_user_id(client) if not user_id: raise HTTPException(status_code=401, detail="Authentication failed") if user_id not in ACTIVE_SQL_CHALLENGES: raise HTTPException(status_code=400, detail="No active SQL session. Type 'db init' first.") challenge = ACTIVE_SQL_CHALLENGES[user_id] run_res = sql_playground.run_user_query(user_id, req.query, challenge) if not run_res.get("success", False): return { "success": False, "error": run_res.get("error", "Unknown database error") } is_correct = run_res.get("is_correct", False) points_awarded = 0 completed_before = False total_score = 0 if is_correct: title = challenge.get("title", "SQL Challenge") points = challenge.get("points", 150) try: hist_res = client.table('challenge_history')\ .select('id')\ .eq('user_id', user_id)\ .eq('challenge_type', 'mysql')\ .eq('challenge_title', title)\ .execute() completed_before = len(hist_res.data) > 0 if hist_res.data else False except Exception: pass if not completed_before: db.increment_user_score(points, "mysql", client) db.add_challenge_history("mysql", title, points, client) points_awarded = points score_record = db.get_user_score(client) if score_record: total_score = score_record["total_score"] return { "success": True, "is_correct": is_correct, "columns": run_res.get("columns", []), "rows": run_res.get("rows", []), "rows_affected": run_res.get("rows_affected", 0), "schema": run_res.get("schema", {}), "points_awarded": points_awarded, "completed_before": completed_before, "total_score": total_score } class CreateProblemRequest(BaseModel): name: str pattern: str difficulty: str reference_code: str description: str = "" class UpdateProblemRequest(BaseModel): name: str pattern: str difficulty: str reference_code: str description: str = "" @app.get("/api/explorer/problems") def get_explorer_problems(client: Client = Depends(get_supabase_client)): try: res = client.table('problems')\ .select('id, name, pattern, difficulty, reference_code, description, user_reviews(box_level, next_review, last_reviewed, times_correct, total_attempts)')\ .execute() problems = res.data if res.data else [] for p in problems: reviews = p.pop("user_reviews", []) if reviews and isinstance(reviews, list) and len(reviews) > 0: rev = reviews[0] p["box_level"] = rev.get("box_level", 1) p["next_review"] = rev.get("next_review") p["last_reviewed"] = rev.get("last_reviewed") p["times_correct"] = rev.get("times_correct", 0) p["total_attempts"] = rev.get("total_attempts", 0) else: p["box_level"] = None p["next_review"] = None p["last_reviewed"] = None p["times_correct"] = 0 p["total_attempts"] = 0 return problems except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/explorer/problems/create") def create_explorer_problem(req: CreateProblemRequest, client: Client = Depends(get_supabase_client)): user_id = db.get_current_user_id(client) if not user_id: raise HTTPException(status_code=401, detail="Authentication failed") if not req.name.strip(): raise HTTPException(status_code=400, detail="Problem name cannot be empty") existing = db.get_problem_by_name(req.name, client) if existing: raise HTTPException(status_code=400, detail=f"A problem named '{req.name}' already exists.") try: prob_data = { "name": req.name.strip(), "pattern": req.pattern.strip() or "General", "difficulty": req.difficulty or "Medium", "reference_code": req.reference_code, "description": req.description } res_prob = db.insert_problem(prob_data, client) if not res_prob or not res_prob.data: raise HTTPException(status_code=500, detail="Failed to create problem record") new_prob_id = res_prob.data[0]['id'] review_data = { "problem_id": new_prob_id, "box_level": 1, "next_review": datetime.now(timezone.utc).isoformat(), "times_correct": 0, "total_attempts": 0 } db.insert_review(review_data, client) return {"success": True, "problem_id": new_prob_id} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.put("/api/explorer/problems/{problem_id}") def update_explorer_problem(problem_id: str, req: UpdateProblemRequest, client: Client = Depends(get_supabase_client)): user_id = db.get_current_user_id(client) if not user_id: raise HTTPException(status_code=401, detail="Authentication failed") try: update_data = { "name": req.name.strip(), "pattern": req.pattern.strip() or "General", "difficulty": req.difficulty or "Medium", "reference_code": req.reference_code, "description": req.description } res = db.update_problem(problem_id, update_data, client) if not res or not res.data: raise HTTPException(status_code=500, detail="Failed to update problem record") return {"success": True} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/explorer/export") def export_solutions_zip(client: Client = Depends(get_supabase_client)): import io import zipfile from fastapi.responses import StreamingResponse import re user_id = db.get_current_user_id(client) if not user_id: raise HTTPException(status_code=401, detail="Authentication failed") try: res = client.table('problems')\ .select('name, pattern, difficulty, reference_code, description')\ .eq('user_id', user_id)\ .execute() problems = res.data if res.data else [] zip_buffer = io.BytesIO() with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zip_file: for p in problems: name = p.get("name", "Unnamed_Problem") pattern = p.get("pattern", "General") difficulty = p.get("difficulty", "Medium") code = p.get("reference_code", "") description = p.get("description", "") safe_name = re.sub(r'[^\w\-]', '_', name) filename = f"{safe_name}.py" if not code or not code.strip(): desc_comment = f"\n# Description: {description}" if description else "" code = f"# Problem: {name}\n# Pattern: {pattern}\n# Difficulty: {difficulty}{desc_comment}\n# Write your solution here!\n\ndef solve():\n pass\n" else: if description: formatted_desc = "\n".join([f"# {line}" for line in description.splitlines()]) code = f"# Description:\n{formatted_desc}\n\n" + code zip_file.writestr(filename, code) zip_buffer.seek(0) headers = { "Content-Disposition": "attachment; filename=algospaced_solutions.zip" } return StreamingResponse( zip_buffer, media_type="application/zip", headers=headers ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # Serve static files from Vite frontend in production from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse frontend_dist = os.path.join(os.path.dirname(__file__), "algospaced-ui", "dist") if os.path.exists(frontend_dist): # Mount the assets directory (contains compiled js, css, images) app.mount("/assets", StaticFiles(directory=os.path.join(frontend_dist, "assets")), name="assets") @app.get("/favicon.ico", include_in_schema=False) def serve_favicon(): fav = os.path.join(frontend_dist, "favicon.ico") if os.path.exists(fav): return FileResponse(fav) raise HTTPException(status_code=404) @app.get("/{fallback_path:path}") def serve_frontend(fallback_path: str): # Prevent intercepting API routes if fallback_path.startswith("api/"): raise HTTPException(status_code=404, detail="API endpoint not found") index_file = os.path.join(frontend_dist, "index.html") if os.path.exists(index_file): return FileResponse(index_file) raise HTTPException(status_code=404, detail="Frontend index.html not found")