from fastapi import FastAPI, Depends, HTTPException, status, Request, UploadFile, File, Form, BackgroundTasks 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, date 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 predictor import calculate_readiness_score from growth import generate_referral_link, process_referral_reward @asynccontextmanager async def lifespan(app: FastAPI): init_db() yield app = FastAPI( title="PragatiVeer Core Engine", version="1.0.0", lifespan=lifespan ) 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]): 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 # --- AUTHENTICATION --- @app.post("/api/auth/otp-request", tags=["Authentication"]) def request_otp(phone_or_email: str): 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)): return {"status": "authenticated", "token": "mock_jwt_session_token_string"} # --- CBT ENGINE --- @app.get("/api/tests", response_model=List[models.Test], tags=["CBT Engine"]) def list_active_tests(exam_pattern: Optional[str] = None, type: Optional[str] = None, session: Session = Depends(get_session)): query = select(models.Test).where(models.Test.is_active == True) if exam_pattern: query = query.where(models.Test.exam_pattern == exam_pattern) if type: query = query.where(models.Test.type == type) else: query = query.where(models.Test.type != "DPS") return session.exec(query).all() @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)): 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_raw = payload.get("source_type", "Test") source_id = payload.get("source_id") client_answers = payload.get("answers", []) if not user_id or not source_id: raise HTTPException(status_code=400, detail="Missing required deployment metadata parameters.") # Normalize source_type string to match SourceTypeEnum values _st_map = {"dps": "DPS", "test": "Test", "microtest": "MicroTest"} source_type = _st_map.get(str(source_type_raw).lower(), "Test") total_questions = len(client_answers) correct_count = 0 processed_answers = [] attempt_id_str = payload.get("id") import uuid as _uuid attempt_id = _uuid.UUID(attempt_id_str) if attempt_id_str else _uuid.uuid4() attempt_record = models.Attempt( id=attempt_id, user_id=user_id, source_type=models.SourceTypeEnum(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=[], score=0, total_questions=total_questions, accuracy=0.0, status="Submitted" ) session.add(attempt_record) session.flush() for item in client_answers: q_id = item.get("question_id") selected = item.get("selected_answer") is_subjective = item.get("is_subjective", False) try: question = session.get(models.Question, q_id) except Exception: continue if not question: continue correct_ans = question.correct_answer # Support both {"value": "A"} and {"correct": "A"} key formats correct_value = correct_ans.get("value") or correct_ans.get("correct") if is_subjective: # Subjective: award credit for any non-trivial written answer student_text = str(selected or "").strip() is_correct = len(student_text) > 5 else: is_correct = bool(correct_value) and correct_value == selected if is_correct: correct_count += 1 processed_answers.append({ "question_id": str(q_id), "selected_answer": selected, "is_correct": is_correct, "is_subjective": is_subjective, "time_spent": item.get("time_spent", 0) }) # Track mistakes only for objective wrong answers if not is_correct and not is_subjective: try: existing_mistake = session.exec( select(models.Mistake).where( models.Mistake.user_id == user_id, models.Mistake.question_id == q_id ) ).first() except Exception: existing_mistake = None if existing_mistake: 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: new_mistake = models.Mistake( user_id=user_id, question_id=q_id, attempt_id=attempt_id, error_type=models.ErrorTypeEnum.CONCEPTUAL, topic=question.topic, subtopic=question.subtopic, next_review=datetime.utcnow() + timedelta(days=1), review_count=1, consecutive_correct=0, is_mastered=False ) session.add(new_mistake) accuracy = (correct_count / total_questions) * 100 if total_questions > 0 else 0.0 attempt_record.answers = processed_answers attempt_record.score = correct_count attempt_record.accuracy = accuracy session.add(attempt_record) session.commit() session.refresh(attempt_record) return attempt_record # --- AI COACH SPACED REPETITION ENDPOINTS --- @app.get("/api/mistakes/today", tags=["AI Coach"]) def get_todays_mistakes(user_id: UUID, session: Session = Depends(get_session)): 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) correct_value = question.correct_answer.get("value") or question.correct_answer.get("correct") is_correct = correct_value == selected_answer q_value = 4 if is_correct else 1 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": correct_value} # --- PARENT SUBSYSTEM --- @app.post("/api/parent/register", response_model=models.Parent, tags=["Parent Subsystem"]) def register_parent(parent_data: models.Parent, session: Session = Depends(get_session)): 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.") 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.") 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.") 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 --- @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.") pricing_map = {"Pro": 29900, "Al_Coach": 59900} 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)): payload_bytes = await request.body() payload_str = payload_bytes.decode("utf-8") signature = request.headers.get("X-Razorpay-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) 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: user.tier = models.TierEnum(tier_requested) session.add(user) 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 --- @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)): 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.") 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]))): 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, "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}" } # --- USERS --- @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 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/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() # --- ADMIN ROUTES --- @app.get("/api/admin/tests", tags=["Admin"]) def get_admin_tests(session: Session = Depends(get_session)): tests = session.exec(select(models.Test)).all() return tests class ManualQuestionInput(BaseModel): section: str text_en: str text_hi: Optional[str] = None options_en: List[str] options_hi: Optional[List[str]] = None correct_answer: str type: str = "SingleChoice" topic: str = "General" difficulty: str = "Medium" marks: int = 4 negative_marks: int = 1 class ManualTestCreate(BaseModel): title: str exam_pattern: str time_limit: int scheduled_date: Optional[datetime] = None end_date: Optional[datetime] = None admin_id: UUID questions_json: List[ManualQuestionInput] @app.post("/api/admin/tests/upload", tags=["Admin"]) async def upload_admin_test( file: UploadFile = File(...), title: str = Form(...), exam_pattern: str = Form(...), time_limit: int = Form(...), scheduled_date: Optional[datetime] = Form(None), admin_id: UUID = Form(...), session: Session = Depends(get_session) ): import storage import uuid from worker import process_pdf_test pdf_bytes = await file.read() object_name = f"tests/{uuid.uuid4()}_{file.filename}" url = storage.upload_file(pdf_bytes, object_name, content_type="application/pdf") if not url: raise HTTPException(status_code=500, detail="Failed to upload PDF to MinIO") test_obj = models.Test( title=title, type="Mock", exam_pattern=models.ExamEnum(exam_pattern), sections={}, total_time=time_limit, difficulty="Medium", scheduled_date=scheduled_date, is_active=False, created_by=admin_id ) session.add(test_obj) session.commit() session.refresh(test_obj) process_pdf_test.delay( test_id_str=str(test_obj.id), object_name=object_name, admin_id_str=str(admin_id), exam_pattern=exam_pattern ) return {"status": "processing", "test_id": test_obj.id, "message": "Test is being processed by AI"} @app.post("/api/admin/tests/create_manual", tags=["Admin"]) def create_manual_test( data: ManualTestCreate, session: Session = Depends(get_session) ): import uuid # 1. Create the test object test_obj = models.Test( title=data.title, type="Mock", exam_pattern=models.ExamEnum(data.exam_pattern), sections={}, total_time=data.time_limit, difficulty="Medium", scheduled_date=data.scheduled_date, end_date=data.end_date, is_active=True, created_by=data.admin_id ) session.add(test_obj) session.commit() session.refresh(test_obj) # 2. Parse and create questions sections_dict = {} for q_data in data.questions_json: q_obj = models.Question( text_en=q_data.text_en, text_hi=q_data.text_hi, options_en=q_data.options_en, options_hi=q_data.options_hi, correct_answer=q_data.correct_answer, type=q_data.type, topic=q_data.topic, difficulty=q_data.difficulty, marks=q_data.marks, negative_marks=q_data.negative_marks ) session.add(q_obj) session.commit() session.refresh(q_obj) sec_name = q_data.section if sec_name not in sections_dict: sections_dict[sec_name] = { "name": sec_name, "questions": [], "time_limit": None } sections_dict[sec_name]["questions"].append(str(q_obj.id)) # 3. Update the test with the sections test_obj.sections = sections_dict session.add(test_obj) session.commit() return {"message": "Test created successfully", "test_id": test_obj.id} @app.put("/api/admin/tests/{test_id}/toggle", tags=["Admin"]) def toggle_test_status(test_id: UUID, session: Session = Depends(get_session)): test_obj = session.get(models.Test, test_id) if not test_obj: raise HTTPException(status_code=404, detail="Test not found") test_obj.is_active = not test_obj.is_active session.add(test_obj) session.commit() return {"status": "success", "is_active": test_obj.is_active}