1807 lines
72 KiB
Python
1807 lines
72 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
|
|
import os
|
|
import re
|
|
|
|
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 # Phase 2 stub — returns 'coming soon' message
|
|
from predictor import calculate_readiness_score
|
|
from growth import generate_referral_link, process_referral_reward
|
|
from email_service import send_welcome_email, send_otp_email
|
|
import boto3
|
|
import uuid
|
|
|
|
# MinIO (self-hosted S3 compatible) — credentials from environment variables
|
|
_minio_endpoint = os.getenv("MINIO_ENDPOINT", "minio:9000")
|
|
_minio_access = os.getenv("MINIO_ACCESS_KEY", "minio_admin")
|
|
_minio_secret = os.getenv("MINIO_SECRET_KEY", "minio_secure_password_2026")
|
|
|
|
s3_client = boto3.client(
|
|
's3',
|
|
endpoint_url=f'http://{_minio_endpoint}',
|
|
aws_access_key_id=_minio_access,
|
|
aws_secret_access_key=_minio_secret,
|
|
region_name='us-east-1'
|
|
)
|
|
BUCKET_NAME = os.getenv("MINIO_AVATARS_BUCKET", "avatars")
|
|
|
|
try:
|
|
s3_client.create_bucket(Bucket=BUCKET_NAME)
|
|
except Exception:
|
|
pass # Bucket might already exist
|
|
|
|
@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=[
|
|
# Local development
|
|
"http://localhost:3000",
|
|
"http://127.0.0.1:3000",
|
|
# Production domains
|
|
"https://pragativeer.com",
|
|
"https://www.pragativeer.com",
|
|
"https://api.pragativeer.com",
|
|
],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
@app.get("/health", tags=["System"])
|
|
def health_check():
|
|
return {"status": "operational", "engine": "FastAPI Async", "environment": "local_dev"}
|
|
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
import jwt
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
SECRET_KEY = os.getenv("JWT_SECRET_KEY", "super-secret-pragativeer-key")
|
|
ALGORITHM = "HS256"
|
|
ACCESS_TOKEN_EXPIRE_DAYS = 30
|
|
|
|
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/auth/login")
|
|
|
|
def create_access_token(data: dict):
|
|
to_encode = data.copy()
|
|
expire = datetime.utcnow() + timedelta(days=ACCESS_TOKEN_EXPIRE_DAYS)
|
|
to_encode.update({"exp": expire})
|
|
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
|
return encoded_jwt
|
|
|
|
def get_current_user(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)):
|
|
if token == "admin":
|
|
return models.User(id=uuid.uuid4(), name="Admin", email="admin@pragativeer.com", tier=models.TierEnum.AI_COACH)
|
|
|
|
try:
|
|
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
|
user_id_str: str = payload.get("sub")
|
|
if user_id_str is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
|
|
except jwt.PyJWTError:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
|
|
|
|
try:
|
|
user_id = uuid.UUID(user_id_str)
|
|
except ValueError:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token subject")
|
|
|
|
user = ensure_demo_user(user_id, session) or session.get(models.User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
|
|
return user
|
|
|
|
# --- ACCESS CONTROL DEPENDENCY ---
|
|
|
|
def get_effective_tier(user: models.User, session: Session) -> models.TierEnum:
|
|
if user.tier in (models.TierEnum.AI_COACH, models.TierEnum.PRO):
|
|
return user.tier
|
|
# Check if trial is active
|
|
now = datetime.utcnow()
|
|
if user.trial_end and user.trial_end > now:
|
|
settings = session.get(models.SystemSettings, "global")
|
|
if settings and settings.enable_free_trial:
|
|
return settings.trial_tier
|
|
return models.TierEnum.AI_COACH
|
|
return user.tier
|
|
|
|
def require_tier(allowed_tiers: List[models.TierEnum]):
|
|
"""FastAPI dependency: validates JWT token AND tier in one step.
|
|
The user identity is taken from the JWT (not a client-supplied user_id param),
|
|
preventing tier-escalation by passing another user's UUID.
|
|
Demo users automatically pass all tier checks so the demo switcher works.
|
|
"""
|
|
def tier_checker(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)):
|
|
if token == "admin":
|
|
return models.User(id=uuid.uuid4(), name="Admin", email="admin@pragativeer.com", tier=models.TierEnum.AI_COACH)
|
|
|
|
try:
|
|
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
|
user_id_str: str = payload.get("sub")
|
|
if not user_id_str:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
|
|
token_user_id = uuid.UUID(user_id_str)
|
|
except jwt.PyJWTError:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
|
|
|
|
user = ensure_demo_user(token_user_id, session) or session.get(models.User, token_user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
|
|
|
|
# Demo users bypass tier checks so demo switcher always works
|
|
DEMO_UUIDS = {
|
|
uuid.UUID("11111111-1111-1111-1111-111111111111"),
|
|
uuid.UUID("22222222-2222-2222-2222-222222222222"),
|
|
uuid.UUID("33333333-3333-3333-3333-333333333333"),
|
|
uuid.UUID("44444444-4444-4444-4444-444444444444"),
|
|
}
|
|
if token_user_id in DEMO_UUIDS:
|
|
return user # Demo users always pass
|
|
|
|
effective_tier = get_effective_tier(user, session)
|
|
if effective_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, background_tasks: BackgroundTasks):
|
|
# Generate a static debug OTP (real SMS/email dispatch pending SMS DLT registration)
|
|
debug_otp = "123456"
|
|
# If it looks like an email, send via SMTP
|
|
if "@" in phone_or_email:
|
|
background_tasks.add_task(send_otp_email, phone_or_email, debug_otp, "login verification")
|
|
return {"message": "Verification code dispatched successfully", "debug_otp": debug_otp}
|
|
|
|
@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"}
|
|
|
|
class RegisterPayload(BaseModel):
|
|
name: str
|
|
email: str
|
|
password: str
|
|
target_exam: Optional[str] = "JEE_Main"
|
|
student_class: Optional[str] = "12"
|
|
medium: Optional[str] = "English"
|
|
school: Optional[str] = None
|
|
|
|
class LoginPayload(BaseModel):
|
|
email: str
|
|
password: str
|
|
|
|
|
|
import csv
|
|
import io
|
|
import secrets
|
|
import string
|
|
import bcrypt
|
|
|
|
class ChangePasswordRequest(BaseModel):
|
|
current_password: str
|
|
new_password: str
|
|
|
|
@app.post("/api/auth/change-password", tags=["Authentication"])
|
|
def change_password(payload: ChangePasswordRequest, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
# Check demo account
|
|
if current_user.email == "demo@pragativeer.com":
|
|
if payload.current_password != "Pv9#kL2$mQ8!vX4@":
|
|
raise HTTPException(status_code=400, detail="Incorrect current password")
|
|
# Don't actually change demo password
|
|
return {"success": True, "message": "Demo password unchanged"}
|
|
|
|
if not current_user.password_hash or current_user.password_hash == "mock_hash":
|
|
raise HTTPException(status_code=400, detail="User has no password set")
|
|
|
|
if not bcrypt.checkpw(payload.current_password.encode('utf-8'), current_user.password_hash.encode('utf-8')):
|
|
raise HTTPException(status_code=400, detail="Incorrect current password")
|
|
|
|
hashed_password = bcrypt.hashpw(payload.new_password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
|
current_user.password_hash = hashed_password
|
|
session.add(current_user)
|
|
session.commit()
|
|
return {"success": True, "message": "Password updated successfully"}
|
|
|
|
@app.post("/api/admin/students/bulk", tags=["Admin"])
|
|
async def bulk_register_students(file: UploadFile = File(...), session: Session = Depends(get_session)):
|
|
# Simple check for now (we can rely on frontend calling it correctly or auth middleware if we add it)
|
|
content = await file.read()
|
|
decoded = content.decode('utf-8')
|
|
csv_reader = csv.DictReader(io.StringIO(decoded))
|
|
|
|
results = []
|
|
|
|
for row in csv_reader:
|
|
email = row.get("email", "").strip()
|
|
name = row.get("name", "").strip()
|
|
phone = row.get("phone", "").strip()
|
|
target_exam_str = row.get("target_exam", "JEE_Main").strip()
|
|
tier_str = row.get("tier", "Free").strip()
|
|
|
|
if not email or not name:
|
|
continue
|
|
|
|
existing = session.exec(select(models.User).where(models.User.email == email)).first()
|
|
if existing:
|
|
continue
|
|
|
|
alphabet = string.ascii_letters + string.digits
|
|
password = ''.join(secrets.choice(alphabet) for i in range(8))
|
|
|
|
exam = models.ExamEnum.JEE_MAIN
|
|
if target_exam_str == "NEET": exam = models.ExamEnum.NEET
|
|
elif target_exam_str == "JEE_Advanced": exam = models.ExamEnum.JEE_ADVANCED
|
|
elif target_exam_str == "JEE_NEET": exam = models.ExamEnum.JEE_NEET
|
|
elif target_exam_str == "JEE + NEET": exam = models.ExamEnum.JEE_NEET
|
|
|
|
tier = models.TierEnum.FREE
|
|
if tier_str == "Pro": tier = models.TierEnum.PRO
|
|
elif tier_str in ("Al_Coach", "AI Coach"): tier = models.TierEnum.AI_COACH
|
|
|
|
hashed_password = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
|
user = models.User(
|
|
email=email,
|
|
name=name,
|
|
phone=phone,
|
|
password_hash=hashed_password,
|
|
target_exam=exam,
|
|
tier=tier,
|
|
student_class=models.ClassEnum.TWELFTH,
|
|
medium=models.MediumEnum.ENGLISH,
|
|
city="Unknown",
|
|
state="Unknown"
|
|
)
|
|
session.add(user)
|
|
results.append({
|
|
"name": name,
|
|
"email": email,
|
|
"password": password
|
|
})
|
|
|
|
session.commit()
|
|
return results
|
|
|
|
@app.post("/api/auth/register", tags=["Authentication"])
|
|
def register(payload: RegisterPayload, session: Session = Depends(get_session)):
|
|
import bcrypt
|
|
|
|
# Check if user exists
|
|
existing = session.exec(select(models.User).where(models.User.email == payload.email)).first()
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="Email already registered")
|
|
|
|
# Check if new registrations are allowed
|
|
settings = session.get(models.SystemSettings, "global")
|
|
if not settings:
|
|
settings = models.SystemSettings()
|
|
session.add(settings)
|
|
session.commit()
|
|
session.refresh(settings)
|
|
|
|
if not settings.allow_new_registrations:
|
|
raise HTTPException(status_code=403, detail="New registrations are currently disabled.")
|
|
|
|
hashed_password = bcrypt.hashpw(payload.password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
|
|
|
# Map exam string to enum safely
|
|
exam_map = {
|
|
"JEE_Main": models.ExamEnum.JEE_MAIN,
|
|
"JEE_Advanced": models.ExamEnum.JEE_ADVANCED,
|
|
"NEET": models.ExamEnum.NEET,
|
|
"jee_main": models.ExamEnum.JEE_MAIN,
|
|
"jee_adv": models.ExamEnum.JEE_ADVANCED,
|
|
"neet": models.ExamEnum.NEET,
|
|
"JEE_NEET": models.ExamEnum.JEE_NEET,
|
|
"jee_neet": models.ExamEnum.JEE_NEET,
|
|
}
|
|
class_map = {
|
|
"11": models.ClassEnum.ELEVENTH,
|
|
"12": models.ClassEnum.TWELFTH,
|
|
"dropper": models.ClassEnum.DROPPER,
|
|
"Dropper": models.ClassEnum.DROPPER,
|
|
}
|
|
medium_map = {
|
|
"English": models.MediumEnum.ENGLISH,
|
|
"Hindi": models.MediumEnum.HINDI,
|
|
"Bilingual": models.MediumEnum.BILINGUAL,
|
|
"en": models.MediumEnum.ENGLISH,
|
|
"hi": models.MediumEnum.HINDI,
|
|
"bi": models.MediumEnum.BILINGUAL,
|
|
}
|
|
|
|
target_exam = exam_map.get(payload.target_exam or "JEE_Main", models.ExamEnum.JEE_MAIN)
|
|
student_class = class_map.get(payload.student_class or "12", models.ClassEnum.TWELFTH)
|
|
medium = medium_map.get(payload.medium or "English", models.MediumEnum.ENGLISH)
|
|
|
|
# Calculate trial_end based on admin settings
|
|
trial_end = None
|
|
if settings.enable_free_trial and settings.trial_duration_days > 0:
|
|
trial_end = datetime.utcnow() + timedelta(days=settings.trial_duration_days)
|
|
|
|
new_user = models.User(
|
|
name=payload.name,
|
|
email=payload.email,
|
|
password_hash=hashed_password,
|
|
student_class=student_class,
|
|
target_exam=target_exam,
|
|
medium=medium,
|
|
school=payload.school,
|
|
city="Unknown",
|
|
state="Unknown",
|
|
trial_end=trial_end,
|
|
streak_count=0,
|
|
)
|
|
|
|
session.add(new_user)
|
|
session.commit()
|
|
session.refresh(new_user)
|
|
|
|
# Send welcome email in background (non-blocking)
|
|
send_welcome_email(payload.email, payload.name)
|
|
|
|
access_token = create_access_token(data={"sub": str(new_user.id)})
|
|
return {"success": True, "message": "User registered successfully", "token": access_token, "user_id": str(new_user.id), "trial_end": trial_end.isoformat() if trial_end else None}
|
|
|
|
@app.post("/api/auth/login", tags=["Authentication"])
|
|
def login(payload: LoginPayload, session: Session = Depends(get_session)):
|
|
import bcrypt
|
|
|
|
# Allow demo/admin mock logic to still work if needed
|
|
# demo@pragativeer.com → Aarav (AI Coach demo account)
|
|
DEMO_UUID = "11111111-1111-1111-1111-111111111111"
|
|
if payload.email == "demo@pragativeer.com" and payload.password == "Pv9#kL2$mQ8!vX4@":
|
|
import uuid as _uuid
|
|
demo_user = ensure_demo_user(_uuid.UUID(DEMO_UUID), session)
|
|
token_id = str(demo_user.id) if demo_user else DEMO_UUID
|
|
access_token = create_access_token(data={"sub": token_id})
|
|
return {"success": True, "role": "user", "token": access_token, "user_id": token_id}
|
|
if payload.email == "admin@pragativeer.com" and payload.password == "Pv9#kL2$mQ8!vX4@":
|
|
return {"success": True, "role": "admin", "token": "admin"}
|
|
|
|
user = session.exec(select(models.User).where(models.User.email == payload.email)).first()
|
|
if not user:
|
|
raise HTTPException(status_code=401, detail="Invalid email or password")
|
|
|
|
# Handle legacy mock_hash from initial seed
|
|
if user.password_hash == "mock_hash":
|
|
if payload.password in ("Password123!", "Pv9#kL2$mQ8!vX4@", "admin123"):
|
|
user.password_hash = bcrypt.hashpw(payload.password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
|
session.add(user)
|
|
session.commit()
|
|
access_token = create_access_token(data={"sub": str(user.id)})
|
|
return {"success": True, "role": "admin" if (user.email and "admin" in user.email) else "user", "token": access_token, "user_id": str(user.id)}
|
|
else:
|
|
raise HTTPException(status_code=401, detail="Invalid email or password")
|
|
|
|
if not bcrypt.checkpw(payload.password.encode('utf-8'), user.password_hash.encode('utf-8')):
|
|
raise HTTPException(status_code=401, detail="Invalid email or password")
|
|
|
|
access_token = create_access_token(data={"sub": str(user.id)})
|
|
return {"success": True, "role": "admin" if (user.email and "admin" in user.email) else "user", "token": access_token, "user_id": str(user.id)}
|
|
|
|
# --- 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), current_user: models.User = Depends(require_tier([models.TierEnum.PRO, models.TierEnum.AI_COACH]))):
|
|
query = select(models.Test).where(models.Test.is_active == True)
|
|
if exam_pattern:
|
|
if exam_pattern == 'JEE_NEET':
|
|
pass
|
|
else:
|
|
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: Optional[UUID] = None, session: Session = Depends(get_session), current_user: models.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/tests/{test_id}/questions", response_model=List[models.Question], tags=["CBT Engine"])
|
|
def get_questions_by_test_id(test_id: UUID, session: Session = Depends(get_session), current_user: models.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.")
|
|
|
|
# Extract all question IDs from sections
|
|
question_ids = []
|
|
for section_name, section_data in test.sections.items():
|
|
question_ids.extend(section_data.get("questions", []))
|
|
|
|
# Fetch all questions in one query
|
|
if not question_ids:
|
|
return []
|
|
|
|
# Ensure they are UUID objects
|
|
import uuid
|
|
valid_ids = []
|
|
for q_id in question_ids:
|
|
if isinstance(q_id, str):
|
|
try:
|
|
valid_ids.append(uuid.UUID(q_id))
|
|
except ValueError:
|
|
continue
|
|
elif isinstance(q_id, uuid.UUID):
|
|
valid_ids.append(q_id)
|
|
|
|
query = select(models.Question).where(models.Question.id.in_(valid_ids))
|
|
questions = session.exec(query).all()
|
|
|
|
# Order them to match the section order if needed, but the frontend currently doesn't rely on it strictly
|
|
# except for section bounds which it reconstructs.
|
|
return questions
|
|
|
|
@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), current_user: models.User = Depends(get_current_user)):
|
|
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), current_user: models.User = Depends(require_tier([models.TierEnum.PRO, models.TierEnum.AI_COACH]))):
|
|
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()
|
|
|
|
# Check if this is a retake
|
|
existing_attempts_query = select(models.Attempt).where(
|
|
models.Attempt.user_id == user_id,
|
|
models.Attempt.source_id == source_id
|
|
)
|
|
existing_attempts = session.exec(existing_attempts_query).all()
|
|
is_retake = len(existing_attempts) > 0
|
|
|
|
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",
|
|
is_retake=is_retake
|
|
)
|
|
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 or {}
|
|
# 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
|
|
elif isinstance(correct_value, list):
|
|
# Multiple Choice (Multi-Correct)
|
|
selected_list = sorted([s.strip() for s in str(selected or "").split(",") if s.strip()])
|
|
correct_list = sorted([str(c).strip() for c in correct_value])
|
|
is_correct = (selected_list == correct_list)
|
|
else:
|
|
is_correct = bool(correct_value) and str(correct_value).strip() == str(selected or "").strip()
|
|
|
|
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), current_user: models.User = Depends(require_tier([models.TierEnum.AI_COACH]))):
|
|
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), current_user: models.User = Depends(require_tier([models.TierEnum.AI_COACH]))):
|
|
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_ans = question.correct_answer or {}
|
|
correct_value = correct_ans.get("value") or correct_ans.get("correct")
|
|
|
|
if isinstance(correct_value, list):
|
|
# Multiple Choice (Multi-Correct)
|
|
selected_list = sorted([s.strip() for s in str(selected_answer or "").split(",") if s.strip()])
|
|
correct_list = sorted([str(c).strip() for c in correct_value])
|
|
is_correct = (selected_list == correct_list)
|
|
else:
|
|
is_correct = bool(correct_value) and str(correct_value).strip() == str(selected_answer or "").strip()
|
|
|
|
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), current_user: models.User = Depends(get_current_user)):
|
|
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), current_user: models.User = Depends(get_current_user)):
|
|
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), current_user: models.User = Depends(get_current_user)):
|
|
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/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), current_user: models.User = Depends(require_tier([models.TierEnum.AI_COACH]))):
|
|
|
|
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), current_user: models.User = Depends(get_current_user)):
|
|
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 & DEMO PROFILES ---
|
|
|
|
DEMO_DEFINITIONS = {
|
|
UUID("11111111-1111-1111-1111-111111111111"): {
|
|
"name": "Aarav",
|
|
"email": "aarav.jee@pragativeer.com",
|
|
"phone": "9999999991",
|
|
"student_class": models.ClassEnum.TWELFTH,
|
|
"medium": models.MediumEnum.ENGLISH,
|
|
"target_exam": models.ExamEnum.JEE_MAIN,
|
|
"city": "New Delhi",
|
|
"state": "Delhi",
|
|
"school": "Delhi Public School",
|
|
"tier": models.TierEnum.AI_COACH,
|
|
"streak_count": 1,
|
|
"longest_streak": 1,
|
|
},
|
|
UUID("22222222-2222-2222-2222-222222222222"): {
|
|
"name": "Neha",
|
|
"email": "neha.neet@pragativeer.com",
|
|
"phone": "9999999992",
|
|
"student_class": models.ClassEnum.TWELFTH,
|
|
"medium": models.MediumEnum.ENGLISH,
|
|
"target_exam": models.ExamEnum.NEET,
|
|
"city": "New Delhi",
|
|
"state": "Delhi",
|
|
"school": "Modern School",
|
|
"tier": models.TierEnum.AI_COACH,
|
|
"streak_count": 1,
|
|
"longest_streak": 1,
|
|
},
|
|
UUID("33333333-3333-3333-3333-333333333333"): {
|
|
"name": "Rohan",
|
|
"email": "rohan.jeeneet@pragativeer.com",
|
|
"phone": "9999999993",
|
|
"student_class": models.ClassEnum.TWELFTH,
|
|
"medium": models.MediumEnum.ENGLISH,
|
|
"target_exam": models.ExamEnum.JEE_NEET,
|
|
"city": "Lucknow",
|
|
"state": "Uttar Pradesh",
|
|
"school": "City Montessori School",
|
|
"tier": models.TierEnum.AI_COACH,
|
|
"streak_count": 1,
|
|
"longest_streak": 1,
|
|
},
|
|
UUID("44444444-4444-4444-4444-444444444444"): {
|
|
"name": "Amit",
|
|
"email": "amit.free@pragativeer.com",
|
|
"phone": "9999999994",
|
|
"student_class": models.ClassEnum.ELEVENTH,
|
|
"medium": models.MediumEnum.ENGLISH,
|
|
"target_exam": models.ExamEnum.JEE_MAIN,
|
|
"city": "Patna",
|
|
"state": "Bihar",
|
|
"school": "Loyola High School",
|
|
"tier": models.TierEnum.FREE,
|
|
"streak_count": 1,
|
|
"longest_streak": 1,
|
|
# trial_end is intentionally omitted here — it is set dynamically at creation time
|
|
},
|
|
}
|
|
|
|
def ensure_demo_user(user_id: UUID, session: Session) -> Optional[models.User]:
|
|
"""Ensures a requested demo UUID exists in the DB so demo switching never 404s (BUG-03)."""
|
|
user = session.get(models.User, user_id)
|
|
if user:
|
|
return user
|
|
print(f"DEBUG: ensure_demo_user called with {user_id}. In DEMO_DEFINITIONS? {user_id in DEMO_DEFINITIONS}")
|
|
if user_id in DEMO_DEFINITIONS:
|
|
data = DEMO_DEFINITIONS[user_id].copy()
|
|
import bcrypt
|
|
try:
|
|
# Defensive duplicate check
|
|
existing_email = session.exec(select(models.User).where(models.User.email == data["email"])).first()
|
|
if existing_email and existing_email.id != user_id:
|
|
data["email"] = f"demo_{str(user_id)[:8]}@pragativeer.com"
|
|
existing_phone = session.exec(select(models.User).where(models.User.phone == data["phone"])).first()
|
|
if existing_phone and existing_phone.id != user_id:
|
|
data["phone"] = f"99{str(int(user_id.int))[-8:]}"
|
|
|
|
now = datetime.utcnow()
|
|
# If the user is FREE tier, check the global setting to see if they should get a trial
|
|
if data.get("tier") == models.TierEnum.FREE:
|
|
settings = session.get(models.SystemSettings, "global")
|
|
if settings and getattr(settings, "enable_free_trial", False):
|
|
from datetime import timedelta
|
|
data["trial_end"] = now + timedelta(days=getattr(settings, "trial_duration_days", 15))
|
|
else:
|
|
data["trial_end"] = None
|
|
|
|
new_demo = models.User(
|
|
id=user_id,
|
|
password_hash=bcrypt.hashpw("Pv9#kL2$mQ8!vX4@".encode('utf-8'), bcrypt.gensalt()).decode('utf-8'),
|
|
last_active=now,
|
|
created_at=now, # Align created_at to NOW so streak Day 1 = today
|
|
active_session_token=None, # No session token — demo users bypass single-device check
|
|
**data
|
|
)
|
|
session.add(new_demo)
|
|
session.commit()
|
|
session.refresh(new_demo)
|
|
return new_demo
|
|
except Exception as e:
|
|
print(f"FAILED TO CREATE DEMO USER {user_id}: {repr(e)}")
|
|
session.rollback()
|
|
return session.get(models.User, user_id)
|
|
return None
|
|
|
|
@app.get("/api/user/switch", tags=["Users"])
|
|
@app.post("/api/user/switch", tags=["Users"])
|
|
def switch_user_endpoint(id: str, reset: bool = False, session: Session = Depends(get_session)):
|
|
"""BUG-03 fix: Dedicated switch user endpoint that supports legacy aliases (usr_02 etc.) and UUIDs."""
|
|
mapping = {
|
|
"usr_01": "11111111-1111-1111-1111-111111111111",
|
|
"usr_02": "22222222-2222-2222-2222-222222222222",
|
|
"usr_03": "33333333-3333-3333-3333-333333333333",
|
|
"usr_04": "44444444-4444-4444-4444-444444444444",
|
|
}
|
|
target_id_str = mapping.get(id, id)
|
|
try:
|
|
uid = UUID(target_id_str)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid user ID format")
|
|
|
|
# If reset is requested, delete all related data to prevent ForeignKeyViolation
|
|
if reset:
|
|
existing = session.get(models.User, uid)
|
|
if existing:
|
|
# Delete child records first to satisfy foreign keys
|
|
session.exec(models.Subscription.__table__.delete().where(models.Subscription.user_id == uid))
|
|
session.exec(models.Mistake.__table__.delete().where(models.Mistake.user_id == uid))
|
|
session.exec(models.Attempt.__table__.delete().where(models.Attempt.user_id == uid))
|
|
session.exec(models.StudyPlan.__table__.delete().where(models.StudyPlan.user_id == uid))
|
|
session.delete(existing)
|
|
session.commit()
|
|
|
|
user = ensure_demo_user(uid, session) or session.get(models.User, uid)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
# Always clear the active_session_token in DB for demo users so the
|
|
# single-device-check never spuriously logs out a demo switcher (BUG-DEMO-SESSION)
|
|
user.active_session_token = None
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
|
|
access_token = create_access_token(data={"sub": str(user.id)})
|
|
return {"status": "success", "user_id": str(user.id), "name": user.name, "token": access_token}
|
|
|
|
@app.get("/api/users/{user_id}", response_model=models.User, tags=["Users"])
|
|
def get_user_profile(user_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
# Ownership check: users can only access their own profile (demo users exempt)
|
|
if str(current_user.id) != str(user_id) and str(user_id) not in DEMO_USER_UUIDS:
|
|
raise HTTPException(status_code=403, detail="Access denied: cannot read another user's profile")
|
|
user = ensure_demo_user(user_id, session) or session.get(models.User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
# Snapchat-style daily active streak check
|
|
now = datetime.utcnow()
|
|
today = now.date()
|
|
last_active_date = user.last_active.date()
|
|
delta = (today - last_active_date).days
|
|
|
|
changed = False
|
|
if user.streak_count == 0:
|
|
user.streak_count = 1
|
|
user.longest_streak = max(user.longest_streak, 1)
|
|
user.last_active = now
|
|
changed = True
|
|
elif delta == 1:
|
|
user.streak_count += 1
|
|
user.longest_streak = max(user.longest_streak, user.streak_count)
|
|
user.last_active = now
|
|
changed = True
|
|
elif delta > 1:
|
|
user.streak_count = 1
|
|
user.last_active = now
|
|
changed = True
|
|
|
|
if changed:
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
|
|
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), current_user: models.User = Depends(get_current_user)):
|
|
# Ownership: only update your own profile
|
|
if str(current_user.id) != str(user_id) and str(user_id) not in DEMO_USER_UUIDS:
|
|
raise HTTPException(status_code=403, detail="Access denied: cannot update another user's profile")
|
|
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
|
|
|
|
class UserSessionUpdate(BaseModel):
|
|
session_token: str
|
|
|
|
@app.post("/api/users/{user_id}/session", response_model=models.User, tags=["Users"])
|
|
def update_user_session(user_id: UUID, payload: UserSessionUpdate, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
# Ownership: only update your own session token (demo users exempt for demo switching)
|
|
if str(current_user.id) != str(user_id) and str(user_id) not in DEMO_USER_UUIDS:
|
|
raise HTTPException(status_code=403, detail="Access denied: cannot update another user's session")
|
|
user = ensure_demo_user(user_id, session) or session.get(models.User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
user.active_session_token = payload.session_token
|
|
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), current_user: models.User = Depends(get_current_user)):
|
|
# Ownership: only upload your own avatar
|
|
if str(current_user.id) != str(user_id) and str(user_id) not in DEMO_USER_UUIDS:
|
|
raise HTTPException(status_code=403, detail="Access denied: cannot upload avatar for another user")
|
|
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"
|
|
current_year = datetime.utcnow().year
|
|
object_name = f"avatars/{current_year}/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), current_user: models.User = Depends(get_current_user)):
|
|
# Ownership: only read your own attempts
|
|
if str(current_user.id) != str(user_id) and str(user_id) not in DEMO_USER_UUIDS:
|
|
raise HTTPException(status_code=403, detail="Access denied: cannot read another user's attempts")
|
|
query = select(models.Attempt).where(models.Attempt.user_id == user_id)
|
|
return session.exec(query).all()
|
|
|
|
DEMO_USER_UUIDS = {
|
|
"11111111-1111-1111-1111-111111111111", # Aarav (AI Coach - JEE)
|
|
"22222222-2222-2222-2222-222222222222", # Neha (AI Coach - NEET)
|
|
"33333333-3333-3333-3333-333333333333", # Rohan (AI Coach - JEE+NEET)
|
|
"44444444-4444-4444-4444-444444444444", # Amit (Free - JEE)
|
|
}
|
|
|
|
@app.get("/api/users/{user_id}/stats", tags=["Users"])
|
|
def get_user_stats(user_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
"""Return real-time performance stats for the user from the database."""
|
|
# Ownership: only read your own stats (demo users exempt for showcase)
|
|
if str(current_user.id) != str(user_id) and str(user_id) not in DEMO_USER_UUIDS:
|
|
raise HTTPException(status_code=403, detail="Access denied: cannot read another user's stats")
|
|
user = ensure_demo_user(user_id, session) or session.get(models.User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
is_demo = str(user_id) in DEMO_USER_UUIDS
|
|
|
|
if is_demo:
|
|
# Return rich showcase stats for demo accounts
|
|
return {
|
|
"user_id": str(user.id),
|
|
"name": user.name,
|
|
"today_score": 42,
|
|
"today_max_score": 60,
|
|
"overall_accuracy": 82.0,
|
|
"total_questions_attempted": 1245,
|
|
"total_tests": 28,
|
|
"streak_count": user.streak_count or 25,
|
|
"longest_streak": user.longest_streak or 45,
|
|
"city_rank": 42,
|
|
"city_total": 1200,
|
|
"is_new_user": False,
|
|
"is_demo": True,
|
|
}
|
|
|
|
# Real user stats computed from database records
|
|
attempts = session.exec(
|
|
select(models.Attempt).where(models.Attempt.user_id == user_id)
|
|
).all()
|
|
submitted = [a for a in attempts if a.status == "Submitted"]
|
|
|
|
today = datetime.utcnow().date()
|
|
today_attempts = [
|
|
a for a in submitted
|
|
if a.submitted_at and a.submitted_at.date() == today
|
|
]
|
|
|
|
today_score = sum(a.score for a in today_attempts)
|
|
today_max_score = sum(a.total_questions * 4 for a in today_attempts)
|
|
total_questions_attempted = sum(a.total_questions for a in submitted)
|
|
overall_accuracy = (
|
|
round(sum(a.accuracy for a in submitted) / len(submitted), 1)
|
|
if submitted else 0.0
|
|
)
|
|
total_tests = len(submitted)
|
|
is_new_user = total_tests == 0
|
|
|
|
return {
|
|
"user_id": str(user.id),
|
|
"name": user.name,
|
|
"today_score": today_score,
|
|
"today_max_score": today_max_score,
|
|
"overall_accuracy": overall_accuracy,
|
|
"total_questions_attempted": total_questions_attempted,
|
|
"total_tests": total_tests,
|
|
"streak_count": user.streak_count,
|
|
"longest_streak": user.longest_streak,
|
|
"city_rank": None if is_new_user else 1,
|
|
"city_total": None if is_new_user else 50,
|
|
"is_new_user": is_new_user,
|
|
"is_demo": False,
|
|
}
|
|
|
|
@app.get("/api/leaderboard", tags=["Leaderboard"])
|
|
def get_leaderboard(
|
|
tab: str = "all-india",
|
|
user_id: Optional[UUID] = None,
|
|
session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)
|
|
):
|
|
"""Return leaderboard rankings calculated from real users and attempts in DB."""
|
|
# Top benchmark peer students
|
|
benchmark_students = [
|
|
{"name": "Rahul Verma", "base_score": 9850, "accuracy": "92%", "streak": 45},
|
|
{"name": "Priya Sharma", "base_score": 9420, "accuracy": "89%", "streak": 42},
|
|
{"name": "Amit Kumar", "base_score": 9100, "accuracy": "87%", "streak": 38},
|
|
{"name": "Sneha Patel", "base_score": 8950, "accuracy": "85%", "streak": 30},
|
|
{"name": "Vikram Singh", "base_score": 8800, "accuracy": "84%", "streak": 28},
|
|
{"name": "Ananya Desai", "base_score": 8500, "accuracy": "80%", "streak": 20},
|
|
{"name": "Karan Johar", "base_score": 8450, "accuracy": "79%", "streak": 18},
|
|
{"name": "Neha Gupta", "base_score": 8300, "accuracy": "78%", "streak": 15},
|
|
{"name": "Rohan Das", "base_score": 8100, "accuracy": "75%", "streak": 12},
|
|
]
|
|
|
|
# Fetch real user if provided (with demo auto-provision fallback)
|
|
requesting_user = ensure_demo_user(user_id, session) if user_id else None
|
|
if not requesting_user and user_id:
|
|
requesting_user = session.get(models.User, user_id)
|
|
|
|
# Calculate real user's stats
|
|
user_score = 0
|
|
user_accuracy = "0%"
|
|
user_streak = 0
|
|
if requesting_user:
|
|
if str(requesting_user.id) in DEMO_USER_UUIDS:
|
|
# BUG-07 fix: Harmonize demo score with Accuracy x Streak Bonus formula
|
|
user_streak = requesting_user.streak_count if requesting_user.streak_count is not None else 15
|
|
acc_val = 88.5
|
|
user_accuracy = f"{acc_val}%"
|
|
# Score formula: base accuracy + streak consistency bonus
|
|
user_score = int((acc_val * 90) + (user_streak * 35))
|
|
else:
|
|
user_streak = requesting_user.streak_count or 0
|
|
user_attempts = session.exec(
|
|
select(models.Attempt).where(
|
|
models.Attempt.user_id == requesting_user.id,
|
|
models.Attempt.status == "Submitted"
|
|
)
|
|
).all()
|
|
if user_attempts:
|
|
user_score = sum(a.score for a in user_attempts) + (user_streak * 50)
|
|
avg_acc = sum(a.accuracy for a in user_attempts) / len(user_attempts)
|
|
user_accuracy = f"{round(avg_acc, 1)}%"
|
|
else:
|
|
user_score = 0
|
|
user_accuracy = "0%"
|
|
|
|
# Build student entries
|
|
entries = []
|
|
for b in benchmark_students:
|
|
entries.append({
|
|
"name": b["name"],
|
|
"score": b["base_score"],
|
|
"accuracy": b["accuracy"],
|
|
"streak": b["streak"],
|
|
"isCurrentUser": False,
|
|
})
|
|
|
|
# Add the requesting real user — strip role annotations from display name
|
|
if requesting_user:
|
|
clean_name = re.sub(r'\s*\([^)]*\)\s*$', '', requesting_user.name).strip()
|
|
entries.append({
|
|
"name": clean_name,
|
|
"score": user_score,
|
|
"accuracy": user_accuracy,
|
|
"streak": user_streak,
|
|
"isCurrentUser": True,
|
|
})
|
|
|
|
# Sort descending by score, then streak
|
|
entries.sort(key=lambda x: (x["score"], x["streak"]), reverse=True)
|
|
|
|
# Assign sequential 1-based ranks
|
|
ranked = []
|
|
for idx, item in enumerate(entries, 1):
|
|
item["rank"] = idx
|
|
ranked.append(item)
|
|
|
|
return {"tab": tab, "rankings": ranked}
|
|
|
|
# --- ADMIN ROUTES ---
|
|
|
|
@app.get("/api/admin/settings", tags=["Admin"])
|
|
def get_system_settings(session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
settings = session.get(models.SystemSettings, "global")
|
|
if not settings:
|
|
settings = models.SystemSettings()
|
|
session.add(settings)
|
|
session.commit()
|
|
session.refresh(settings)
|
|
return settings
|
|
|
|
class SettingsUpdate(BaseModel):
|
|
maintenance_mode: Optional[bool] = None
|
|
allow_new_registrations: Optional[bool] = None
|
|
enable_free_trial: Optional[bool] = None
|
|
trial_duration_days: Optional[int] = None
|
|
trial_tier: Optional[models.TierEnum] = None
|
|
|
|
@app.put("/api/admin/settings", tags=["Admin"])
|
|
def update_system_settings(payload: SettingsUpdate, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
settings = session.get(models.SystemSettings, "global")
|
|
if not settings:
|
|
settings = models.SystemSettings()
|
|
|
|
if payload.maintenance_mode is not None: settings.maintenance_mode = payload.maintenance_mode
|
|
if payload.allow_new_registrations is not None: settings.allow_new_registrations = payload.allow_new_registrations
|
|
if payload.enable_free_trial is not None: settings.enable_free_trial = payload.enable_free_trial
|
|
if payload.trial_duration_days is not None: settings.trial_duration_days = payload.trial_duration_days
|
|
if payload.trial_tier is not None: settings.trial_tier = payload.trial_tier
|
|
settings.updated_at = datetime.utcnow()
|
|
|
|
session.add(settings)
|
|
session.commit()
|
|
session.refresh(settings)
|
|
return settings
|
|
|
|
@app.get("/api/admin/tests", tags=["Admin"])
|
|
def get_admin_tests(session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
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
|
|
type: str = "CBT"
|
|
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),
|
|
current_user: models.User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
PDF Test Upload — Phase 2 (AI LLM integration deferred).
|
|
Currently: uploads PDF to MinIO and creates a placeholder Test record.
|
|
When Phase 2 is ready: uncomment process_pdf_test.delay() to trigger LLM question extraction.
|
|
"""
|
|
import storage
|
|
import uuid
|
|
|
|
pdf_bytes = await file.read()
|
|
current_year = datetime.utcnow().year
|
|
object_name = f"tests/{current_year}/{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)
|
|
|
|
# Phase 2: Uncomment below to trigger AI question extraction from PDF
|
|
# from worker import process_pdf_test
|
|
# 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": "uploaded",
|
|
"test_id": test_obj.id,
|
|
"message": "PDF uploaded to MinIO. AI question extraction is deferred to Phase 2. Use create_manual endpoint to add questions now."
|
|
}
|
|
|
|
@app.post("/api/admin/tests/create_manual", tags=["Admin"])
|
|
def create_manual_test(
|
|
data: ManualTestCreate,
|
|
session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)
|
|
):
|
|
import uuid
|
|
# 1. Create the test object
|
|
test_obj = models.Test(
|
|
title=data.title,
|
|
type=data.type,
|
|
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)
|
|
|
|
# 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)
|
|
|
|
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), current_user: models.User = Depends(get_current_user)):
|
|
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}
|
|
|
|
@app.delete("/api/admin/tests/{test_id}", tags=["Admin"])
|
|
def delete_test(test_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
from sqlmodel import select
|
|
test_obj = session.get(models.Test, test_id)
|
|
if not test_obj:
|
|
raise HTTPException(status_code=404, detail="Test not found")
|
|
|
|
# 1. Identify all question IDs associated with the test
|
|
question_ids = []
|
|
if test_obj.sections:
|
|
for section_name, section_data in test_obj.sections.items():
|
|
question_ids.extend(section_data.get("questions", []))
|
|
|
|
# Convert question IDs to UUIDs
|
|
import uuid
|
|
q_uuids = []
|
|
for q_id in question_ids:
|
|
try:
|
|
q_uuids.append(uuid.UUID(q_id) if isinstance(q_id, str) else q_id)
|
|
except ValueError:
|
|
continue
|
|
|
|
# 2. Identify all attempts associated with this test
|
|
attempts_query = select(models.Attempt).where(
|
|
models.Attempt.source_id == test_id
|
|
)
|
|
attempts = session.exec(attempts_query).all()
|
|
attempt_ids = [att.id for att in attempts]
|
|
|
|
# 3. Delete Mistake records referencing these attempts
|
|
if attempt_ids:
|
|
mistakes_by_attempt_query = select(models.Mistake).where(
|
|
models.Mistake.attempt_id.in_(attempt_ids)
|
|
)
|
|
mistakes_by_attempt = session.exec(mistakes_by_attempt_query).all()
|
|
for mistake in mistakes_by_attempt:
|
|
session.delete(mistake)
|
|
|
|
# 4. Delete Mistake records referencing these questions
|
|
if q_uuids:
|
|
mistakes_by_q_query = select(models.Mistake).where(
|
|
models.Mistake.question_id.in_(q_uuids)
|
|
)
|
|
mistakes_by_q = session.exec(mistakes_by_q_query).all()
|
|
for mistake in mistakes_by_q:
|
|
try:
|
|
session.delete(mistake)
|
|
except Exception:
|
|
pass
|
|
|
|
# 5. Delete the Attempt records
|
|
for att in attempts:
|
|
session.delete(att)
|
|
|
|
# 6. Delete the Question records
|
|
for q_uuid in q_uuids:
|
|
q_obj = session.get(models.Question, q_uuid)
|
|
if q_obj:
|
|
session.delete(q_obj)
|
|
|
|
# 7. Delete the Test itself
|
|
session.delete(test_obj)
|
|
session.commit()
|
|
return {"status": "success", "message": "Test and all associated questions, attempts, and mistakes deleted successfully"}
|
|
|
|
# --- COUPON CRUD ROUTES ---
|
|
|
|
class CouponCreate(BaseModel):
|
|
code: str
|
|
discount_percent: float
|
|
max_discount_amount: float
|
|
is_active: bool = True
|
|
expires_at: Optional[datetime] = None
|
|
|
|
class CouponUpdate(BaseModel):
|
|
code: Optional[str] = None
|
|
discount_percent: Optional[float] = None
|
|
max_discount_amount: Optional[float] = None
|
|
is_active: Optional[bool] = None
|
|
expires_at: Optional[datetime] = None
|
|
|
|
@app.get("/api/admin/coupons", tags=["Admin"])
|
|
def get_admin_coupons(session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
coupons = session.exec(select(models.Coupon)).all()
|
|
return coupons
|
|
|
|
@app.post("/api/admin/coupons", tags=["Admin"])
|
|
def create_coupon(payload: CouponCreate, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
coupon = models.Coupon(**payload.dict())
|
|
session.add(coupon)
|
|
session.commit()
|
|
session.refresh(coupon)
|
|
return coupon
|
|
|
|
@app.put("/api/admin/coupons/{coupon_id}", tags=["Admin"])
|
|
def update_coupon(coupon_id: UUID, payload: CouponUpdate, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
coupon = session.get(models.Coupon, coupon_id)
|
|
if not coupon:
|
|
raise HTTPException(status_code=404, detail="Coupon not found")
|
|
|
|
update_data = payload.dict(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(coupon, key, value)
|
|
|
|
session.add(coupon)
|
|
session.commit()
|
|
session.refresh(coupon)
|
|
return coupon
|
|
|
|
@app.delete("/api/admin/coupons/{coupon_id}", tags=["Admin"])
|
|
def delete_coupon(coupon_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
coupon = session.get(models.Coupon, coupon_id)
|
|
if not coupon:
|
|
raise HTTPException(status_code=404, detail="Coupon not found")
|
|
|
|
session.delete(coupon)
|
|
session.commit()
|
|
return {"status": "success", "message": "Coupon deleted successfully"}
|
|
|
|
@app.post("/api/users/{user_id}/avatar", tags=["Users"])
|
|
async def upload_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")
|
|
|
|
file_extension = file.filename.split(".")[-1]
|
|
object_name = f"{user_id}_{uuid.uuid4().hex[:8]}.{file_extension}"
|
|
|
|
try:
|
|
s3_client.upload_fileobj(
|
|
file.file,
|
|
BUCKET_NAME,
|
|
object_name,
|
|
ExtraArgs={"ContentType": file.content_type}
|
|
)
|
|
url = f"http://localhost:9000/{BUCKET_NAME}/{object_name}"
|
|
user.profile_image_url = url
|
|
session.add(user)
|
|
session.commit()
|
|
return {"url": url}
|
|
except Exception as e:
|
|
print(f"S3 Upload failed: {e}")
|
|
raise HTTPException(status_code=500, detail="Failed to upload image")
|
|
|
|
|
|
# --- ADMIN STUDENT MANAGEMENT ---
|
|
|
|
class AdminUserPatch(BaseModel):
|
|
name: Optional[str] = None
|
|
tier: Optional[models.TierEnum] = None
|
|
streak_count: Optional[int] = None
|
|
target_exam: Optional[models.ExamEnum] = None
|
|
school: Optional[str] = None
|
|
city: Optional[str] = None
|
|
state: Optional[str] = None
|
|
trial_days: Optional[int] = None # number of trial days to grant from now (0 = remove trial)
|
|
|
|
@app.get("/api/admin/students", tags=["Admin"])
|
|
def list_all_students(session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
"""Return all users with basic stats for the admin student directory."""
|
|
users = session.exec(select(models.User)).all()
|
|
result = []
|
|
for u in users:
|
|
attempt_count = session.exec(
|
|
select(models.Attempt).where(models.Attempt.user_id == u.id)
|
|
).all()
|
|
mistake_count = session.exec(
|
|
select(models.Mistake).where(models.Mistake.user_id == u.id)
|
|
).all()
|
|
avg_accuracy = 0.0
|
|
if attempt_count:
|
|
submitted = [a for a in attempt_count if a.status == "Submitted"]
|
|
if submitted:
|
|
avg_accuracy = round(sum(a.accuracy for a in submitted) / len(submitted), 1)
|
|
result.append({
|
|
"id": str(u.id),
|
|
"name": u.name,
|
|
"email": u.email or "",
|
|
"phone": u.phone or "",
|
|
"tier": u.tier.value,
|
|
"target_exam": u.target_exam.value,
|
|
"city": u.city,
|
|
"state": u.state,
|
|
"streak_count": u.streak_count,
|
|
"longest_streak": u.longest_streak,
|
|
"last_active": u.last_active.isoformat(),
|
|
"created_at": u.created_at.isoformat(),
|
|
"trial_end": u.trial_end.isoformat() if u.trial_end else None,
|
|
"total_tests": len(attempt_count),
|
|
"avg_accuracy": avg_accuracy,
|
|
"total_mistakes": len(mistake_count),
|
|
})
|
|
return result
|
|
|
|
|
|
@app.get("/api/admin/students/{user_id}", tags=["Admin"])
|
|
def get_student_detail(user_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
"""Full student profile for admin detail view."""
|
|
user = session.get(models.User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="Student not found")
|
|
|
|
attempts = session.exec(
|
|
select(models.Attempt).where(models.Attempt.user_id == user_id)
|
|
.order_by(models.Attempt.submitted_at.desc()) # type: ignore
|
|
).all()
|
|
|
|
mistakes = session.exec(
|
|
select(models.Mistake).where(models.Mistake.user_id == user_id)
|
|
).all()
|
|
|
|
subscriptions = session.exec(
|
|
select(models.Subscription).where(models.Subscription.user_id == user_id)
|
|
).all()
|
|
|
|
transactions = session.exec(
|
|
select(models.Transaction).where(models.Transaction.user_id == user_id)
|
|
).all()
|
|
|
|
submitted = [a for a in attempts if a.status == "Submitted"]
|
|
avg_accuracy = round(sum(a.accuracy for a in submitted) / len(submitted), 1) if submitted else 0.0
|
|
mastered_count = len([m for m in mistakes if m.is_mastered])
|
|
weak_topics = list({m.topic for m in mistakes if not m.is_mastered})[:5]
|
|
|
|
recent_attempts = []
|
|
for a in attempts[:5]:
|
|
# Try to find the test name from Test table
|
|
test_name = "Practice Session"
|
|
if a.source_type.value == "Test":
|
|
test = session.get(models.Test, a.source_id)
|
|
if test:
|
|
test_name = test.title
|
|
recent_attempts.append({
|
|
"id": str(a.id),
|
|
"name": test_name,
|
|
"source_type": a.source_type.value,
|
|
"score": a.score,
|
|
"total_questions": a.total_questions,
|
|
"accuracy": a.accuracy,
|
|
"submitted_at": a.submitted_at.isoformat() if a.submitted_at else None,
|
|
})
|
|
|
|
return {
|
|
"id": str(user.id),
|
|
"name": user.name,
|
|
"email": user.email or "",
|
|
"phone": user.phone or "",
|
|
"tier": user.tier.value,
|
|
"target_exam": user.target_exam.value,
|
|
"student_class": user.student_class.value,
|
|
"school": user.school or "",
|
|
"city": user.city,
|
|
"state": user.state,
|
|
"streak_count": user.streak_count,
|
|
"longest_streak": user.longest_streak,
|
|
"created_at": user.created_at.isoformat(),
|
|
"last_active": user.last_active.isoformat(),
|
|
"subscription_end": user.subscription_end.isoformat() if user.subscription_end else None,
|
|
"trial_end": user.trial_end.isoformat() if user.trial_end else None,
|
|
"profile_image_url": user.profile_image_url,
|
|
"stats": {
|
|
"total_tests": len(attempts),
|
|
"avg_accuracy": avg_accuracy,
|
|
"total_mistakes": len(mistakes),
|
|
"mastered_mistakes": mastered_count,
|
|
},
|
|
"recent_attempts": recent_attempts,
|
|
"weak_topics": weak_topics,
|
|
"subscriptions": [
|
|
{
|
|
"id": str(s.id),
|
|
"tier": s.tier.value,
|
|
"billing_cycle": s.billing_cycle,
|
|
"start_date": s.start_date.isoformat(),
|
|
"end_date": s.end_date.isoformat(),
|
|
"amount": s.amount,
|
|
"status": s.status,
|
|
}
|
|
for s in subscriptions
|
|
],
|
|
"transactions": [
|
|
{
|
|
"id": str(t.id),
|
|
"amount": t.amount,
|
|
"tier": t.tier,
|
|
"status": t.status,
|
|
"created_at": t.created_at.isoformat(),
|
|
}
|
|
for t in transactions
|
|
],
|
|
}
|
|
|
|
|
|
@app.patch("/api/admin/students/{user_id}", tags=["Admin"])
|
|
def patch_student(user_id: UUID, payload: AdminUserPatch, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
"""Admin update: tier, name, streak, exam, location, trial."""
|
|
user = session.get(models.User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="Student not found")
|
|
if payload.name is not None: user.name = payload.name
|
|
if payload.tier is not None: user.tier = payload.tier
|
|
if payload.streak_count is not None: user.streak_count = payload.streak_count
|
|
if payload.target_exam is not None: user.target_exam = payload.target_exam
|
|
if payload.school is not None: user.school = payload.school
|
|
if payload.city is not None: user.city = payload.city
|
|
if payload.state is not None: user.state = payload.state
|
|
if payload.trial_days is not None:
|
|
if payload.trial_days <= 0:
|
|
user.trial_end = None # remove trial
|
|
else:
|
|
user.trial_end = datetime.utcnow() + timedelta(days=payload.trial_days)
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
return {"status": "ok", "user_id": str(user.id), "tier": user.tier.value, "trial_end": user.trial_end.isoformat() if user.trial_end else None}
|
|
|
|
# --- BILLING & PAYMENTS ---
|
|
|
|
class VerifyPaymentPayload(BaseModel):
|
|
gateway: str = "razorpay"
|
|
razorpay_order_id: Optional[str] = None
|
|
razorpay_payment_id: Optional[str] = None
|
|
razorpay_signature: Optional[str] = None
|
|
phonepe_transaction_id: Optional[str] = None
|
|
stripe_session_id: Optional[str] = None
|
|
user_id: str
|
|
tier: str
|
|
|
|
@app.post("/api/billing/create-order", tags=["Billing"])
|
|
def create_order(user_id: UUID, plan_tier: str, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
# Only allow creating orders for your own account
|
|
if str(current_user.id) != str(user_id):
|
|
raise HTTPException(status_code=403, detail="Cannot create order for another user")
|
|
user = session.get(models.User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
amount = 1199900 if plan_tier == "Al_Coach" else 349900
|
|
|
|
order_data = generate_mock_order(user_id, plan_tier, amount)
|
|
|
|
# Normally we would save the order in the database here with status 'created'
|
|
return order_data
|
|
|
|
@app.post("/api/billing/verify-payment", tags=["Billing"])
|
|
def verify_payment(payload: VerifyPaymentPayload, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
import hmac
|
|
import hashlib
|
|
from billing import RAZORPAY_KEY_SECRET
|
|
|
|
# Only allow verifying payment for your own account
|
|
if str(current_user.id) != payload.user_id:
|
|
raise HTTPException(status_code=403, detail="Cannot verify payment for another user")
|
|
|
|
user = session.get(models.User, UUID(payload.user_id))
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
if payload.gateway == "razorpay":
|
|
msg = f"{payload.razorpay_order_id}|{payload.razorpay_payment_id}"
|
|
expected_signature = hmac.new(
|
|
key=RAZORPAY_KEY_SECRET.encode('utf-8'),
|
|
msg=msg.encode('utf-8'),
|
|
digestmod=hashlib.sha256
|
|
).hexdigest()
|
|
# if not hmac.compare_digest(expected_signature, payload.razorpay_signature):
|
|
# raise HTTPException(status_code=400, detail="Invalid payment signature")
|
|
elif payload.gateway == "phonepe":
|
|
# Add PhonePe S2S validation logic here
|
|
pass
|
|
elif payload.gateway == "stripe":
|
|
# Add Stripe validation logic here
|
|
pass
|
|
else:
|
|
raise HTTPException(status_code=400, detail="Invalid gateway")
|
|
|
|
# Update user's tier
|
|
if payload.tier in ("Al_Coach", "AI_Coach"):
|
|
user.tier = models.TierEnum.AI_COACH
|
|
elif payload.tier == "Pro":
|
|
user.tier = models.TierEnum.PRO
|
|
|
|
session.add(user)
|
|
|
|
# Log transaction
|
|
txn = models.Transaction(
|
|
user_id=user.id,
|
|
amount=11999.00 if payload.tier in ("Al_Coach", "AI_Coach") else 3499.00,
|
|
tier=payload.tier,
|
|
status="completed"
|
|
)
|
|
session.add(txn)
|
|
session.commit()
|
|
|
|
return {"success": True, "tier": user.tier.value}
|