routes.py 54 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793
  1. from flask import render_template, request, session, jsonify, flash, make_response, redirect, url_for, abort, send_file, render_template_string
  2. from flask_login import login_required, current_user
  3. from app.main import bp
  4. from app.extensions import db, dict_row, dict_instance, extract_exc
  5. from app.models.users import tbl_users
  6. from app.models.cv_data import *
  7. from app.auth import check_admin, check_capture
  8. from sqlalchemy import insert, delete, update, select
  9. from sqlalchemy import and_, or_, not_, func
  10. import json, requests, re, os
  11. from .forms import RecordForm, QualificationTypeForm, QualificationForm, QualificationTypeRemovalForm, QualificationRemovalForm, RecordQualificationForm, RecordQualificationRemovalForm
  12. from .forms import RoleDepartmentForm, RoleForm, RoleDepartmentRemovalForm, RoleRemovalForm, \
  13. UploadForm, UploadRemovalForm, FilterForm
  14. from .forms import UserForm, UserRemovalForm
  15. import re
  16. import datetime as datetime_class
  17. from datetime import datetime, time, timezone, timedelta, date
  18. from app import Config
  19. from app.extensions import dict_instance, dict_row
  20. import io
  21. def end_user_debugging(s_additional:str = "", bl_html:bool = True):
  22. """just for initial testing"""
  23. d_exc = extract_exc(s_additional)
  24. s_dt = datetime.now().astimezone(timezone(timedelta(hours=2))).strftime("%Y-%m-%d %H:%M:%S")
  25. s_logging = f"<ul><li>date-time: {s_dt}</li>\n<li>line no.: {d_exc['i_lineno']}</li>\n<li>filename: {d_exc['s_filename']}</li>\n<li>except: {d_exc['s_except']}</li>\n<li>additional: {d_exc['s_additional']}</li>\n</ul>" if (isinstance(d_exc, dict) and bl_html) else f"date-time: {s_dt}\nindex-landing page\n" + str(d_exc["s_except"])
  26. return f"<!doctype html>\n<html lang=\"en\">\n<head>\n<title>Issue</title>\n</head>\n<body>\n<h2>Issue details</h2>\n{s_logging}\n</body>\n</html>"
  27. # end of end_user_debugging function
  28. @bp.route("/", methods=["GET", "POST"])
  29. @login_required
  30. def index():
  31. """landing page"""
  32. try:
  33. # filtering choices and current data
  34. filter_form = FilterForm()
  35. # populate some select choices
  36. l_role_departments = list(map(dict_row, db.session.query(tbl_role_departments.id, tbl_role_departments.v_department_name).order_by(tbl_role_departments.v_department_name).all()))
  37. for ix, d in enumerate(l_role_departments): l_role_departments[ix] = (d["id"], d["v_department_name"])
  38. filter_form.sel_role_department.choices = [(0, "---")] + l_role_departments
  39. # populate roles if department was selected
  40. i_department_id = request.form.get("sel_role_department", 0, type=int)
  41. if i_department_id>0:
  42. q_roles = db.session.query(tbl_roles.id, tbl_roles.v_role_name).filter(tbl_roles.i_department_id==i_department_id)
  43. if q_roles.count()>0:
  44. q_roles = q_roles.order_by(tbl_roles.v_role_name)
  45. l_roles = list(map(dict_row, q_roles.all()))
  46. for ix, d in enumerate(l_roles): l_roles[ix] = (d["id"], d["v_role_name"])
  47. filter_form.sel_role.choices = [(0, "---")] + l_roles
  48. # end of checking result count of query
  49. # end of checking if department submitted
  50. l_all_languages = list(map(dict_row, db.session.query(tbl_all_languages.v_language_abbreviation, tbl_all_languages.v_language_name).filter(tbl_all_languages.bl_south_african==True).order_by(tbl_all_languages.v_language_name).all()))
  51. for ix, d in enumerate(l_all_languages): l_all_languages[ix] = (d["v_language_abbreviation"], d["v_language_name"])
  52. filter_form.sel_language.choices = [("", "---")] + l_all_languages
  53. l_qualification_types = list(map(dict_row, db.session.query(tbl_qualification_types.id, tbl_qualification_types.v_qualification_type).order_by(tbl_qualification_types.v_qualification_type).all()))
  54. for ix, d in enumerate(l_qualification_types): l_qualification_types[ix] = (d["id"], d["v_qualification_type"])
  55. filter_form.sel_qualification_type_1.choices = [(0, "---")] + l_qualification_types
  56. filter_form.sel_qualification_type_2.choices = [(0, "---")] + l_qualification_types
  57. i_type_1 = request.form.get("sel_qualification_type_1", 0, type=int)
  58. i_type_2 = request.form.get("sel_qualification_type_2", 0, type=int)
  59. q_qualifications = db.session.query(tbl_qualifications.id, tbl_qualifications.i_qualification_type, tbl_qualifications.v_qualification_name).order_by(tbl_qualifications.v_qualification_name)
  60. l_qualifications_1 = q_qualifications.filter(tbl_qualifications.i_qualification_type==i_type_1).order_by(tbl_qualifications.v_qualification_name).all() if i_type_1>0 else []
  61. l_qualifications_2 = q_qualifications.filter(tbl_qualifications.i_qualification_type==i_type_2).order_by(tbl_qualifications.v_qualification_name).all() if i_type_2>0 else []
  62. l_qualifications_1 = list(map(dict_row, l_qualifications_1))
  63. l_qualifications_2 = list(map(dict_row, l_qualifications_2))
  64. for ix, d in enumerate(l_qualifications_1): l_qualifications_1[ix] = (d["id"], d["v_qualification_name"])
  65. for ix, d in enumerate(l_qualifications_2): l_qualifications_2[ix] = (d["id"], d["v_qualification_name"])
  66. filter_form.sel_qualification_1.choices = [(0, "---")] + l_qualifications_1
  67. filter_form.sel_qualification_2.choices = [(0, "---")] + l_qualifications_2
  68. # initiate query that will possibly get filtered
  69. q_records = db.session.query(tbl_records.id, tbl_records.v_name_1, tbl_records.v_surname, tbl_records.si_years_experience, tbl_records.v_contact_number, tbl_records.v_email)
  70. if filter_form.validate_on_submit():
  71. # possible filter values
  72. bl_case_sensitive, i_role_department_id, i_role_id, s_name, s_surname, s_id_number = (filter_form.chk_case_sensitive.data, filter_form.sel_role_department.data, filter_form.sel_role.data, str(filter_form.txt_name.data), str(filter_form.txt_surname.data), str(filter_form.txt_id_number.data))
  73. s_sap_k_level, s_language = (filter_form.sel_sap_k_level.data, filter_form.sel_language.data)
  74. i_qualification_type_1, i_qualification_type_2, i_qualification_1, i_qualification_2 = (filter_form.sel_qualification_type_1.data, filter_form.sel_qualification_type_2.data, filter_form.sel_qualification_1.data, filter_form.sel_qualification_2.data)
  75. # apply filters
  76. if i_role_department_id>0: q_records = q_records.filter(tbl_records.i_department_id==i_role_department_id)
  77. if i_role_id>0: q_records = q_records.filter(tbl_records.i_role_id==i_role_id)
  78. if s_name!="":
  79. s_name = f"%{s_name}%"
  80. if bl_case_sensitive:
  81. sq_names = select(tbl_records.id).filter(or_(tbl_records.v_name_1.like(s_name), tbl_records.v_name_2.like(s_name), tbl_records.v_name_3.like(s_name)))
  82. else:
  83. sq_names = select(tbl_records.id).filter(or_(lower(tbl_records.v_name_1).like(func.lower(s_name)), func.lower(tbl_records.v_name_2).like(func.lower(s_name)), func.lower(tbl_records.v_name_3).like(func.lower(s_name))))
  84. # end of checking if case sensitivity applies
  85. q_records = q_records.filter(tbl_records.id.in_(sq_names))
  86. # end of checking if need to check for name substrings
  87. if s_surname!="":
  88. s_surname = f"%{s_surname}%"
  89. if bl_case_sensitive:
  90. q_records = q_records.filter(tbl_records.v_surname.like(s_surname))
  91. else:
  92. q_records = q_records.filter(func.lower(tbl_records.v_surname).like(func.lower(s_surname)))
  93. # end of checking if case sensitivity applies
  94. # end of checking for surname filter
  95. if s_id_number!="":
  96. s_id_number = f"%{s_id_number}%"
  97. if bl_case_sensitive:
  98. q_records = q_records.filter(tbl_records.v_id_number.like(s_id_number))
  99. else:
  100. q_records = q_records.filter(func.lower(tbl_records.v_id_number).like(func.lower(s_id_number)))
  101. # end of checking if case sensitivity applies
  102. # end of checking for ID number filter
  103. if s_sap_k_level!="": q_records = q_records.filter(tbl_records.v_sap_k_level==s_sap_k_level)
  104. if s_language!="":
  105. sq_languages = select(tbl_languages.i_record_id.distinct()).filter(tbl_languages.v_language_abbreviation==s_language)
  106. q_records = q_records.filter(tbl_records.id.in_(sq))
  107. # end of checking if need to use language-matching subquery
  108. if i_qualification_type_1>0:
  109. if i_qualification_1>0:
  110. sq_qualification_1 = select(tbl_record_qualifications.i_record_id.distinct()).filter(tbl_record_qualifications.i_qualification_id==i_qualification_1)
  111. q_records = q_records.filter(tbl_records.id.in_(sq_qualification_1))
  112. else:
  113. sq_qualification_1 = select(tbl_record_qualifications.i_record_id.distinct()).join(tbl_qualifications, tbl_qualifications.id==tbl_record_qualifications.i_qualification_id, isouter=False).filter(tbl_qualifications.i_qualification_type==i_qualification_type_1)
  114. q_records = q_records.filter(tbl_records.id.in_(sq_qualification_1))
  115. # end of checking if need to filter qualification itself or just type
  116. # end of checking for qualification type 1
  117. if i_qualification_type_2>0:
  118. if i_qualification_2>0:
  119. sq_qualification_2 = select(tbl_record_qualifications.i_record_id.distinct()).filter(tbl_record_qualifications.i_qualification_id==i_qualification_2)
  120. q_records = q_records.filter(tbl_records.id.in_(sq_qualification_2))
  121. else:
  122. sq_qualification_2 = select(tbl_record_qualifications.i_record_id.distinct()).join(tbl_qualifications, tbl_qualifications.id==tbl_record_qualifications.i_qualification_id, isouter=False).filter(tbl_qualifications.i_qualification_type==i_qualification_type_2)
  123. q_records = q_records.filter(tbl_records.id.in_(sq_qualification_2))
  124. # end of checking if need to filter qualification itself or just type
  125. # end of checking for qualification type 2
  126. # end of checking for filter form submission
  127. i_page = request.form.get("hid_page", 1, type=int)
  128. q_records = q_records.order_by(tbl_records.v_surname, tbl_records.v_name_1)
  129. pagination = q_records.paginate(page=i_page, per_page=20)
  130. l_records = list(map(dict_row, pagination.items))
  131. # l_records = list(map(dict_row, q_records.all()))
  132. for ix, record in enumerate(l_records):
  133. i_record = record["id"]
  134. l_record_qualifications = list(map(dict_row, db.session.query(tbl_record_qualifications.id, tbl_qualification_types.v_qualification_type, tbl_qualifications.v_qualification_name, tbl_record_qualifications.d_acquired).join(tbl_qualifications, tbl_record_qualifications.i_qualification_id==tbl_qualifications.id, isouter=False).join(tbl_qualification_types, tbl_qualifications.i_qualification_type==tbl_qualification_types.id, isouter=False).filter(tbl_record_qualifications.i_record_id==i_record).order_by(tbl_record_qualifications.d_acquired, tbl_qualifications.v_qualification_name).all()))
  135. l_records[ix]["record_qualifications"] = l_record_qualifications
  136. # end of looping through records to retrieve additional information
  137. s_base = "base_bs.html" if Config.BOOTSTRAP else "base.html"
  138. s_url = url_for("main.index")#, _external=True)
  139. return render_template("index.html", js=True, base=s_base, url=s_url, records=l_records, filter_form=filter_form, paging=pagination)
  140. except Exception as exc:
  141. return render_template_string(end_user_debugging("index/landing page"))
  142. # end of surrounding try-except
  143. # end of index view function for route /
  144. @bp.route("/icons", methods=["GET"])
  145. def icons():
  146. """icons preview - purely testing"""
  147. return render_template("icons.html")
  148. # end of icons view function for route /
  149. @bp.route("/capture_record/<int:i_record>/", methods=["GET", "POST"])
  150. @bp.route("/capture_record/", defaults={"i_record": 0}, methods=["GET", "POST"])
  151. @login_required
  152. @check_capture
  153. def capture_record(i_record:int = 0):
  154. """add/update C.V. record"""
  155. try:
  156. l_all_languages = list(map(dict_row, db.session.query(tbl_all_languages.v_language_abbreviation, tbl_all_languages.v_language_name).filter(tbl_all_languages.bl_south_african==True).order_by(tbl_all_languages.v_language_name).all()))
  157. l_departments = list(map(dict_row, db.session.query(tbl_role_departments.id, tbl_role_departments.v_department_name).order_by(tbl_role_departments.v_department_name).all()))
  158. l_roles = []
  159. i_department_id, i_role_id =(0, 0)
  160. form = RecordForm()
  161. remove_qualification_form = RecordQualificationRemovalForm()
  162. qualification_form = RecordQualificationForm()
  163. i_qualification_type_id = request.form.get("sel_qualification_type", default=0, type=int)
  164. l_qualification_types = list(map(dict_row, db.session.query(tbl_qualification_types.id, tbl_qualification_types.v_qualification_type).order_by(tbl_qualification_types.v_qualification_type).all()))
  165. l_qualifications = []
  166. if len(l_qualification_types)>0:
  167. i_qualification_type_id = int(l_qualification_types[0]["id"]) if i_qualification_type_id<1 else i_qualification_type_id
  168. l_qualifications = list(map(dict_row, db.session.query(tbl_qualifications.id, tbl_qualifications.v_qualification_name).filter(tbl_qualifications.i_qualification_type==i_qualification_type_id).order_by(tbl_qualifications.v_qualification_name).all()))
  169. if len(l_qualifications)>0:
  170. l_choices = list(map((lambda x : (x["id"], x["v_qualification_name"])), l_qualifications))
  171. qualification_form.sel_qualification.choices = l_choices
  172. # end of checking if qualifications were retrieved
  173. # end of checking if there are qualification types
  174. if form.validate_on_submit() and request.form.get("btn_save", "")=="Save":
  175. s_action = "updated"
  176. o_record = None
  177. s_name_1, s_name_2, s_name_3, s_surname = (form.txt_name_1.data, form.txt_name_2.data, form.txt_name_3.data, form.txt_surname.data)
  178. s_id_number = form.txt_id_number.data
  179. s_gender = form.sel_gender.data
  180. i_years_experience = form.txt_years_experience.data
  181. s_sap_k_level = form.sel_sap_k_level.data
  182. s_contact_number = form.txt_contact_number.data
  183. s_email = form.txt_email.data
  184. i_department_id = request.form.get("sel_department_id", 0, type=int)
  185. i_role_id = request.form.get("sel_role_id", 0, type=int)
  186. l_languages = request.form.getlist("hid_langs")
  187. l_language_levels = request.form.getlist("hid_lang_levels")
  188. if i_record > 0:
  189. o_record = db.session.get(tbl_records, i_record)
  190. o_record.v_name_1, o_record.v_name_2, o_record.v_name_3, o_record.v_surname = (s_name_1, s_name_2, s_name_3, s_surname)
  191. o_record.v_id_number, o_record.c_gender, o_record.si_years_experience, o_record.v_sap_k_level = (s_id_number, s_gender, i_years_experience, s_sap_k_level)
  192. o_record.v_contact_number, o_record.v_email = (s_contact_number, s_email)
  193. o_record.i_department_id, o_record.i_role_id = (i_department_id, i_role_id)
  194. db.session.add(o_record)
  195. db.session.commit()
  196. else:
  197. s_action = "inserted"
  198. # save initial record
  199. o_record = tbl_records(v_name_1=s_name_1, v_name_2=s_name_2, v_name_3=s_name_3, v_surname=s_surname, v_id_number=s_id_number, c_gender=s_gender, si_years_experience=i_years_experience, v_sap_k_level=s_sap_k_level, v_contact_number=s_contact_number, v_email=s_email, i_department_id=i_department_id, i_role_id=i_role_id)
  200. db.session.add(o_record)
  201. db.session.commit()
  202. i_record = int(o_record.id) if o_record.id is not None else 0
  203. # end of checking if new or existing record
  204. # save languages linked to parent record
  205. if i_record > 0:
  206. if len(l_languages)==len(l_language_levels):
  207. # first remove any prior language records that do not match currently submitted languages
  208. q_delete = delete(tbl_languages).where(tbl_languages.i_record_id==i_record).where(tbl_languages.v_language_abbreviation.not_in(l_languages))
  209. db.session.execute(q_delete)
  210. # insert/update language records
  211. for ix, s_lang in enumerate(l_languages):
  212. s_language_name = ""
  213. q_language = db.session.query(tbl_all_languages.v_language_name).filter(tbl_all_languages.v_language_abbreviation==s_lang)
  214. if q_language.count() > 0:
  215. s_language_name = q_language.first()[0]
  216. # end of checking for matching language name record
  217. del q_language
  218. q_language = None
  219. if db.session.query(tbl_languages).filter(tbl_languages.i_record_id==i_record, tbl_languages.v_language_abbreviation==s_lang).count()<1:
  220. q_language = insert(tbl_languages).values({"i_record_id": i_record, "v_language_abbreviation": s_lang, "v_language_name": s_language_name, "si_level": int(l_language_levels[ix]), "si_ranking": ix+1})
  221. else:
  222. q_language = update(tbl_languages).where(tbl_languages.i_record_id==i_record,tbl_languages.v_language_abbreviation==s_lang).values({"v_language_name": s_language_name, "si_level": int(l_language_levels[ix]), "si_ranking": ix+1})
  223. # end of checking if insert or update
  224. db.session.execute(q_language)
  225. db.session.commit()
  226. # end of looping through languages
  227. # end of checking language list lengths
  228. flash(f"Record successfully {s_action}", "success")
  229. # end of checking if initial record submission took place
  230. elif qualification_form.validate_on_submit() and i_record>0 and request.form.get("btn_save_qualification", "")=="Save":
  231. i_record_qualification_id, i_qualification_id, d_acquired = (int(qualification_form.hid_record_qualification_id.data), int(qualification_form.sel_qualification.data), qualification_form.d_acquired.data)
  232. if i_record_qualification_id>0:
  233. o_record_qualification = db.session.get(tbl_record_qualifications, i_record_qualification_id)
  234. if o_record_qualification is not None:
  235. o_record_qualification.i_qualification_id = i_qualification_id
  236. o_record_qualification.d_acquired = d_acquired
  237. db.session.add(o_record_qualification)
  238. db.session.commit()
  239. flash("Qualification record updated")
  240. # end of checking if prior record retrieved
  241. else:
  242. o_record_qualification = tbl_record_qualifications(i_record_id=i_record, i_qualification_id=i_qualification_id, d_acquired=d_acquired)
  243. db.session.add(o_record_qualification)
  244. db.session.commit()
  245. flash("Qualification record recorded")
  246. # end of checking if existing or new qualification record
  247. elif remove_qualification_form.validate_on_submit() and request.form.get("hid_remove_qualification_id", None) is not None:
  248. i_qualification_removal_id = request.form.get("hid_remove_qualification_id", type=int)
  249. o_record_qualification = db.session.get(tbl_record_qualifications, i_qualification_removal_id)
  250. if o_record_qualification is not None:
  251. db.session.delete(o_record_qualification)
  252. db.session.commit()
  253. flash("Qualification removed")
  254. # end of making sure retrieved record
  255. # end of checking for form submission
  256. form.hid_record_id.data = i_record
  257. remove_qualification_form.hid_remove_qualification_id.data = 0
  258. # populate record info if existing record
  259. if i_record>0:
  260. o_record = db.session.get(tbl_records, i_record)
  261. if o_record is not None:
  262. form.hid_record_id.data = i_record
  263. if o_record.i_department_id>0:
  264. i_department_id = o_record.i_department_id
  265. i_role_id = o_record.i_role_id
  266. l_roles = list(map(dict_row, db.session.query(tbl_roles.id, tbl_roles.v_role_name).filter(tbl_roles.i_department_id==i_department_id).order_by(tbl_roles.v_role_name).all()))
  267. # end of checking which department roles should populate
  268. form.txt_name_1.data, form.txt_name_2.data, form.txt_name_3.data, form.txt_surname.data = (o_record.v_name_1, o_record.v_name_2, o_record.v_name_3, o_record.v_surname)
  269. form.txt_id_number.data = o_record.v_id_number
  270. form.sel_gender.data = o_record.c_gender
  271. form.txt_years_experience.data = o_record.si_years_experience
  272. form.sel_sap_k_level.data = o_record.v_sap_k_level
  273. form.txt_contact_number.data = o_record.v_contact_number
  274. form.txt_email.data = o_record.v_email
  275. else:
  276. i_record = 0
  277. # end of making sure valid record
  278. # end of checking if existing record
  279. qualification_form.hid_record_qualification_id.data, qualification_form.sel_qualification.data, qualification_form.d_acquired.data = (0, None, None)
  280. # form.txt_name_1.data, form.txt_name_2.data, form.txt_name_3.data, form.txt_surname.data = ("", "", "", "")
  281. # form.txt_id_number.data, form.sel_gender.data, form.txt_years_experience.data, form.sel_sap_k_level.data, form.txt_contact_number.data, form.txt_email.data = ("", "m", "0", "", "", "")
  282. l_languages = db.session.query(tbl_languages.v_language_abbreviation, tbl_languages.v_language_name, tbl_languages.si_level).filter(tbl_languages.i_record_id==i_record).order_by(tbl_languages.si_ranking).all()
  283. l_languages = list(map(dict_row, l_languages))
  284. l_record_qualifications = list(map(dict_row, db.session.query(tbl_record_qualifications.id, tbl_qualification_types.v_qualification_type, tbl_qualifications.v_qualification_name, tbl_record_qualifications.d_acquired).join(tbl_qualifications, tbl_record_qualifications.i_qualification_id==tbl_qualifications.id, isouter=False).join(tbl_qualification_types, tbl_qualifications.i_qualification_type==tbl_qualification_types.id, isouter=False).filter(tbl_record_qualifications.i_record_id==i_record).order_by(tbl_record_qualifications.d_acquired, tbl_qualifications.v_qualification_name).all()))
  285. s_base = "base_bs.html" if Config.BOOTSTRAP else "base.html"
  286. s_url = url_for("main.capture_record")#, _external=True)
  287. return render_template("capture_record.html", js=True, base=s_base, url=s_url, form=form, record_id=i_record, languages=l_languages, all_languages=l_all_languages, qualification_types=l_qualification_types, qualification_form=qualification_form, record_qualifications=l_record_qualifications, departments=l_departments, roles=l_roles, department_id=i_department_id, role_id=i_role_id, remove_qualification_form=remove_qualification_form)
  288. except Exception as exc:
  289. return render_template_string(end_user_debugging("capture_record"))
  290. # end of surrounding try-except
  291. # end of capture_record view function for route /capture_record
  292. @bp.route("/download_upload/<int:i_upload>/<bl_download>/", methods=["GET"])
  293. @bp.route("/download_upload/<int:i_upload>/", defaults={"bl_download": True}, methods=["GET"])
  294. @login_required
  295. def download_upload(i_upload:int = 0, bl_download:bool = True):
  296. """open upload within browser or activate download thereof"""
  297. try:
  298. bl_download = True if str(bl_download)=="True" else False
  299. bl_download = bool(bl_download)
  300. o_out = None
  301. o_upload = db.session.get(tbl_uploads, i_upload)
  302. if o_upload is not None:
  303. o_out = io.BytesIO(o_upload.b_file)
  304. s_mimetype = o_upload.v_mime_type
  305. s_filename = o_upload.v_filename
  306. # resp = send_file(o_out, mimetype=s_mimetype, as_attachment=bl_download, download_name=s_filename)
  307. resp = make_response(o_out)
  308. resp.headers.set("Content-Disposition", f"attachment; filename=\"{s_filename}\"" if bl_download else f"inline; filename=\"{s_filename}\"")
  309. resp.headers.set("Content-Type", s_mimetype)
  310. return resp
  311. else:
  312. return make_response("No upload retrieved", 500)
  313. # end of None check for record
  314. except Exception as exc:
  315. return render_template_string(end_user_debugging("download_upload"))
  316. # end of surrounding try-except
  317. # end of download_upload view function for route /download_upload/<int:i_upload>/[<bl_download>]
  318. def qualification_type_choices():
  319. """return possible qualification type choices"""
  320. l_choices = []
  321. l_qualification_types = list(map(dict_row, db.session.query(tbl_qualification_types.id, tbl_qualification_types.v_qualification_type).order_by(tbl_qualification_types.v_qualification_type).all()))
  322. l_type_choices = list(map((lambda x : (x["id"], x["v_qualification_type"])), l_qualification_types))
  323. if len(l_type_choices)>0:
  324. l_choices = l_type_choices
  325. # end of checking if type choices exist
  326. return (l_qualification_types, l_choices)
  327. # end of qualification_type_choices function
  328. @bp.route("/qualifications/", methods=["GET", "POST"])
  329. @login_required
  330. @check_admin
  331. def qualifications():
  332. """manage qualification types and qualifications to make use of in rest of site"""
  333. try:
  334. i_type_filter = 0
  335. type_form = QualificationTypeForm()
  336. form = QualificationForm()
  337. l_qualification_types, form.sel_qualification_type.choices = qualification_type_choices()
  338. remove_type_form = QualificationTypeRemovalForm()
  339. remove_form = QualificationRemovalForm()
  340. if type_form.validate_on_submit() and request.form.get("btn_save_type", "")=="Save":
  341. i_type_id, s_type = (int(type_form.hid_qualification_type_id.data), type_form.txt_qualification_type.data)
  342. if i_type_id > 0:
  343. o_type = db.session.get(tbl_qualification_types, i_type_id)
  344. if o_type is not None:
  345. o_type.v_qualification_type = s_type
  346. db.session.add(o_type)
  347. db.session.commit()
  348. flash("Type updated", "success")
  349. # end of checking if record existed
  350. else:
  351. if db.session.query(tbl_qualification_types.id).filter(tbl_qualification_types.v_qualification_type==s_type).count()<1:
  352. db.session.add(tbl_qualification_types(v_qualification_type=s_type))
  353. db.session.commit()
  354. flash("Type inserted", "success")
  355. # end of making sure new type is not a duplicate
  356. # end of checking if new or existing type record
  357. elif remove_type_form.validate_on_submit() and request.form.get("hid_remove_qualification_type_id", None) is not None:
  358. i_type_id = int(remove_type_form.hid_remove_qualification_type_id.data)
  359. o_type = db.session.get(tbl_qualification_types, i_type_id)
  360. if o_type is not None:
  361. db.session.delete(o_type)
  362. db.session.commit()
  363. flash("Type removed", "success")
  364. # end of checking if record existed
  365. elif form.validate_on_submit() and request.form.get("btn_save_qualification", "")=="Save":
  366. i_qualification_id, i_qualification_type, s_qualification_name, s_description = (int(form.hid_qualification_id.data), int(form.sel_qualification_type.data), str(form.txt_qualification_name.data), str(form.txt_description.data))
  367. if i_qualification_id>0:
  368. o_qualification = db.session.get(tbl_qualifications, i_qualification_id)
  369. if o_qualification is not None:
  370. if db.session.query(tbl_qualifications).filter(tbl_qualifications.id!=i_qualification_id, tbl_qualifications.v_qualification_name==s_qualification_name, tbl_qualifications.i_qualification_type==i_qualification_type).count()<1:
  371. o_qualification.i_qualification_type = i_qualification_type
  372. o_qualification.v_qualification_name = s_qualification_name
  373. o_qualification.v_description = s_description
  374. db.session.add(o_qualification)
  375. db.session.commit()
  376. flash("Qualification updated")
  377. i_type_filter = i_qualification_type
  378. else:
  379. flash("There is already a qualification under same type category with same name")
  380. i_type_filter = i_qualification_type
  381. # end of checking for alternative duplicate
  382. # end of retrieval check
  383. else:
  384. if db.session.query(tbl_qualifications).filter(tbl_qualifications.v_qualification_name==s_qualification_name, tbl_qualifications.i_qualification_type==i_qualification_type).count()<1:
  385. o_qualification = tbl_qualifications(i_qualification_type=i_qualification_type, v_qualification_name=s_qualification_name, v_description=s_description)
  386. db.session.add(o_qualification)
  387. db.session.commit()
  388. flash("Qualification added")
  389. i_type_filter = i_qualification_type
  390. else:
  391. flash("There is an existing qualification with same name under same type category")
  392. # end of making sure would not count as duplicate
  393. # end of checking if new or existing qualification record
  394. elif remove_form.validate_on_submit() and request.form.get("hid_remove_qualification_id", None) is not None:
  395. i_qualification_id = int(remove_form.hid_remove_qualification_id.data)
  396. o_qualification = db.session.get(tbl_qualifications, i_qualification_id)
  397. if o_qualification is not None:
  398. db.session.delete(o_qualification)
  399. db.session.commit()
  400. flash("Qualification removed", "success")
  401. # end of checking if record existed
  402. # end of checking for form submissions
  403. type_form.hid_qualification_type_id.data, type_form.txt_qualification_type.data = (0, "")
  404. remove_type_form.hid_remove_qualification_type_id.data = 0
  405. # re-populate in case changes have occurred
  406. l_qualification_types, form.sel_qualification_type.choices = qualification_type_choices()
  407. s_base = "base_bs.html" if Config.BOOTSTRAP else "base.html"
  408. s_url = url_for("main.qualifications")#, _external=True)
  409. return render_template("qualifications.html", js=True, base=s_base, url=s_url, type_form=type_form, form=form, qualification_types=l_qualification_types, remove_type_form=remove_type_form, type_filter=i_type_filter, remove_form=remove_form)
  410. except Exception as exc:
  411. return render_template_string(end_user_debugging("qualifications"))
  412. # end of surrounding try-except
  413. # end of qualifications view function for route /qualifications
  414. @bp.route("/qualifications_list/<int:i_type_id>/", methods=["GET"])
  415. @login_required
  416. def qualifications_list(i_type_id):
  417. """return list of qualification entries that match type filter"""
  418. try:
  419. l_out = []
  420. q_qualifications = db.session.query(tbl_qualifications.id, tbl_qualifications.v_qualification_name, tbl_qualifications.v_description, tbl_qualification_types.v_qualification_type).join(tbl_qualification_types, tbl_qualifications.i_qualification_type==tbl_qualification_types.id, isouter=True)
  421. q_qualifications = q_qualifications.filter(tbl_qualifications.i_qualification_type==i_type_id) if i_type_id>0 else q_qualifications
  422. if q_qualifications.count()>0:
  423. q_qualifications = q_qualifications.order_by(tbl_qualifications.v_qualification_name)
  424. l_out = list(map(dict_row, q_qualifications.all())) if q_qualifications.count()>0 else []
  425. # end of checking result count of query
  426. return make_response(jsonify(l_out), 200)
  427. except Exception as exc:
  428. return render_template_string(end_user_debugging("qualifications_list"))
  429. # end of surrounding try-except
  430. # end of qualifications_list view function
  431. @bp.route("/qualification_details/<int:i_qualification_id>/", methods=["GET"])
  432. @login_required
  433. def qualification_details(i_qualification_id):
  434. """retrieve qualification details"""
  435. try:
  436. d_out = {}
  437. o_qualification = db.session.get(tbl_qualifications, i_qualification_id)
  438. if o_qualification is not None:
  439. d_out = dict_instance(o_qualification)
  440. # end of checking if record retrieved
  441. return make_response(jsonify(d_out), 200)
  442. except Exception as exc:
  443. return render_template_string(end_user_debugging("qualification_details"))
  444. # end of surrounding try-except
  445. # end of qualification_details view function
  446. def role_department_choices():
  447. """return list of possible|existing role departments"""
  448. l_choices = []
  449. l_departments = list(map(dict_row, db.session.query(tbl_role_departments.id, tbl_role_departments.v_department_name).order_by(tbl_role_departments.v_department_name).all()))
  450. l_department_choices = list(map((lambda x : (x["id"], x["v_department_name"])), l_departments))
  451. if len(l_department_choices)>0:
  452. l_choices = l_department_choices
  453. # end of checking if there were records
  454. return (l_departments, l_choices)
  455. # end of role_department_choices function
  456. @bp.route("/roles/", methods=["GET", "POST"])
  457. @login_required
  458. @check_admin
  459. def roles():
  460. """manage roles available for assignment to records"""
  461. try:
  462. i_department_filter = 0
  463. department_form = RoleDepartmentForm()
  464. form = RoleForm()
  465. # need to populate this first to allow form validation
  466. l_departments, form.sel_department.choices = role_department_choices()
  467. remove_department_form = RoleDepartmentRemovalForm()
  468. remove_form = RoleRemovalForm()
  469. if department_form.validate_on_submit() and request.form.get("btn_save_department", "")=="Save":
  470. i_department_id, s_department_name = (int(department_form.hid_department_id.data), department_form.txt_department_name.data)
  471. if i_department_id > 0:
  472. o_department = db.session.get(tbl_role_departments, i_department_id)
  473. if o_department is not None:
  474. o_department.v_department_name = s_department_name
  475. db.session.add(o_department)
  476. db.session.commit()
  477. flash("Department updated", "success")
  478. # end of checking if record existed
  479. else:
  480. if db.session.query(tbl_role_departments.id).filter(tbl_role_departments.v_department_name==s_department_name).count()<1:
  481. db.session.add(tbl_role_departments(v_department_name=s_department_name))
  482. db.session.commit()
  483. flash("Department inserted", "success")
  484. # end of making sure new department is not a duplicate
  485. # end of checking if new or existing department record
  486. elif form.validate_on_submit() and request.form.get("btn_save_role", "")=="Save":
  487. i_role_id, i_department_id, s_role_name, s_description = (int(form.hid_role_id.data), int(form.sel_department.data), str(form.txt_role_name.data), str(form.txt_description.data))
  488. if i_role_id>0:
  489. o_role = db.session.get(tbl_roles, i_role_id)
  490. if o_role is not None:
  491. if db.session.query(tbl_roles).filter(tbl_roles.id!=i_role_id, tbl_roles.v_role_name==s_role_name, tbl_roles.i_department_id==i_department_id).count()<1:
  492. o_role.i_department_id = i_department_id
  493. o_role.v_role_name = s_role_name
  494. o_role.v_description = s_description
  495. db.session.add(o_role)
  496. db.session.commit()
  497. flash("Role updated")
  498. i_department_filter = i_department_id
  499. else:
  500. flash("There is already a role under same department with same name")
  501. i_department_filter = i_department_id
  502. # end of checking for alternative duplicate
  503. # end of retrieval check
  504. else:
  505. if db.session.query(tbl_roles).filter(tbl_roles.v_role_name==s_role_name, tbl_roles.i_department_id==i_department_id).count()<1:
  506. o_role = tbl_roles(i_department_id=i_department_id, v_role_name=s_role_name, v_description=s_description)
  507. db.session.add(o_role)
  508. db.session.commit()
  509. flash("Role added")
  510. i_department_filter = i_department_id
  511. else:
  512. flash("There is an existing role with same name under same department")
  513. # end of making sure would not count as duplicate
  514. # end of checking if new or existing role record
  515. elif remove_department_form.validate_on_submit():
  516. i_department_id = int(remove_department_form.hid_remove_department_id.data)
  517. o_department = db.session.get(tbl_role_departments, i_department_id)
  518. if o_department is not None:
  519. db.session.delete(o_department)
  520. db.session.commit()
  521. flash("Department removed", "success")
  522. # end of checking if record existed
  523. elif remove_form.validate_on_submit():
  524. i_role_id = int(remove_form.hid_remove_role_id.data)
  525. o_role = db.session.get(tbl_roles, i_role_id)
  526. if o_role is not None:
  527. db.session.delete(o_role)
  528. db.session.commit()
  529. flash("Role removed", "success")
  530. # end of checking if record existed
  531. # end of checking for form submission
  532. department_form.hid_department_id.data, department_form.txt_department_name.data = (0, "")
  533. remove_form.hid_remove_role_id.data, remove_department_form.hid_remove_department_id.data = ("0", "0")
  534. # re-populate in case changes have occurred
  535. l_departments, form.sel_department.choices = role_department_choices()
  536. s_base = "base_bs.html" if Config.BOOTSTRAP else "base.html"
  537. s_url = url_for("main.roles")#, _external=True)
  538. return render_template("roles.html", js=True, base=s_base, url=s_url, departments=l_departments, department_filter=i_department_filter, department_form=department_form, form=form, remove_department_form=remove_department_form, remove_form=remove_form)
  539. except Exception as exc:
  540. return render_template_string(end_user_debugging("roles"))
  541. # end of surrounding try-except
  542. # end of roles view function for route /positions
  543. @bp.route("/roles_list/<int:i_department_id>/", methods=["GET"])
  544. @login_required
  545. def roles_list(i_department_id):
  546. """return list of role entries that match department filter"""
  547. try:
  548. l_out = []
  549. try:
  550. q_roles = db.session.query(tbl_roles.id, tbl_roles.v_role_name, tbl_roles.v_description, tbl_role_departments.v_department_name).join(tbl_role_departments, tbl_roles.i_department_id==tbl_role_departments.id, isouter=True)
  551. q_roles = q_roles.filter(tbl_roles.i_department_id==i_department_id) if i_department_id>0 else q_roles
  552. if q_roles.count()>0:
  553. q_roles = q_roles.order_by(tbl_roles.v_role_name)
  554. l_out = list(map(dict_row, q_roles.all()))
  555. # end of checking result count of query
  556. except Exception as exc:
  557. print(extract_exc("list roles?"))
  558. # end of try-except
  559. return make_response(jsonify(l_out), 200)
  560. except Exception as exc:
  561. return render_template_string(end_user_debugging("roles_list"))
  562. # end of surrounding try-except
  563. # end of roles_list view function for route /roles_list/<int:i_department_id>/
  564. @bp.route("/role_details/<int:i_role_id>/", methods=["GET"])
  565. @login_required
  566. def role_details(i_role_id):
  567. """retrieve role details"""
  568. try:
  569. d_out = {}
  570. o_role = db.session.get(tbl_roles, i_role_id)
  571. if o_role is not None:
  572. d_out = dict_instance(o_role)
  573. # end of checking if record retrieved
  574. return make_response(jsonify(d_out), 200)
  575. except Exception as exc:
  576. return render_template_string(end_user_debugging("role_details"))
  577. # end of surrounding try-except
  578. # end of role_details view function for route /role_details/<int:i_role_id>/
  579. @bp.route("/uploads/<int:i_record>/", methods=["GET", "POST"])
  580. @login_required
  581. @check_capture
  582. def uploads(i_record:int = 0):
  583. """capture uploads linked to CV record"""
  584. try:
  585. # https://blog.miguelgrinberg.com/post/handling-file-uploads-with-flask
  586. o_record = db.session.get(tbl_records, i_record)
  587. if o_record is None:
  588. return redirect(url_for("main.index"))
  589. # end of just making sure valid record
  590. remove_upload_form = UploadRemovalForm()
  591. upload_form = UploadForm()
  592. s_names = o_record.v_surname + ", " + " ".join([o_record.v_name_1, o_record.v_name_2, o_record.v_name_3])
  593. l_qualifications = list(map(dict_row, db.session.query(tbl_record_qualifications.id, tbl_qualifications.v_qualification_name, tbl_record_qualifications.d_acquired).join(tbl_qualifications, tbl_record_qualifications.i_qualification_id==tbl_qualifications.id, isouter=False).filter(tbl_record_qualifications.i_record_id==i_record).order_by(tbl_qualifications.v_qualification_name).all()))
  594. upload_form.sel_match.choices.extend(list(map((lambda x : (x["id"], x["v_qualification_name"])), l_qualifications)))
  595. if upload_form.validate_on_submit() and request.form.get("btn_upload", None) is not None:
  596. i_upload_id, i_upload_type, i_matching_id, s_description = (int(upload_form.hid_upload_id.data), int(upload_form.sel_upload_type.data), int(upload_form.sel_match.data), str(upload_form.txt_description.data))
  597. fil_upload = request.files.get("fil_upload_document")
  598. if fil_upload is not None and fil_upload.filename!="":
  599. s_ext = os.path.splitext(fil_upload.filename)[1]
  600. if s_ext not in [".jpg", ".png", ".pdf", ".doc", ".docx"]:
  601. abort(400)
  602. # end of double-checking file extension
  603. # extract data record values
  604. s_filename= fil_upload.filename
  605. s_mime_type = fil_upload.mimetype
  606. try:
  607. if i_upload_id>0:
  608. q_update = update(tbl_uploads).where(id==i_upload_id)
  609. q_update = q_update.values({"i_record_id": i_record, "i_matching_id": i_matching_id, "si_upload_type": i_upload_type, "v_description": s_description, "v_filename": s_filename, "v_mime_type": s_mime_type, "b_file": fil_upload.read()})
  610. db.session.execute(q_update)
  611. flash("Document record updated")
  612. else:
  613. o_upload = tbl_uploads(i_record_id=i_record, i_matching_id=i_matching_id, si_upload_type=i_upload_type, v_description=s_description, v_filename=s_filename, v_mime_type=s_mime_type, b_file=fil_upload.read())
  614. db.session.add(o_upload)
  615. db.session.commit()
  616. flash("Document uploaded")
  617. except Exception as exc:
  618. print(str(exc))
  619. # end of try-except around saving data record
  620. else:
  621. flash("Invalid file upload")
  622. # end of making sure file was uploaded
  623. elif remove_upload_form.validate_on_submit():
  624. i_upload_id = int(remove_upload_form.hid_remove_upload_id.data)
  625. o_upload = db.session.get(tbl_uploads, i_upload_id)
  626. if o_upload is not None:
  627. db.session.delete(o_upload)
  628. db.session.commit()
  629. flash("Uploaded document removed")
  630. # end of checking if record was retrieved
  631. # end of checking for form submission
  632. remove_upload_form.hid_remove_upload_id.data = 0
  633. upload_form.hid_upload_id.data, upload_form.sel_upload_type.data, upload_form.sel_match.data, upload_form.txt_description.data = (0, 0, 0, "")
  634. l_uploads = list(map(dict_row, db.session.query(tbl_uploads.id, tbl_uploads.si_upload_type, tbl_uploads.v_description, tbl_uploads.v_filename, tbl_qualifications.v_qualification_name).join(tbl_qualifications, tbl_qualifications.id==tbl_uploads.i_matching_id, isouter=True).filter(tbl_uploads.i_record_id==i_record).order_by(tbl_uploads.v_filename).all()))
  635. for ix, d in enumerate(l_uploads): d["si_upload_type"] = {0: "Original CV", 1: "Alteram CV", 2: "Certificate/Diploma/Degree", 3: "I.D. Document or Passport"}[d["si_upload_type"]]; d["v_qualification_name"] = d["v_qualification_name"] if d["v_qualification_name"] is not None else ""
  636. s_base = "base_bs.html" if Config.BOOTSTRAP else "base.html"
  637. s_url = url_for("main.uploads", i_record=i_record)#, _external=True)
  638. return render_template("uploads.html", js=True, base=s_base, url=s_url, record_id=i_record, names=s_names, record_qualifications=l_qualifications, remove_upload_form=remove_upload_form, upload_form=upload_form, document_uploads=l_uploads)
  639. except Exception as exc:
  640. return render_template_string(end_user_debugging("uploads"))
  641. # end of surrounding try-except
  642. # end of uploads view function for route /uploads/<int:i_record>/
  643. @bp.route("/record_details/<int:i_record_id>/", methods=["GET"])
  644. @login_required
  645. def record_details(i_record_id):
  646. """view all details for a record"""
  647. try:
  648. d_out = {}
  649. q_details = db.session.query(tbl_records.id, tbl_records.v_surname, tbl_records.v_name_1, tbl_records.v_name_2, tbl_records.v_name_3, tbl_records.v_id_number, tbl_records.c_gender, tbl_records.si_years_experience, tbl_records.v_sap_k_level, tbl_records.v_contact_number, tbl_records.v_email, tbl_role_departments.v_department_name, tbl_roles.v_role_name).join(tbl_role_departments, tbl_role_departments.id==tbl_records.i_department_id, isouter=True).join(tbl_roles, tbl_roles.id==tbl_records.i_role_id, isouter=True).filter(tbl_records.id==i_record_id)
  650. if q_details.count()>0:
  651. d_out["record"] = dict_row(q_details.first())
  652. d_out["languages"] = list(map(dict_row, db.session.query(tbl_languages.v_language_abbreviation, tbl_languages.v_language_name, tbl_languages.si_level).filter(tbl_languages.i_record_id==i_record_id).order_by(tbl_languages.si_ranking).all()))
  653. for ix, d in enumerate(d_out["languages"]): d_out["languages"][ix]["level"] = ["basic", "intermediate", "proficient"][d["si_level"]]
  654. d_out["qualifications"] = list(map(dict_row, db.session.query(tbl_record_qualifications.d_acquired, tbl_qualifications.v_qualification_name, tbl_qualification_types.v_qualification_type).join(tbl_qualifications, tbl_qualifications.id==tbl_record_qualifications.i_qualification_id, isouter=True).join(tbl_qualification_types, tbl_qualification_types.id==tbl_qualifications.i_qualification_type, isouter=False).filter(tbl_record_qualifications.i_record_id==i_record_id).all()))
  655. for ix, d in enumerate(d_out["qualifications"]):
  656. if isinstance(d["d_acquired"], datetime_class.date): d["d_acquired"] = d["d_acquired"].strftime("%Y-%m-%d")
  657. # end of looping through qualifications
  658. d_out["uploads"] = list(map(dict_row, db.session.query(tbl_uploads.id, tbl_uploads.si_upload_type, tbl_uploads.v_filename, tbl_uploads.v_description, tbl_qualifications.v_qualification_name).join(tbl_qualifications, tbl_qualifications.id==tbl_uploads.i_matching_id, isouter=True).filter(tbl_uploads.i_record_id==i_record_id).all()))
  659. for ix, d in enumerate(d_out["uploads"]): d["si_upload_type"] = {0: "Original CV", 1: "Alteram CV", 2: "Certificate/Diploma/Degree", 3: "I.D. Document or Passport"}[d["si_upload_type"]]
  660. # end of initial query length check
  661. return make_response(jsonify(d_out), 200)
  662. except Exception as exc:
  663. s_logging = end_user_debugging("record_details", False)
  664. return make_response(s_logging, 200)
  665. # end of surrounding try-except
  666. # end of record_details view function for route /record_details/<int:i_record_id>/
  667. @bp.route("/users/", methods=["GET", "POST"])
  668. @login_required
  669. @check_admin
  670. def users():
  671. """manage access|authentication IDs for admin, capture and view"""
  672. try:
  673. form = UserForm()
  674. removal_form = UserRemovalForm()
  675. if form.validate_on_submit() and request.form.get("btn_save", None)=="Save":
  676. i_user_id, s_user_id, s_password, s_password_confirm, bl_admin, bl_capture = (int(form.hid_user_id.data), str(form.txt_user_id.data), str(form.txt_password.data), str(form.txt_password_confirm.data), bool(form.chk_admin.data), bool(form.chk_capture.data))
  677. if i_user_id>0:
  678. o_user = db.session.get(tbl_users, i_user_id)
  679. if o_user is not None:
  680. if db.session.query(tbl_users.id).filter(tbl_users.id!=i_user_id, tbl_users.v_user_id==s_user_id).count()>0:
  681. flash("Another User with same User ID exists")
  682. else:
  683. o_user.v_user_id = s_user_id
  684. o_user.bl_admin = bl_admin
  685. o_user.bl_capture = bl_capture
  686. s_password_change = ""
  687. if len(s_password)>3 and s_password==s_password_confirm:
  688. o_user.set_password(s_password)
  689. s_password_change = " (with password changed)"
  690. # end of checking if password must be changed
  691. db.session.add(o_user)
  692. db.session.commit()
  693. flash(f"User record updated{s_password_change}")
  694. # end of checking for another user record with same user ID
  695. # end of checking for record retrieval
  696. else:
  697. if len(s_password)>3 and s_password==s_password_confirm:
  698. if db.session.query(tbl_users.id).filter(tbl_users.v_user_id==s_user_id).count()>0:
  699. flash("Another User with same User ID already exists")
  700. else:
  701. o_user = tbl_users(v_user_id=s_user_id, v_password=s_password, bl_admin=bl_admin, bl_capture=bl_capture)
  702. db.session.add(o_user)
  703. db.session.commit()
  704. i_user_id = int(o_user.id) if o_user.id is not None else 0
  705. if o_user.id>0:
  706. flash("User record inserted")
  707. # end of checking for record insertion
  708. # end of checking for matching record with same User ID
  709. else:
  710. flash("Password values must be longer than 3 characters, and must match")
  711. # end of making sure passwords match
  712. # end of checking if existing user
  713. elif removal_form.validate_on_submit():
  714. i_user_id = int(removal_form.hid_remove_user_id.data)
  715. o_user = db.session.get(tbl_users, i_user_id)
  716. if o_user is not None:
  717. db.session.delete(o_user)
  718. db.session.commit()
  719. flash("User record removed")
  720. # end of checking for record retrieval
  721. # end of checking for form submission
  722. form.hid_user_id.data, form.txt_user_id.data, form.txt_password.data, form.txt_password_confirm.data, form.chk_admin.data, form.chk_capture.data = (0, "", "", "", False, False)
  723. removal_form.hid_remove_user_id.data = 0
  724. l_users = list(map(dict_row, db.session.query(tbl_users.id, tbl_users.v_user_id, tbl_users.bl_admin, tbl_users.bl_capture).filter(tbl_users.v_user_id!="admin").order_by(tbl_users.v_user_id).all()))
  725. s_base = "base_bs.html" if Config.BOOTSTRAP else "base.html"
  726. s_url = url_for("main.users")#, _external=True)
  727. return render_template("users.html", js=True, base=s_base, url=s_url, users=l_users, form=form, removal_form=removal_form)
  728. except Exception as exc:
  729. return render_template_string(end_user_debugging("users"))
  730. # end of surrounding try-except
  731. # end of users view function for route /users
  732. @bp.route("/user_details/<int:i_user_id>/", methods=["GET"])
  733. @login_required
  734. @check_admin
  735. def user_details(i_user_id:int = 0):
  736. """retrieve user record details"""
  737. try:
  738. d_out = {}
  739. o_user = db.session.get(tbl_users, i_user_id)
  740. if o_user is not None:
  741. d_out = dict_instance(o_user)
  742. del d_out["v_password"]
  743. # end of checking for record retrieval
  744. return make_response(jsonify(d_out), 200)
  745. except Exception as exc:
  746. return render_template_string(end_user_debugging("user_details"))
  747. # end of surrounding try-except
  748. # end of user_details view function for route /user_details/<int:i_user_id>/
  749. @bp.route("/css_sample", methods=["GET"])
  750. def css_sample():
  751. """CSS sample page - purely testing"""
  752. s_base = "base_bs.html" if Config.BOOTSTRAP else "base.html"
  753. s_url = url_for("main.index")#, _external=True)
  754. return render_template("css_sample.html", js=True, base=s_base, url=s_url)
  755. # end of css_sample view function for route /css_sample
  756. s_todo = """
  757. double-check both try-except across the board, along with then logging exceptions
  758. """