survey_monitor / supp.py
potatoslc's picture
Upload 5 files
f6910c8 verified
Raw
History Blame Contribute Delete
6 kB
import pandas as pd
import numpy as np
import os
import requests
import sys
import zipfile,io
import time
import random
import json
school_name = r"sy24 BTS School List v2.csv"
all_school = r"sy24 title I list v2.csv"
df1 = pd.read_csv(all_school)
bts_school = pd.read_csv(school_name)
school_list = list(df1["school"])
school_dict = set(list(df1["school"]))
result_df = df1.copy()
result_df = result_df.iloc[:,0:2]
sz1,sz2 = result_df.shape
sz3,sz4 = bts_school.shape
result_df["Principal"] = [0]*sz1
result_df["Educator"] = [0]*sz1
result_df["BTS_SchoolList"] = ["NO"]*sz1
bts_school_dict = {}
for i in range(sz3):
if bts_school.iloc[i,0] not in bts_school_dict:
bts_school_dict[bts_school.iloc[i,0]] = set(bts_school.iloc[i,1])
else:
bts_school_dict[bts_school.iloc[i,0]].add(bts_school.iloc[i,1])
"""
This section is for api token, survey id, datacenter info,
"""
api_token = "cnfvvtLBU93Jlk9CVJMzJHqarMZeBVP7KfSlSY3U"
survey_id_2023 = "SV_0JrTQmOw07819rg"
survey_id = "SV_0B6bHCWd9hSE1eK"
datacenter = "ca1"
def get_survey_responses():
# part 1 get progress id
url1 = "https://yul1.qualtrics.com/API/v3/surveys/{0}/export-responses".format(survey_id)
payload1 = { "format": "csv","useLabels":True }
headers1 = {
"Content-Type": "application/json",
"Accept": "application/json",
"X-API-TOKEN": "cnfvvtLBU93Jlk9CVJMzJHqarMZeBVP7KfSlSY3U"
}
response1 = requests.post(url1, json=payload1, headers=headers1)
r1_json = response1.json()
try:
progess_id = r1_json["result"]["progressId"]
except:
print("error on step1")
time.sleep(random.random()*2+5.6)
# part 2 get file id
url2 = "https://yul1.qualtrics.com/API/v3/surveys/{0}/export-responses/{1}".format(survey_id,progess_id)
headers2 = {
"Accept": "application/json",
"X-API-TOKEN": "cnfvvtLBU93Jlk9CVJMzJHqarMZeBVP7KfSlSY3U"
}
response2 = requests.get(url2, headers=headers2)
r2_json = response2.json()
try:
file_id = r2_json["result"]["fileId"]
except:
print("error on step 2")
time.sleep(random.random()*3+3.14)
# part 3 get export file
url3 ="https://yul1.qualtrics.com/API/v3/surveys/{0}/export-responses/{1}/file".format(survey_id,file_id)
headers3 = {
"Accept": "application/octet-stream, application/json",
"X-API-TOKEN": "cnfvvtLBU93Jlk9CVJMzJHqarMZeBVP7KfSlSY3U"
}
response3 = requests.get(url3, headers=headers3)
try:
df = pd.read_csv(io.BytesIO(response3.content),compression='zip', header=0, sep=',', quotechar='"')
df_new = df[df["Finished"]==True]
except:
df = []
return df
def get_totalsize()->int:
sz1,_= result_df.shape
return sz1
def checkpassword(username:str,password:str)->int:
if username == "admin" and password =="Checkit2024!":
return 1
else:
return 0
return 0
def get_response_data(df,role,role_label,district_label,school_label)->dict:
res = {}
df_new = df.copy()
df_new = df_new[df_new[str(role_label)]==role]
col1 = list(df_new.columns)
p1_school = 0
p2_school = 0
fd =[]
for i in range(len(col1)):
if school_label in col1[i]:
fd.append(i)
p1_school = fd[0]
p2_school = fd[-1]
dstr_loc = 0
for i in range(len(col1)):
if district_label in col1[i]:
dstr_loc = i
break
sz1,_ = df_new.shape
for i in range(sz1):
temp_dstr = df_new.iloc[i,dstr_loc]
try:
temp_schl = (df_new.iloc[i,p1_school:p2_school+1].dropna())[0]
if (temp_dstr not in res) :
res[temp_dstr]= {temp_schl:1}
else:
if temp_schl not in res[temp_dstr]:
res[temp_dstr][temp_schl] = 1
else:
res[temp_dstr][temp_schl] +=1
except:
pass
return res
def get_responses_check_df(df,role_label,district_label,school_label):
# better to do that in {key:{key:value}}
# {district:{school:vcount}}
prn_dict = get_response_data(df,"Principal",role_label,district_label,school_label)
edu_dict = get_response_data(df,"Educator",role_label,district_label,school_label)
sz1,_ = result_df.shape
for i in range(sz1):
if "District" not in result_df.iloc[i,0]:
try:
result_df.iloc[i,2] = prn_dict["Chart School or Other"][result_df.iloc[i,1]]
except:
result_df.iloc[i,2] = "NO"
try:
result_df.iloc[i,3] = edu_dict["Charter School or Other"][result_df.iloc[i,1]]
except:
result_df.iloc[i,3] = "NO"
else:
try:
result_df.iloc[i,2] = prn_dict[result_df.iloc[i,0]][result_df.iloc[i,1]]
except:
result_df.iloc[i,2] = "NO"
try:
result_df.iloc[i,3] = edu_dict[result_df.iloc[i,0]][result_df.iloc[i,1]]
except:
result_df.iloc[i,3] = "NO"
if (result_df.iloc[i,0] in bts_school_dict) and (result_df.iloc[i,1] in bts_school_dict[result_df.iloc[i,0]]):
result_df.iloc[i,4] = "YES"
else:
result_df.iloc[i,4] = "NO"
#result_df.iloc[i,2] = "Yes" if (result_df.iloc[i,1] in prn_list )else "No"
#result_df.iloc[i,3] = "Yes" if (result_df.iloc[i,1] in edu_list )else "No"
return result_df
def get_responses_check_df_v2():
return
def get_dict_from_str(str1:str)->dict:
return json.unload(str1)