Coverage for app/controllers/admin/userManagement.py: 28%
159 statements
« prev ^ index » next coverage.py v7.10.2, created at 2026-08-24 19:35 +0000
1import os
2from pathlib import Path
3import re
5from flask import render_template,request, flash, g, abort, redirect, send_file, url_for, jsonify, session
6from peewee import fn, JOIN, DoesNotExist
7from playhouse.shortcuts import model_to_dict
8from werkzeug.utils import secure_filename
11from app import app
12from app.logic.term import changeCurrentTerm
13from app.controllers.admin import admin_bp
14from app.logic.fileHandler import FileHandler
15from app.logic.userManagement import addCeltsAdmin,addCeltsStudentStaff,createSpreadsheetForRosters,addCeltsOperationsTeam,removeCeltsAdmin,removeCeltsStudentStaff,removeCeltsOperationsTeam
16from app.logic.userManagement import changeProgramInfo
17from app.logic.participants import getTrainingsForInterestedParticipants, getParticipantsForProgramForAY
18from app.logic.utils import selectSurroundingTerms
19from app.logic.term import addNextTerm, changeCurrentTerm
20from app.logic.users import getProgramInterest
21from app.logic.volunteers import setProgramManager
22from app.models.attachmentUpload import AttachmentUpload
23from app.models.programManager import ProgramManager
24from app.models.programBan import ProgramBan
25from app.models.user import User
26from app.models.term import Term
27from app.models.user import User
28from app.models.program import Program
30@admin_bp.route('/admin/manageUsers', methods = ['POST'])
31def manageUsers():
32 eventData = request.form
33 user = eventData['user']
34 method = eventData['method']
35 username = re.sub("[()]","", (user.split())[-1])
37 try:
38 user = User.get_by_id(username)
39 except Exception as e:
40 print(e)
41 flash(username + " is an invalid user.", "danger")
42 return ("danger", 500)
44 if method == "addCeltsAdmin":
45 if user.isStudent and not user.isCeltsStudentStaff:
46 flash(user.firstName + " " + user.lastName + " cannot be added as a CELTS-Link admin", 'danger')
47 else:
48 if user.isCeltsAdmin:
49 flash(user.firstName + " " + user.lastName + " is already a CELTS-Link Admin", 'danger')
50 else:
51 addCeltsAdmin(user)
52 flash(user.firstName + " " + user.lastName + " has been added as a CELTS-Link Admin", 'success')
53 elif method == "addCeltsStudentStaff":
54 if not user.isStudent:
55 flash(username + " cannot be added as CELTS Student Staff", 'danger')
56 else:
57 if user.isCeltsStudentStaff:
58 flash(user.firstName + " " + user.lastName + " is already a CELTS Student Staff", 'danger')
59 else:
60 addCeltsStudentStaff(user)
61 flash(user.firstName + " " + user.lastName + " has been added as a CELTS Student Staff", 'success')
62 elif method == "addCeltsOperationsTeam":
63 addCeltsOperationsTeam(user)
64 flash(user.firstName + " " + user.lastName + " has been added as a CELTS Operations Team member", "success")
65 elif method == "removeCeltsAdmin":
66 removeCeltsAdmin(user)
67 flash(user.firstName + " " + user.lastName + " is no longer a CELTS Admin ", 'success')
68 elif method == "removeCeltsStudentStaff":
69 removeCeltsStudentStaff(user)
70 flash(user.firstName + " " + user.lastName + " is no longer a CELTS Student Staff", 'success')
71 elif method == "removeCeltsOperationsTeam":
72 removeCeltsOperationsTeam(user)
73 flash(user.firstName + " " + user.lastName + " is no longer a CELTS Operations Team member", "success")
74 return ("success", 200)
76@admin_bp.route('/deleteProgramFile', methods=['POST'])
77def deleteProgramFile():
78 programFile=FileHandler(programId=request.form["programID"])
79 programFile.deleteFile(request.form["fileId"])
80 return ""
82@admin_bp.route('/admin/updateProgramInfo/<programID>', methods=['POST'])
83def updateProgramInfo(programID):
84 if g.current_user.canManageProgram(programID):
85 try:
86 programInfo = request.form # grabs user inputs
87 uploadedFile = request.files.get('modalProgramImage')
88 changeProgramInfo(programID, uploadedFile, **programInfo)
90 flash("Program updated", "success")
91 return redirect(url_for("admin.userManagement", accordion="program"))
92 except Exception as e:
93 flash('Error while updating program info.','warning')
94 abort(500,'Error while updating program.')
95 abort(403)
98@admin_bp.route('/admin/getProgramInfo/<programID>', methods = ['GET'])
99def getProgramInfo(programID):
100 if g.current_user.canManageProgram(programID):
101 try:
102 targetProgram = Program.get_by_id(programID)
103 programInfo = model_to_dict(targetProgram, recurse=False)
104 return jsonify([programInfo])
105 except DoesNotExist as e:
106 flash('Program not found')
107 print("Debug Here \n", e)
108 abort(404)
109 except Exception as e:
110 flash('Failed to retrieve data','warning')
111 print(e)
112 abort(500, 'Failed to retrieve data')
113 abort(403)
116@admin_bp.route('/admin', methods = ['GET'])
117def userManagement():
118 terms = selectSurroundingTerms(g.current_term)
120 currentPrograms = (
121 Program
122 .select(
123 Program,
124 fn.GROUP_CONCAT(fn.COALESCE(fn.CONCAT(User.firstName, ' ', User.lastName, '#', User.username), '')).alias('managers')
125 )
126 .join(ProgramManager, JOIN.LEFT_OUTER, on=(Program.id == ProgramManager.program))
127 .join(User, JOIN.LEFT_OUTER, on=(ProgramManager.user == User.username))
128 )
130 if not g.current_user.isCeltsAdmin and not g.current_user.isCeltsOperationsTeam: #Allows CELTS Operations Team to view all programs.
131 currentPrograms = currentPrograms.where(ProgramManager.user == g.current_user.username)
133 currentPrograms = list(currentPrograms.group_by(Program.id))
134 currentAdmins = list(User.select().where(User.isCeltsAdmin))
135 currentStudentStaff = list(User.select().where(User.isCeltsStudentStaff))
136 if g.current_user.isCeltsAdmin or g.current_user.isProgramManager or g.current_user.isCeltsOperationsTeam:
137 return render_template('admin/userManagement.html',
138 terms = terms,
139 programs = currentPrograms,
140 currentAdmins = currentAdmins,
141 currentStudentStaff = currentStudentStaff,
142 )
143 abort(403)
145@admin_bp.route('/admin/changeTerm', methods=['POST'])
146def changeTerm():
147 newTerm = changeCurrentTerm(int(request.form.get('id')))
148 flash(f"Current term successfully changed to {newTerm.description}", "success")
150 return ""
152@admin_bp.route('/admin/addNewTerm', methods = ['POST'])
153def addNewTerm():
154 newTerm = addNextTerm()
155 flash(f"Successfully added {newTerm.description}", "success")
157 return ""
159@admin_bp.route('/upload/<userFileCategory>/<userTermId>', methods = ['POST'])
160def upload(userFileCategory, userTermId):
161 try:
162 handbookTerm = Term.get_by_id(userTermId)
163 except DoesNotExist:
164 abort(405)
166 if userFileCategory not in ["laborHandbook", "volunteerHandbook"]:
167 abort(405)
168 fileCategory = userFileCategory
170 # Save file to fs
171 file = request.files[fileCategory]
172 newFilename = g.current_term.academicYear + "-" + fileCategory + "." + secure_filename(file.filename).split(".")[-1]
174 dir_path = Path(app.config['files']['base_path'], fileCategory)
175 dir_path.mkdir(parents=True, exist_ok=True)
176 full_path = os.path.join(dir_path, newFilename)
177 if os.path.exists(full_path):
178 os.remove(full_path)
179 file.save(full_path)
181 # Update all terms in the Academic Year with the handbook filename
182 for ayTerm in Term.select().where(Term.academicYear == handbookTerm.academicYear):
183 setattr(ayTerm, fileCategory, newFilename)
184 ayTerm.save()
186 # refresh the session with the new object
187 changeCurrentTerm(g.current_term, refreshOnly=True)
189 flash(f"Handbook saved successfully to {ayTerm.description}!", "success")
190 return redirect(request.referrer)
192@admin_bp.route('/viewRoster/<programID>', methods = ['GET'])
193def viewRoster(programID):
194 program = Program.get_by_id(programID)
196 interestedUsers = list(getProgramInterest(program))
197 trainedAndInterested = getTrainingsForInterestedParticipants(program, interestedUsers)
198 lastYearsParticipants = getParticipantsForProgramForAY(program, g.current_term.previousAcademicYear)
199 currentYearsParticipants = getParticipantsForProgramForAY(program, g.current_term.academicYear)
200 return render_template('admin/viewRoster.html',
201 program = program,
202 interestedUsers = interestedUsers,
203 trainedAndInterested = trainedAndInterested,
204 lastYearsParticipants = lastYearsParticipants,
205 currentYearsParticipants = currentYearsParticipants
206 )
208@admin_bp.route('/exportRosters/<programID>/<academicYear>', methods = ['GET'])
209def exportRosters(programID, academicYear):
210 try:
211 outFile = createSpreadsheetForRosters(academicYear, programID)
212 filepath = os.path.abspath(outFile)
213 return send_file(filepath, as_attachment=True, download_name=filepath.split("/")[-1], mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')
214 except DoesNotExist:
215 abort(403)