619 lines
23 KiB
Python
619 lines
23 KiB
Python
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 = "General"
|
|
subject: str
|
|
topic: str
|
|
subtopic: str
|
|
difficulty: str
|
|
type: str
|
|
question_text: str
|
|
options: Optional[Dict[str, Any]] = None
|
|
correct_answer: Dict[str, Any]
|
|
solution_text: str
|
|
language: str
|
|
|
|
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(
|
|
subject=q_data.subject,
|
|
topic=q_data.topic,
|
|
subtopic=q_data.subtopic,
|
|
difficulty=q_data.difficulty,
|
|
type=models.QuestionTypeEnum(q_data.type),
|
|
question_text=q_data.question_text,
|
|
options=q_data.options,
|
|
correct_answer=q_data.correct_answer,
|
|
solution_text=q_data.solution_text,
|
|
language=models.MediumEnum(q_data.language),
|
|
created_by=data.admin_id
|
|
)
|
|
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} |