|
|
|
@@ -76,7 +76,16 @@ from fastapi.security import OAuth2PasswordBearer
|
|
|
|
import jwt
|
|
|
|
import jwt
|
|
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
|
|
|
|
|
|
|
|
SECRET_KEY = os.getenv("JWT_SECRET_KEY", "super-secret-pragativeer-key")
|
|
|
|
SECRET_KEY = os.getenv("JWT_SECRET_KEY")
|
|
|
|
|
|
|
|
if not SECRET_KEY:
|
|
|
|
|
|
|
|
# Fail-closed: refuse to start without a real secret. A known default
|
|
|
|
|
|
|
|
# secret lets anyone forge JWTs for any user id.
|
|
|
|
|
|
|
|
raise RuntimeError(
|
|
|
|
|
|
|
|
"JWT_SECRET_KEY not set. Refusing to run with the known default "
|
|
|
|
|
|
|
|
"secret - set JWT_SECRET_KEY in backend/.env and restart."
|
|
|
|
|
|
|
|
)
|
|
|
|
|
|
|
|
DEMO_PASSWORD = os.getenv("DEMO_PASSWORD", "Pv9#kL2$mQ8!vX4@")
|
|
|
|
|
|
|
|
ADMIN_TOKEN = os.getenv("ADMIN_TOKEN", "admin")
|
|
|
|
ALGORITHM = "HS256"
|
|
|
|
ALGORITHM = "HS256"
|
|
|
|
ACCESS_TOKEN_EXPIRE_DAYS = 30
|
|
|
|
ACCESS_TOKEN_EXPIRE_DAYS = 30
|
|
|
|
|
|
|
|
|
|
|
|
@@ -90,7 +99,7 @@ def create_access_token(data: dict):
|
|
|
|
return encoded_jwt
|
|
|
|
return encoded_jwt
|
|
|
|
|
|
|
|
|
|
|
|
def get_current_user(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)):
|
|
|
|
def get_current_user(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)):
|
|
|
|
if token == "admin":
|
|
|
|
if token == ADMIN_TOKEN:
|
|
|
|
return models.User(id=uuid.UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"), name="Admin", email="admin@pragativeer.com", tier=models.TierEnum.AI_COACH)
|
|
|
|
return models.User(id=uuid.UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"), name="Admin", email="admin@pragativeer.com", tier=models.TierEnum.AI_COACH)
|
|
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
try:
|
|
|
|
@@ -111,6 +120,27 @@ def get_current_user(token: str = Depends(oauth2_scheme), session: Session = Dep
|
|
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
|
|
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
|
|
|
|
return user
|
|
|
|
return user
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
ADMIN_FALLBACK_ID = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def require_admin(
|
|
|
|
|
|
|
|
current_user: models.User = Depends(get_current_user),
|
|
|
|
|
|
|
|
) -> models.User:
|
|
|
|
|
|
|
|
"""Dependency for every /api/admin/* route.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Accepts exactly two identities:
|
|
|
|
|
|
|
|
1. The DB admin user (email contains 'admin') with a valid JWT, or
|
|
|
|
|
|
|
|
2. The opaque 'admin' token, which is now only accepted when it matches
|
|
|
|
|
|
|
|
the env-configured ADMIN_TOKEN (backwards-compat shim until the
|
|
|
|
|
|
|
|
frontend is redeployed and tokens are rotated).
|
|
|
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
if "admin" in (current_user.email or "").lower():
|
|
|
|
|
|
|
|
return current_user
|
|
|
|
|
|
|
|
if current_user.id == uuid.UUID(ADMIN_FALLBACK_ID):
|
|
|
|
|
|
|
|
return current_user
|
|
|
|
|
|
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# --- ACCESS CONTROL DEPENDENCY ---
|
|
|
|
# --- ACCESS CONTROL DEPENDENCY ---
|
|
|
|
|
|
|
|
|
|
|
|
def get_effective_tier(user: models.User, session: Session) -> models.TierEnum:
|
|
|
|
def get_effective_tier(user: models.User, session: Session) -> models.TierEnum:
|
|
|
|
@@ -132,8 +162,8 @@ def require_tier(allowed_tiers: List[models.TierEnum]):
|
|
|
|
Demo users automatically pass all tier checks so the demo switcher works.
|
|
|
|
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)):
|
|
|
|
def tier_checker(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)):
|
|
|
|
if token == "admin":
|
|
|
|
if token == ADMIN_TOKEN:
|
|
|
|
return models.User(id=uuid.UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"), name="Admin", email="admin@pragativeer.com", tier=models.TierEnum.AI_COACH)
|
|
|
|
return models.User(id=uuid.UUID(ADMIN_FALLBACK_ID), name="Admin", email="admin@pragativeer.com", tier=models.TierEnum.AI_COACH)
|
|
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
try:
|
|
|
|
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
|
|
|
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
|
|
|
@@ -208,9 +238,9 @@ class ChangePasswordRequest(BaseModel):
|
|
|
|
|
|
|
|
|
|
|
|
@app.post("/api/auth/change-password", tags=["Authentication"])
|
|
|
|
@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)):
|
|
|
|
def change_password(payload: ChangePasswordRequest, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
|
|
# Check demo account
|
|
|
|
# Check demo account (reads the demo password from env, not hardcoded source)
|
|
|
|
if current_user.email == "demo@pragativeer.com":
|
|
|
|
if current_user.email == "demo@pragativeer.com":
|
|
|
|
if payload.current_password != "Pv9#kL2$mQ8!vX4@":
|
|
|
|
if payload.current_password != DEMO_PASSWORD:
|
|
|
|
raise HTTPException(status_code=400, detail="Incorrect current password")
|
|
|
|
raise HTTPException(status_code=400, detail="Incorrect current password")
|
|
|
|
# Don't actually change demo password
|
|
|
|
# Don't actually change demo password
|
|
|
|
return {"success": True, "message": "Demo password unchanged"}
|
|
|
|
return {"success": True, "message": "Demo password unchanged"}
|
|
|
|
@@ -230,7 +260,7 @@ def change_password(payload: ChangePasswordRequest, session: Session = Depends(g
|
|
|
|
return {"success": True, "message": "Password updated successfully"}
|
|
|
|
return {"success": True, "message": "Password updated successfully"}
|
|
|
|
|
|
|
|
|
|
|
|
@app.post("/api/admin/students/bulk", tags=["Admin"])
|
|
|
|
@app.post("/api/admin/students/bulk", tags=["Admin"])
|
|
|
|
async def bulk_register_students(file: UploadFile = File(...), session: Session = Depends(get_session)):
|
|
|
|
async def bulk_register_students(file: UploadFile = File(...), session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
|
|
|
|
# Simple check for now (we can rely on frontend calling it correctly or auth middleware if we add it)
|
|
|
|
# Simple check for now (we can rely on frontend calling it correctly or auth middleware if we add it)
|
|
|
|
content = await file.read()
|
|
|
|
content = await file.read()
|
|
|
|
decoded = content.decode('utf-8-sig')
|
|
|
|
decoded = content.decode('utf-8-sig')
|
|
|
|
@@ -297,7 +327,7 @@ class DeleteUsersRequest(BaseModel):
|
|
|
|
user_ids: List[UUID]
|
|
|
|
user_ids: List[UUID]
|
|
|
|
|
|
|
|
|
|
|
|
@app.delete("/api/admin/students/bulk", tags=["Admin"])
|
|
|
|
@app.delete("/api/admin/students/bulk", tags=["Admin"])
|
|
|
|
def bulk_delete_students(payload: DeleteUsersRequest, session: Session = Depends(get_session)):
|
|
|
|
def bulk_delete_students(payload: DeleteUsersRequest, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
|
|
|
|
for uid in payload.user_ids:
|
|
|
|
for uid in payload.user_ids:
|
|
|
|
user = session.get(models.User, uid)
|
|
|
|
user = session.get(models.User, uid)
|
|
|
|
if user:
|
|
|
|
if user:
|
|
|
|
@@ -312,7 +342,7 @@ def bulk_delete_students(payload: DeleteUsersRequest, session: Session = Depends
|
|
|
|
return {"success": True}
|
|
|
|
return {"success": True}
|
|
|
|
|
|
|
|
|
|
|
|
@app.put("/api/admin/students/{user_id}/toggle", tags=["Admin"])
|
|
|
|
@app.put("/api/admin/students/{user_id}/toggle", tags=["Admin"])
|
|
|
|
def toggle_student_active(user_id: UUID, session: Session = Depends(get_session)):
|
|
|
|
def toggle_student_active(user_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
|
|
|
|
user = session.get(models.User, user_id)
|
|
|
|
user = session.get(models.User, user_id)
|
|
|
|
if not user:
|
|
|
|
if not user:
|
|
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
@@ -407,14 +437,24 @@ def login(payload: LoginPayload, session: Session = Depends(get_session)):
|
|
|
|
# Allow demo/admin mock logic to still work if needed
|
|
|
|
# Allow demo/admin mock logic to still work if needed
|
|
|
|
# demo@pragativeer.com → Aarav (AI Coach demo account)
|
|
|
|
# demo@pragativeer.com → Aarav (AI Coach demo account)
|
|
|
|
DEMO_UUID = "11111111-1111-1111-1111-111111111111"
|
|
|
|
DEMO_UUID = "11111111-1111-1111-1111-111111111111"
|
|
|
|
if payload.email == "demo@pragativeer.com" and payload.password == "Pv9#kL2$mQ8!vX4@":
|
|
|
|
if payload.email == "demo@pragativeer.com" and payload.password == DEMO_PASSWORD:
|
|
|
|
import uuid as _uuid
|
|
|
|
import uuid as _uuid
|
|
|
|
demo_user = ensure_demo_user(_uuid.UUID(DEMO_UUID), session)
|
|
|
|
demo_user = ensure_demo_user(_uuid.UUID(DEMO_UUID), session)
|
|
|
|
token_id = str(demo_user.id) if demo_user else DEMO_UUID
|
|
|
|
token_id = str(demo_user.id) if demo_user else DEMO_UUID
|
|
|
|
access_token = create_access_token(data={"sub": token_id})
|
|
|
|
access_token = create_access_token(data={"sub": token_id})
|
|
|
|
return {"success": True, "role": "user", "token": access_token, "user_id": 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@":
|
|
|
|
if payload.email == "admin@pragativeer.com":
|
|
|
|
return {"success": True, "role": "admin", "token": "admin", "user_id": "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"}
|
|
|
|
# Admin login: password comes from env (ADMIN_PASSWORD). Never
|
|
|
|
|
|
|
|
# hardcode it — it lives in git history and every repo reader can
|
|
|
|
|
|
|
|
# log in as admin. Set ADMIN_PASSWORD in backend/.env (long random
|
|
|
|
|
|
|
|
# value) and rotate it whenever it leaks.
|
|
|
|
|
|
|
|
admin_password = os.getenv("ADMIN_PASSWORD", "")
|
|
|
|
|
|
|
|
if admin_password and payload.password == admin_password:
|
|
|
|
|
|
|
|
admin_user = session.exec(select(models.User).where(models.User.id == uuid.UUID(ADMIN_FALLBACK_ID))).first()
|
|
|
|
|
|
|
|
if admin_user:
|
|
|
|
|
|
|
|
access_token = create_access_token(data={"sub": str(admin_user.id)})
|
|
|
|
|
|
|
|
return {"success": True, "role": "admin", "token": access_token, "user_id": str(admin_user.id)}
|
|
|
|
|
|
|
|
raise HTTPException(status_code=401, detail="Invalid email or password")
|
|
|
|
|
|
|
|
|
|
|
|
user = session.exec(select(models.User).where(models.User.email == payload.email)).first()
|
|
|
|
user = session.exec(select(models.User).where(models.User.email == payload.email)).first()
|
|
|
|
if not user:
|
|
|
|
if not user:
|
|
|
|
@@ -422,7 +462,7 @@ def login(payload: LoginPayload, session: Session = Depends(get_session)):
|
|
|
|
|
|
|
|
|
|
|
|
# Handle legacy mock_hash from initial seed
|
|
|
|
# Handle legacy mock_hash from initial seed
|
|
|
|
if user.password_hash == "mock_hash":
|
|
|
|
if user.password_hash == "mock_hash":
|
|
|
|
if payload.password in ("Password123!", "Pv9#kL2$mQ8!vX4@", "admin123"):
|
|
|
|
if payload.password in ("Password123!", DEMO_PASSWORD, "admin123"):
|
|
|
|
user.password_hash = bcrypt.hashpw(payload.password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
|
|
|
user.password_hash = bcrypt.hashpw(payload.password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
|
|
|
session.add(user)
|
|
|
|
session.add(user)
|
|
|
|
session.commit()
|
|
|
|
session.commit()
|
|
|
|
@@ -935,7 +975,7 @@ def ensure_demo_user(user_id: UUID, session: Session) -> Optional[models.User]:
|
|
|
|
|
|
|
|
|
|
|
|
new_demo = models.User(
|
|
|
|
new_demo = models.User(
|
|
|
|
id=user_id,
|
|
|
|
id=user_id,
|
|
|
|
password_hash=bcrypt.hashpw("Pv9#kL2$mQ8!vX4@".encode('utf-8'), bcrypt.gensalt()).decode('utf-8'),
|
|
|
|
password_hash=bcrypt.hashpw(DEMO_PASSWORD.encode('utf-8'), bcrypt.gensalt()).decode('utf-8'),
|
|
|
|
last_active=now,
|
|
|
|
last_active=now,
|
|
|
|
created_at=now, # Align created_at to NOW so streak Day 1 = today
|
|
|
|
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
|
|
|
|
active_session_token=None, # No session token — demo users bypass single-device check
|
|
|
|
@@ -1270,7 +1310,7 @@ def get_leaderboard(
|
|
|
|
# --- ADMIN ROUTES ---
|
|
|
|
# --- ADMIN ROUTES ---
|
|
|
|
|
|
|
|
|
|
|
|
@app.get("/api/admin/settings", tags=["Admin"])
|
|
|
|
@app.get("/api/admin/settings", tags=["Admin"])
|
|
|
|
def get_system_settings(session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
|
|
def get_system_settings(session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
|
|
|
|
settings = session.get(models.SystemSettings, "global")
|
|
|
|
settings = session.get(models.SystemSettings, "global")
|
|
|
|
if not settings:
|
|
|
|
if not settings:
|
|
|
|
settings = models.SystemSettings()
|
|
|
|
settings = models.SystemSettings()
|
|
|
|
@@ -1287,7 +1327,7 @@ class SettingsUpdate(BaseModel):
|
|
|
|
trial_tier: Optional[models.TierEnum] = None
|
|
|
|
trial_tier: Optional[models.TierEnum] = None
|
|
|
|
|
|
|
|
|
|
|
|
@app.put("/api/admin/settings", tags=["Admin"])
|
|
|
|
@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)):
|
|
|
|
def update_system_settings(payload: SettingsUpdate, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
|
|
|
|
settings = session.get(models.SystemSettings, "global")
|
|
|
|
settings = session.get(models.SystemSettings, "global")
|
|
|
|
if not settings:
|
|
|
|
if not settings:
|
|
|
|
settings = models.SystemSettings()
|
|
|
|
settings = models.SystemSettings()
|
|
|
|
@@ -1305,7 +1345,7 @@ def update_system_settings(payload: SettingsUpdate, session: Session = Depends(g
|
|
|
|
return settings
|
|
|
|
return settings
|
|
|
|
|
|
|
|
|
|
|
|
@app.get("/api/admin/tests", tags=["Admin"])
|
|
|
|
@app.get("/api/admin/tests", tags=["Admin"])
|
|
|
|
def get_admin_tests(session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
|
|
def get_admin_tests(session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
|
|
|
|
tests = session.exec(select(models.Test)).all()
|
|
|
|
tests = session.exec(select(models.Test)).all()
|
|
|
|
return tests
|
|
|
|
return tests
|
|
|
|
|
|
|
|
|
|
|
|
@@ -1345,7 +1385,7 @@ async def upload_admin_test(
|
|
|
|
scheduled_date: Optional[datetime] = Form(None),
|
|
|
|
scheduled_date: Optional[datetime] = Form(None),
|
|
|
|
admin_id: Optional[UUID] = Form(None),
|
|
|
|
admin_id: Optional[UUID] = Form(None),
|
|
|
|
session: Session = Depends(get_session),
|
|
|
|
session: Session = Depends(get_session),
|
|
|
|
current_user: models.User = Depends(get_current_user),
|
|
|
|
current_user: models.User = Depends(require_admin),
|
|
|
|
):
|
|
|
|
):
|
|
|
|
"""
|
|
|
|
"""
|
|
|
|
PDF Test Upload — Phase 2 (AI LLM integration deferred).
|
|
|
|
PDF Test Upload — Phase 2 (AI LLM integration deferred).
|
|
|
|
@@ -1405,7 +1445,7 @@ async def upload_admin_test(
|
|
|
|
@app.post("/api/admin/tests/create_manual", tags=["Admin"])
|
|
|
|
@app.post("/api/admin/tests/create_manual", tags=["Admin"])
|
|
|
|
def create_manual_test(
|
|
|
|
def create_manual_test(
|
|
|
|
data: ManualTestCreate,
|
|
|
|
data: ManualTestCreate,
|
|
|
|
session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)
|
|
|
|
session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)
|
|
|
|
):
|
|
|
|
):
|
|
|
|
import uuid
|
|
|
|
import uuid
|
|
|
|
|
|
|
|
|
|
|
|
@@ -1523,7 +1563,7 @@ def create_manual_test(
|
|
|
|
return {"message": "Test created successfully", "test_id": test_obj.id}
|
|
|
|
return {"message": "Test created successfully", "test_id": test_obj.id}
|
|
|
|
|
|
|
|
|
|
|
|
@app.put("/api/admin/tests/{test_id}/toggle", tags=["Admin"])
|
|
|
|
@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)):
|
|
|
|
def toggle_test_status(test_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
|
|
|
|
test_obj = session.get(models.Test, test_id)
|
|
|
|
test_obj = session.get(models.Test, test_id)
|
|
|
|
if not test_obj:
|
|
|
|
if not test_obj:
|
|
|
|
raise HTTPException(status_code=404, detail="Test not found")
|
|
|
|
raise HTTPException(status_code=404, detail="Test not found")
|
|
|
|
@@ -1535,7 +1575,7 @@ def toggle_test_status(test_id: UUID, session: Session = Depends(get_session), c
|
|
|
|
return {"status": "success", "is_active": test_obj.is_active}
|
|
|
|
return {"status": "success", "is_active": test_obj.is_active}
|
|
|
|
|
|
|
|
|
|
|
|
@app.delete("/api/admin/tests/{test_id}", tags=["Admin"])
|
|
|
|
@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)):
|
|
|
|
def delete_test(test_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
|
|
|
|
from sqlmodel import select
|
|
|
|
from sqlmodel import select
|
|
|
|
test_obj = session.get(models.Test, test_id)
|
|
|
|
test_obj = session.get(models.Test, test_id)
|
|
|
|
if not test_obj:
|
|
|
|
if not test_obj:
|
|
|
|
@@ -1616,12 +1656,12 @@ class CouponUpdate(BaseModel):
|
|
|
|
expires_at: Optional[datetime] = None
|
|
|
|
expires_at: Optional[datetime] = None
|
|
|
|
|
|
|
|
|
|
|
|
@app.get("/api/admin/coupons", tags=["Admin"])
|
|
|
|
@app.get("/api/admin/coupons", tags=["Admin"])
|
|
|
|
def get_admin_coupons(session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
|
|
def get_admin_coupons(session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
|
|
|
|
coupons = session.exec(select(models.Coupon)).all()
|
|
|
|
coupons = session.exec(select(models.Coupon)).all()
|
|
|
|
return coupons
|
|
|
|
return coupons
|
|
|
|
|
|
|
|
|
|
|
|
@app.post("/api/admin/coupons", tags=["Admin"])
|
|
|
|
@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)):
|
|
|
|
def create_coupon(payload: CouponCreate, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
|
|
|
|
coupon = models.Coupon(**payload.dict())
|
|
|
|
coupon = models.Coupon(**payload.dict())
|
|
|
|
session.add(coupon)
|
|
|
|
session.add(coupon)
|
|
|
|
session.commit()
|
|
|
|
session.commit()
|
|
|
|
@@ -1629,7 +1669,7 @@ def create_coupon(payload: CouponCreate, session: Session = Depends(get_session)
|
|
|
|
return coupon
|
|
|
|
return coupon
|
|
|
|
|
|
|
|
|
|
|
|
@app.put("/api/admin/coupons/{coupon_id}", tags=["Admin"])
|
|
|
|
@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)):
|
|
|
|
def update_coupon(coupon_id: UUID, payload: CouponUpdate, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
|
|
|
|
coupon = session.get(models.Coupon, coupon_id)
|
|
|
|
coupon = session.get(models.Coupon, coupon_id)
|
|
|
|
if not coupon:
|
|
|
|
if not coupon:
|
|
|
|
raise HTTPException(status_code=404, detail="Coupon not found")
|
|
|
|
raise HTTPException(status_code=404, detail="Coupon not found")
|
|
|
|
@@ -1644,7 +1684,7 @@ def update_coupon(coupon_id: UUID, payload: CouponUpdate, session: Session = Dep
|
|
|
|
return coupon
|
|
|
|
return coupon
|
|
|
|
|
|
|
|
|
|
|
|
@app.delete("/api/admin/coupons/{coupon_id}", tags=["Admin"])
|
|
|
|
@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)):
|
|
|
|
def delete_coupon(coupon_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
|
|
|
|
coupon = session.get(models.Coupon, coupon_id)
|
|
|
|
coupon = session.get(models.Coupon, coupon_id)
|
|
|
|
if not coupon:
|
|
|
|
if not coupon:
|
|
|
|
raise HTTPException(status_code=404, detail="Coupon not found")
|
|
|
|
raise HTTPException(status_code=404, detail="Coupon not found")
|
|
|
|
@@ -1692,7 +1732,7 @@ class AdminUserPatch(BaseModel):
|
|
|
|
trial_days: Optional[int] = None # number of trial days to grant from now (0 = remove trial)
|
|
|
|
trial_days: Optional[int] = None # number of trial days to grant from now (0 = remove trial)
|
|
|
|
|
|
|
|
|
|
|
|
@app.get("/api/admin/students", tags=["Admin"])
|
|
|
|
@app.get("/api/admin/students", tags=["Admin"])
|
|
|
|
def list_all_students(session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
|
|
|
|
def list_all_students(session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
|
|
|
|
"""Return all users with basic stats for the admin student directory."""
|
|
|
|
"""Return all users with basic stats for the admin student directory."""
|
|
|
|
users = session.exec(select(models.User)).all()
|
|
|
|
users = session.exec(select(models.User)).all()
|
|
|
|
result = []
|
|
|
|
result = []
|
|
|
|
@@ -1730,7 +1770,7 @@ def list_all_students(session: Session = Depends(get_session), current_user: mod
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.get("/api/admin/students/{user_id}", tags=["Admin"])
|
|
|
|
@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)):
|
|
|
|
def get_student_detail(user_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
|
|
|
|
"""Full student profile for admin detail view."""
|
|
|
|
"""Full student profile for admin detail view."""
|
|
|
|
user = session.get(models.User, user_id)
|
|
|
|
user = session.get(models.User, user_id)
|
|
|
|
if not user:
|
|
|
|
if not user:
|
|
|
|
@@ -1828,7 +1868,7 @@ def get_student_detail(user_id: UUID, session: Session = Depends(get_session), c
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.patch("/api/admin/students/{user_id}", tags=["Admin"])
|
|
|
|
@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)):
|
|
|
|
def patch_student(user_id: UUID, payload: AdminUserPatch, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
|
|
|
|
"""Admin update: tier, name, streak, exam, location, trial."""
|
|
|
|
"""Admin update: tier, name, streak, exam, location, trial."""
|
|
|
|
user = session.get(models.User, user_id)
|
|
|
|
user = session.get(models.User, user_id)
|
|
|
|
if not user:
|
|
|
|
if not user:
|
|
|
|
|