Spaces:
Sleeping
Sleeping
File size: 5,834 Bytes
4624679 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 | from fastapi import APIRouter, HTTPException, Query, Depends
from typing import Optional
from pydantic import BaseModel
import database as db
router = APIRouter(prefix="/admin", tags=["admin"])
import time
_admin_cache = {}
def verify_admin(email: str = Query(...)):
"""Dependency to check if user is admin."""
now = time.time()
if email in _admin_cache and now - _admin_cache[email] < 3600:
return email
with db.db_cursor() as cur:
cur.execute('SELECT role FROM "user" WHERE email = %s', (email,))
row = cur.fetchone()
if not row or row["role"] != "admin":
raise HTTPException(status_code=403, detail="Admin privileges required")
_admin_cache[email] = now
return email
class SchoolCreate(BaseModel):
school_name: str
class ProgramCreate(BaseModel):
school_id: int
program_name: str
credit_weightage: float
total_eligible_students: int
class OfficerUpdate(BaseModel):
name: str
phone: Optional[str] = None
class StudentCreate(BaseModel):
name: str
usn: str
school_id: int
program_id: int
batch: str
is_active: bool = True
# --- Schools ---
@router.get("/schools")
def get_schools(email: str = Depends(verify_admin)):
return db.get_all_schools()
@router.post("/schools", status_code=201)
def create_school(body: SchoolCreate, email: str = Depends(verify_admin)):
return db.create_school(body.school_name)
@router.delete("/schools/{school_id}")
def delete_school(school_id: int, email: str = Depends(verify_admin)):
success = db.delete_school(school_id)
if not success:
raise HTTPException(status_code=404, detail="School not found")
return {"status": "ok"}
# --- Programs ---
@router.get("/programs")
def get_programs(email: str = Depends(verify_admin)):
return db.get_all_programs()
@router.post("/programs", status_code=201)
def create_program(body: ProgramCreate, email: str = Depends(verify_admin)):
return db.create_program(body.school_id, body.program_name, body.credit_weightage, body.total_eligible_students)
@router.delete("/programs/{program_id}")
def delete_program(program_id: int, email: str = Depends(verify_admin)):
success = db.delete_program(program_id)
if not success:
raise HTTPException(status_code=404, detail="Program not found")
return {"status": "ok"}
# --- Officers ---
@router.get("/officers")
def get_officers(email: str = Depends(verify_admin)):
return db.get_all_officers()
@router.get("/officers/{officer_id}/history")
def get_officer_history_admin(officer_id: int, year: int = Query(...), email: str = Depends(verify_admin)):
import datetime
now = datetime.datetime.now()
# 1. Fetch history first (1 query)
rows = db.get_officer_history(officer_id, year)
# 2. Check if current month is missing
if year == now.year:
has_current = any(r["month"] == now.month for r in rows)
if not has_current:
from main import recalculate_and_save_snapshot
snapshot = recalculate_and_save_snapshot(officer_id, now.month, now.year)
rows.append({
"month": now.month,
"starting_pool": snapshot["starting_pool"],
"target": snapshot["target"],
"placed": snapshot["placed"],
"prism_credits": snapshot["prism_credits"],
"prism_score": snapshot["prism_score"]
})
rows.sort(key=lambda x: x["month"])
return [
{
"month": r["month"],
"starting_pool": r["starting_pool"],
"target": r["target"],
"placed": r["placed"],
"prism_credits": r["prism_credits"],
"score": r["prism_score"],
}
for r in rows
]
@router.put("/officers/{officer_id}")
def update_officer(officer_id: int, body: OfficerUpdate, email: str = Depends(verify_admin)):
res = db.update_officer(officer_id, body.name, body.phone)
if not res:
raise HTTPException(status_code=404, detail="Officer not found")
return res
@router.delete("/officers/{officer_id}")
def delete_officer(officer_id: int, email: str = Depends(verify_admin)):
success = db.delete_officer(officer_id)
if not success:
raise HTTPException(status_code=404, detail="Officer not found")
return {"status": "ok"}
# --- Officer-Program Assignments ---
@router.post("/officers/{officer_id}/programs/{program_id}", status_code=201)
def assign_program(officer_id: int, program_id: int, email: str = Depends(verify_admin)):
db.assign_officer_to_program(officer_id, program_id)
return {"status": "assigned"}
@router.delete("/officers/{officer_id}/programs/{program_id}")
def unassign_program(officer_id: int, program_id: int, email: str = Depends(verify_admin)):
db.unassign_officer_from_program(officer_id, program_id)
return {"status": "unassigned"}
@router.get("/officers/{officer_id}/programs")
def get_officer_programs(officer_id: int, email: str = Depends(verify_admin)):
return db.get_officer_programs(officer_id)
# --- Students ---
@router.get("/students")
def get_students(email: str = Depends(verify_admin)):
return db.get_all_students(active_only=False)
@router.post("/students", status_code=201)
def create_student(body: StudentCreate, email: str = Depends(verify_admin)):
return db.create_student(body.name, body.usn, body.school_id, body.program_id, body.batch, body.is_active)
@router.delete("/students/{student_id}")
def delete_student(student_id: int, email: str = Depends(verify_admin)):
success = db.delete_student(student_id)
if not success:
raise HTTPException(status_code=404, detail="Student not found")
return {"status": "ok"}
|