1206 lines
44 KiB
Python
1206 lines
44 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
|
|
import boto3
|
|
import uuid
|
|
|
|
s3_client = boto3.client(
|
|
's3',
|
|
endpoint_url='http://localhost:9000',
|
|
aws_access_key_id='minioadmin',
|
|
aws_secret_access_key='minioadmin',
|
|
region_name='us-east-1'
|
|
)
|
|
BUCKET_NAME = "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=["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"}
|
|
|
|
class RegisterPayload(BaseModel):
|
|
name: str
|
|
email: str
|
|
password: str
|
|
|
|
class LoginPayload(BaseModel):
|
|
email: str
|
|
password: str
|
|
|
|
@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")
|
|
|
|
hashed_password = bcrypt.hashpw(payload.password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
|
|
|
# Creating a dummy user with default enums since the payload is simplified
|
|
new_user = models.User(
|
|
name=payload.name,
|
|
email=payload.email,
|
|
password_hash=hashed_password,
|
|
student_class=models.ClassEnum.TWELFTH,
|
|
target_exam=models.ExamEnum.JEE_MAIN,
|
|
medium=models.MediumEnum.ENGLISH,
|
|
city="Unknown",
|
|
state="Unknown"
|
|
)
|
|
|
|
session.add(new_user)
|
|
session.commit()
|
|
session.refresh(new_user)
|
|
|
|
return {"success": True, "message": "User registered successfully"}
|
|
|
|
@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
|
|
if payload.email == "demo@pragativeer.com" and payload.password == "Pv9#kL2$mQ8!vX4@":
|
|
demo_user = session.exec(select(models.User).where(models.User.email == "student@pragativeer.com")).first()
|
|
token_id = str(demo_user.id) if demo_user else "b0e17378-864d-439e-8806-708dc48a891b"
|
|
return {"success": True, "role": "user", "token": token_id}
|
|
if payload.email == "admin@pragativeer.com" and payload.password == "admin123":
|
|
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()
|
|
return {"success": True, "role": "admin" if (user.email and "admin" in user.email) else "user", "token": 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")
|
|
|
|
# Real application would issue a JWT here
|
|
return {"success": True, "role": "admin" if (user.email and "admin" in user.email) else "user", "token": 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)):
|
|
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: Optional[UUID] = None, 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/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)):
|
|
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)):
|
|
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()
|
|
|
|
# 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)):
|
|
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_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)):
|
|
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/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")
|
|
|
|
# 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)):
|
|
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)):
|
|
user = 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)):
|
|
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)):
|
|
query = select(models.Attempt).where(models.Attempt.user_id == user_id)
|
|
return session.exec(query).all()
|
|
|
|
# --- ADMIN ROUTES ---
|
|
|
|
@app.get("/api/admin/settings", tags=["Admin"])
|
|
def get_system_settings(session: Session = Depends(get_session)):
|
|
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)):
|
|
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)):
|
|
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)
|
|
):
|
|
import storage
|
|
import uuid
|
|
from worker import process_pdf_test
|
|
|
|
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)
|
|
|
|
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=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)):
|
|
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)):
|
|
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)):
|
|
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)):
|
|
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)):
|
|
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)):
|
|
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
|
|
|
|
@app.get("/api/admin/students", tags=["Admin"])
|
|
def list_all_students(session: Session = Depends(get_session)):
|
|
"""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(),
|
|
"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)):
|
|
"""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,
|
|
"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)):
|
|
"""Admin update: tier, name, streak, exam, location."""
|
|
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
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
return {"status": "ok", "user_id": str(user.id), "tier": user.tier.value}
|
|
|
|
# --- 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)):
|
|
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)):
|
|
import hmac
|
|
import hashlib
|
|
from billing import RAZORPAY_KEY_SECRET
|
|
|
|
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}
|