Spaces:
Paused
Paused
File size: 4,629 Bytes
33b8269 c34ad9d 8d7151f 33b8269 276f9fd 6c54712 33b8269 |
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 |
import uvicorn
from aiohttp import ClientSession
from gen_session import gen_session
from get_marks import get_marks_data
from get_grades import get_grades_data
from get_profile import get_profile_data
from fastapi.responses import JSONResponse
from get_timetable import get_timetable_data
from get_attendance import get_attendance_data
from get_exam_schedule import get_examSchedule_data
from get_sem_id import _get_all_sem_ids, _get_sem_id
from fastapi import FastAPI, Request, HTTPException, status, Form
import logging
logger = logging.getLogger(__name__)
app = FastAPI()
def basic_creds_check(username: str | None, password: str | None):
if not username or not password:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
@app.get("/")
async def root():
return {"message": "VTOP-AP API"}
async def handle_request(data_func, num_parameters, username, password):
basic_creds_check(username, password)
async with ClientSession() as sess:
print("password :" + password)
print("username : "+ username)
logger.info("username : "+ username + " password :" + password)
session_result = await gen_session(sess, username, password)
if session_result == 0:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
if num_parameters == 3:
return await data_func(sess, username, session_result)
else:
return await data_func(
sess,
username,
await _get_sem_id(sess, username, session_result),
session_result,
)
@app.post("/api/attendance")
@app.post("/api/timetable")
@app.post("/api/examSchedule")
async def handle_4param_data_functions(
request: Request, username: str = Form(...), password: str = Form(...)
):
data_func = {
"/api/attendance": get_attendance_data,
"/api/timetable": get_timetable_data,
"/api/examSchedule": get_examSchedule_data,
}
return await handle_request(data_func[request.url.path], 4, username, password)
@app.post("/api/grades")
@app.post("/api/profile")
@app.post("/api/semIDs")
async def handle_3param_data_functions(
request: Request, username: str = Form(...), password: str = Form(...)
):
data_func = {
"/api/grades": get_grades_data,
"/api/profile": get_profile_data,
"/api/semIDs": _get_all_sem_ids,
}
return await handle_request(data_func[request.url.path], 3, username, password)
@app.post("/api/verify")
async def verify_creds(username: str = Form(...), password: str = Form(...)):
basic_creds_check(username, password)
async with ClientSession() as sess:
session_result = await gen_session(sess, username, password)
if session_result == 0:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
else:
return JSONResponse(
content={"csrf_token": session_result}, status_code=status.HTTP_200_OK
)
@app.post("/api/all")
async def all_data(username: str = Form(...), password: str = Form(...)):
basic_creds_check(username, password)
async with ClientSession() as sess:
session_result = await gen_session(sess, username, password)
if session_result == 0:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
semID = await _get_sem_id(sess, username, session_result)
data = {
"profile": await get_profile_data(sess, username, session_result),
"attendance": await get_attendance_data(
sess, username, semID, session_result
),
"semIDs": await _get_all_sem_ids(sess, username, session_result),
"grades": await get_grades_data(sess, username, session_result),
"examSchedule": await get_examSchedule_data(
sess, username, semID, session_result
),
"timetable": await get_timetable_data(
sess, username, semID, session_result
),
}
return data
@app.post("/api/marks")
async def marks(
username: str = Form(...), password: str = Form(...), semID: str = Form(...)
):
basic_creds_check(username, password)
async with ClientSession() as sess:
session_result = await gen_session(sess, username, password)
if session_result == 0:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
return await get_marks_data(sess, username, semID, session_result)
if __name__ == "__main__":
uvicorn.run(app, host="localhost", port=7860)
|