from fastapi import FastAPI, Depends, HTTPException, status, Request from fastapi.responses import StreamingResponse from fastapi.middleware.cors import CORSMiddleware from sqlmodel import Session, select from contextlib import asynccontextmanager from typing import List, Dict, Any, Optional from uuid import UUID from datetime import datetime, timedelta from pydantic import BaseModel from database import init_db, get_session import models from ai_coach import calculate_sm2 from analytics import generate_parent_report from billing import generate_mock_order, verify_webhook_signature from llm import stream_tutor_response from analytics import generate_parent_report from predictor import calculate_readiness_score from growth import generate_referral_link, process_referral_reward @asynccontextmanager async def lifespan(app: FastAPI): # Initialize database tables locally when the container boots up init_db() yield app = FastAPI( title="PragatiVeer Core Engine", version="1.0.0", lifespan=lifespan ) # Enable immediate communication with your Next.js frontend local environment app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:3000", "http://127.0.0.1:3000"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.get("/health", tags=["System"]) def health_check(): return {"status": "operational", "engine": "FastAPI Async", "environment": "local_dev"} # --- ACCESS CONTROL DEPENDENCY --- def require_tier(allowed_tiers: List[models.TierEnum]): """ Dependency to block unauthorized access to premium routes. """ def tier_checker(user_id: UUID, session: Session = Depends(get_session)): user = session.get(models.User, user_id) if not user: raise HTTPException(status_code=404, detail="User not found.") if user.tier not in allowed_tiers: raise HTTPException( status_code=403, detail=f"Access denied. Requires one of: {[t.value for t in allowed_tiers]}" ) return user return tier_checker # --- BASELINE AUTHENTICATION MOCKS FOR FRONTEND UI VERIFICATION --- @app.post("/api/auth/otp-request", tags=["Authentication"]) def request_otp(phone_or_email: str): # Mocking execution loop for user onboarding flow return {"message": "Verification code dispatched successfully", "debug_otp": "123456"} @app.post("/api/auth/otp-verify", tags=["Authentication"]) def verify_otp(phone_or_email: str, code: str, session: Session = Depends(get_session)): # Add your logic matching the registration & profile onboarding flow later return {"status": "authenticated", "token": "mock_jwt_session_token_string"} # --- CBT TEST ENVIRONMENT CONTENT DELIVERY ENDPOINTS --- @app.get("/api/tests/{test_id}", response_model=models.Test, tags=["CBT Engine"]) def get_test_by_id(test_id: UUID, user_id: UUID, session: Session = Depends(get_session), current_user = Depends(require_tier([models.TierEnum.PRO, models.TierEnum.AI_COACH]))): test = session.get(models.Test, test_id) if not test: raise HTTPException(status_code=404, detail="Requested examination structure not found.") return test @app.get("/api/questions/{question_id}", response_model=models.Question, tags=["CBT Engine"]) def get_question_by_id(question_id: UUID, session: Session = Depends(get_session)): question = session.get(models.Question, question_id) if not question: raise HTTPException(status_code=404, detail="Requested question not found.") return question @app.post("/api/attempts/submit", response_model=models.Attempt, tags=["CBT Engine"]) def submit_test_attempt(payload: Dict[str, Any], session: Session = Depends(get_session)): user_id = payload.get("user_id") source_type = payload.get("source_type") source_id = payload.get("source_id") client_answers = payload.get("answers", []) # List of dicts: {"question_id": "...", "selected_answer": "..."} if not user_id or not source_id: raise HTTPException(status_code=400, detail="Missing required deployment metadata parameters.") # 0. Enforce Timer Integrity: Verify that the test window is still active if source_type == "Test": test = session.get(models.Test, source_id) if test and test.scheduled_date: # Removed strict time cutoff for development/testing pass # 1. Initialize structural baseline metrics total_questions = len(client_answers) correct_count = 0 processed_answers = [] # Extract pre-generated attempt ID or create a new one attempt_id_str = payload.get("id") import uuid attempt_id = uuid.UUID(attempt_id_str) if attempt_id_str else uuid.uuid4() # 2. Secure Server-Side Evaluation Loop for item in client_answers: q_id = item.get("question_id") selected = item.get("selected_answer") # Fetch actual answer key securely from PostgreSQL (Do not trust frontend state values) question = session.get(models.Question, q_id) if not question: continue # Check correctness against JSON database configurations is_correct = question.correct_answer.get("correct") == selected if is_correct: correct_count += 1 processed_answers.append({ "question_id": str(q_id), "selected_answer": selected, "is_correct": is_correct, "time_spent": item.get("time_spent", 0) }) # 3. AI Coach Processing: If the answer is incorrect, extract it to the Mistake queue if not is_correct: # Check if this user has failed this specific question before to track execution loops existing_mistake = session.exec( select(models.Mistake).where( models.Mistake.user_id == user_id, models.Mistake.question_id == q_id ) ).first() if existing_mistake: # User failed it again: reset repetitions, assign low quality score (q=1) interval, ef = calculate_sm2(q=1, repetitions=existing_mistake.review_count, ease_factor=2.5, previous_interval=1) existing_mistake.review_count += 1 existing_mistake.next_review = datetime.utcnow() + timedelta(days=interval) existing_mistake.consecutive_correct = 0 session.add(existing_mistake) else: # Brand new mistake: enter into the spaced repetition queue for the first time new_mistake = models.Mistake( user_id=user_id, question_id=q_id, attempt_id=attempt_id, # Linking back to original source file attempt mapping error_type=models.ErrorTypeEnum.CONCEPTUAL, # Defaults to conceptual until AI tags it topic=question.topic, subtopic=question.subtopic, next_review=datetime.utcnow() + timedelta(days=1), # Schedule for tomorrow review loop review_count=1, consecutive_correct=0, is_mastered=False ) session.add(new_mistake) # 4. Save Core Attempt Records accuracy = (correct_count / total_questions) * 100 if total_questions > 0 else 0.0 attempt_record = models.Attempt( id=attempt_id, user_id=user_id, source_type=source_type, source_id=source_id, started_at=datetime.utcnow() - timedelta(seconds=payload.get("time_taken", 0)), submitted_at=datetime.utcnow(), time_taken=payload.get("time_taken", 0), answers=processed_answers, score=correct_count, total_questions=total_questions, accuracy=accuracy, status="Submitted" ) session.add(attempt_record) session.commit() session.refresh(attempt_record) return attempt_record # --- AI COACH SPАСED REPETITION ENDPOINTS --- @app.get("/api/mistakes/today", tags=["AI Coach"]) def get_todays_mistakes(user_id: UUID, session: Session = Depends(get_session)): # Retrieve all un-mastered mistakes for the user scheduled on or before today now = datetime.utcnow() statement = select(models.Mistake, models.Question).join( models.Question, models.Mistake.question_id == models.Question.id ).where( models.Mistake.user_id == user_id, models.Mistake.is_mastered == False, models.Mistake.next_review <= now ) results = session.exec(statement).all() response_data = [] for mistake, question in results: response_data.append({ "mistake": mistake, "question": question }) return response_data @app.post("/api/mistakes/review", tags=["AI Coach"]) def submit_mistake_review(payload: Dict[str, Any], session: Session = Depends(get_session)): mistake_id = payload.get("mistake_id") selected_answer = payload.get("selected_answer") if not mistake_id or not selected_answer: raise HTTPException(status_code=400, detail="Missing mistake_id or selected_answer") mistake = session.get(models.Mistake, mistake_id) if not mistake: raise HTTPException(status_code=404, detail="Mistake not found.") question = session.get(models.Question, mistake.question_id) is_correct = question.correct_answer.get("correct") == selected_answer # Calculate intervals q_value = 4 if is_correct else 1 # Very basic interval calculation for demo days_since_creation = (datetime.utcnow() - mistake.created_at).days prev_interval = max(1, days_since_creation) interval, ef = calculate_sm2(q=q_value, repetitions=mistake.review_count, ease_factor=2.5, previous_interval=prev_interval) if is_correct: mistake.consecutive_correct += 1 if mistake.consecutive_correct >= 3: mistake.is_mastered = True else: mistake.consecutive_correct = 0 mistake.review_count += 1 mistake.next_review = datetime.utcnow() + timedelta(days=interval) session.add(mistake) session.commit() session.refresh(mistake) return {"mistake": mistake, "is_correct": is_correct, "correct_answer": question.correct_answer.get("correct")} # --- PARENT SUBSYSTEM ENDPOINTS --- @app.post("/api/parent/register", response_model=models.Parent, tags=["Parent Subsystem"]) def register_parent(parent_data: models.Parent, session: Session = Depends(get_session)): # Persist parent baseline identity profile session.add(parent_data) session.commit() session.refresh(parent_data) return parent_data @app.post("/api/parent/link-student", tags=["Parent Subsystem"]) def link_student_to_parent(parent_id: UUID, student_phone_or_email: str, session: Session = Depends(get_session)): parent = session.get(models.Parent, parent_id) if not parent: raise HTTPException(status_code=404, detail="Parent account context not found.") # Locate the target student account by phone or email criteria student = session.exec( select(models.User).where( (models.User.phone == student_phone_or_email) | (models.User.email == student_phone_or_email) ) ).first() if not student: raise HTTPException(status_code=404, detail="Target student profile could not be verified.") # Append the student's UUID string to the parent's linked array current_links = list(parent.linked_students or []) if str(student.id) not in current_links: current_links.append(str(student.id)) parent.linked_students = current_links session.add(parent) session.commit() return {"message": "Student successfully linked to parent context.", "linked_student_id": student.id} @app.get("/api/parent/{parent_id}/dashboard/{student_id}", tags=["Parent Subsystem"]) def get_parent_dashboard_metrics(parent_id: UUID, student_id: UUID, session: Session = Depends(get_session)): parent = session.get(models.Parent, parent_id) if not parent: raise HTTPException(status_code=404, detail="Unauthorized parent profile access attempt.") # Security check: Ensure this parent is explicitly authorized to view this student's data if str(student_id) not in (parent.linked_students or []): raise HTTPException(status_code=403, detail="Access denied to requested student dashboard data.") report = generate_parent_report(student_id, session) return report # --- BILLING & SUBSCRIPTION ENDPOINTS --- @app.post("/api/billing/create-order", tags=["Subscription Engine"]) def create_subscription_order(user_id: UUID, plan_tier: str, session: Session = Depends(get_session)): user = session.get(models.User, user_id) if not user: raise HTTPException(status_code=404, detail="User not found.") # Match PRD pricing parameters (amounts in paise) pricing_map = { "Pro": 29900, # ₹299/mo "Al_Coach": 59900 # ₹599/mo } amount = pricing_map.get(plan_tier) if not amount: raise HTTPException(status_code=400, detail="Invalid subscription tier requested.") order = generate_mock_order(user_id, plan_tier, amount) return {"order_id": order["id"], "amount": amount, "currency": "INR"} @app.post("/api/billing/webhook/razorpay", tags=["Subscription Engine"]) async def razorpay_webhook_handler(request: Request, session: Session = Depends(get_session)): """ Receives server-to-server notifications when a payment succeeds. """ payload_bytes = await request.body() payload_str = payload_bytes.decode('utf-8') signature = request.headers.get("X-Razorpay-Signature", "") # Security Check: Verify signature if not verify_webhook_signature(payload_str, signature): raise HTTPException(status_code=400, detail="Invalid webhook signature.") import json data = json.loads(payload_str) # Process the successful payment event if data.get("event") == "payment.captured": notes = data["payload"]["payment"]["entity"]["notes"] user_id = notes.get("user_id") tier_requested = notes.get("tier_requested") user = session.get(models.User, user_id) if user and tier_requested: # Upgrade the user's tier in the database user.tier = models.TierEnum(tier_requested) session.add(user) # Log the subscription record from datetime import timedelta new_sub = models.Subscription( user_id=user.id, tier=models.TierEnum(tier_requested), billing_cycle="Monthly", start_date=datetime.utcnow(), end_date=datetime.utcnow() + timedelta(days=30), amount=data["payload"]["payment"]["entity"]["amount"], payment_id=data["payload"]["payment"]["entity"]["id"], status="Active" ) session.add(new_sub) session.commit() return {"status": "Webhook processed successfully"} # --- STUDY PLAN ENDPOINTS --- from datetime import date @app.get("/api/ai-coach/study-plan/today", tags=["AI Coach"]) def get_todays_study_plan(user_id: UUID, session: Session = Depends(get_session), current_user = Depends(require_tier([models.TierEnum.AI_COACH]))): today = datetime.utcnow().date() plan = session.exec( select(models.StudyPlan) .where(models.StudyPlan.user_id == user_id) .where(models.StudyPlan.date == today) ).first() if not plan: raise HTTPException(status_code=404, detail="No study plan generated for today yet.") return plan class ChatRequest(BaseModel): user_id: UUID message: str @app.post("/api/ai-coach/chat/stream", tags=["AI Coach"]) async def ai_tutor_chat_stream(request: ChatRequest, session: Session = Depends(get_session)): """ Streams the context-aware LLM response back to the client. """ user = session.get(models.User, request.user_id) if not user: raise HTTPException(status_code=404, detail="Student profile not found.") if user.tier != models.TierEnum.AI_COACH: raise HTTPException(status_code=403, detail="Access denied. Gated to AI_COACH tier.") # Enforce Daily Limit (50 messages) as per PRD A-04 # (In a full production build, you'd increment a Redis counter here) return StreamingResponse( stream_tutor_response(request.user_id, request.message, session), media_type="text/event-stream" ) @app.get("/api/ai-coach/readiness/{user_id}", tags=["AI Coach"]) def get_exam_readiness(user_id: UUID, session: Session = Depends(get_session), current_user = Depends(require_tier([models.TierEnum.AI_COACH]))): """ Returns the dynamic 0-100 Exam Readiness Score and College Prediction. """ readiness_data = calculate_readiness_score(user_id, session) if not readiness_data: raise HTTPException(status_code=404, detail="Could not calculate readiness score.") return readiness_data @app.get("/api/growth/referral/{user_id}", tags=["Viral Growth"]) def get_referral_dashboard(user_id: UUID, session: Session = Depends(get_session)): user = session.get(models.User, user_id) if not user: raise HTTPException(status_code=404, detail="User not found.") link = generate_referral_link(user_id) return { "referral_link": link, "total_referrals": 0, # In production, query a referral tracking table "rewards_earned_days": 0, "share_message": f"I'm on a {user.streak_count}-day study streak on PragatiVeer! Use my link to get 7 days of the AI Coach for free: {link}" } # --- USER CRUD & TEST LISTING --- @app.get("/api/users/{user_id}", response_model=models.User, tags=["Users"]) def get_user_profile(user_id: UUID, session: Session = Depends(get_session)): user = session.get(models.User, user_id) if not user: raise HTTPException(status_code=404, detail="User not found") return user class UserUpdate(BaseModel): name: Optional[str] = None target_exam: Optional[models.ExamEnum] = None student_class: Optional[models.ClassEnum] = None school: Optional[str] = None state: Optional[str] = None @app.put("/api/users/{user_id}", response_model=models.User, tags=["Users"]) def update_user_profile(user_id: UUID, payload: UserUpdate, session: Session = Depends(get_session)): user = session.get(models.User, user_id) if not user: raise HTTPException(status_code=404, detail="User not found") if payload.name is not None: user.name = payload.name if payload.target_exam is not None: user.target_exam = payload.target_exam if payload.student_class is not None: user.student_class = payload.student_class if payload.school is not None: user.school = payload.school if payload.state is not None: user.state = payload.state session.add(user) session.commit() session.refresh(user) return user from fastapi import UploadFile, File from storage import upload_file @app.post("/api/users/{user_id}/upload-avatar", tags=["Users"]) async def upload_user_avatar(user_id: UUID, file: UploadFile = File(...), session: Session = Depends(get_session)): user = session.get(models.User, user_id) if not user: raise HTTPException(status_code=404, detail="User not found") contents = await file.read() file_extension = file.filename.split(".")[-1] if "." in file.filename else "jpg" object_name = f"avatar_{user_id}.{file_extension}" public_url = upload_file(contents, object_name, content_type=file.content_type) if not public_url: raise HTTPException(status_code=500, detail="Failed to upload image to Minio.") return {"url": public_url} @app.get("/api/tests", response_model=List[models.Test], tags=["CBT Engine"]) def list_tests(exam_pattern: Optional[models.ExamEnum] = None, session: Session = Depends(get_session)): query = select(models.Test) if exam_pattern is not None: query = query.where(models.Test.exam_pattern == exam_pattern) return session.exec(query).all() @app.get("/api/users/{user_id}/attempts", tags=["Users"]) def get_user_attempts(user_id: UUID, session: Session = Depends(get_session)): query = select(models.Attempt).where(models.Attempt.user_id == user_id) return session.exec(query).all()