from appservices.common.util import * kyc_masters = Blueprint("kyc_masters",__name__) @kyc_masters.route("/add_ekyc",methods=["POST","GET"]) @adminid_access_token_required def add_ekyc(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") if request.method == "POST": name = request.form.get("name","") code = request.form.get("code","") ekycType = request.form.get("ekycType","") if name and code and ekycType: try: ekyc_table = Ekycs( adminId=adminId, name = name, code = code, ekycType = ekycType, createdOn = datetime.datetime.now(), status = 1, ) save_table = ekyc_table.save() flash("Ekyc saved successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Ekyc")) except Exception as e: flash("Unable to save ekyc!!") app.logger.error(traceback.format_exc()) return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Ekyc")) else: flash("Required fields are missing!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Ekyc")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to save ekyc!!" return render_template("super_admin_templates/kyc_masters_list.html",error=error) def fetching_ekyc_details(ekyc_queryset): ekyc_dict = {} try: ekyc_dict={ "id":str(ekyc_queryset.id), "name":ekyc_queryset.name, "code":ekyc_queryset.code, "ekycType":ekyc_queryset.ekycType } if ekyc_queryset.status==1: ekyc_dict["actionText"] = "Active" else: ekyc_dict["actionText"] = "Deactive" if ekyc_queryset.createdOn: ekyc_dict["createdOn"] = ekyc_queryset.createdOn.strftime("%m-%d-%Y") else: ekyc_dict["createdOn"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return ekyc_dict @kyc_masters.route("/kyc_masters_list",methods=["POST","GET"]) @adminid_access_token_required def kyc_masters_list(): if not session.get("adminId"): return redirect("admin_login") ekycList = [] documentsList = [] groupsList = [] configsList = [] attributesList = [] adminId = session.get("adminId") try: redirectTo = request.args.get("redirectTo","Ekyc") if redirectTo: redirectval = redirectTo else: redirectval = "Ekyc" ekyc_search_element = request.form.get("ekyc_search_element","") document_search_element = request.form.get("document_search_element","") group_search_element = request.form.get("group_search_element","") config_search_element = request.form.get("config_search_element","") attribute_search_element = request.form.get("attribute_search_element","") ekycs_queryset = Ekycs.objects(status__in=[0,1]).order_by("-id") if ekyc_search_element: ekycs_queryset = ekycs_queryset.filter(Q(name__icontains=ekyc_search_element)) for each_ekyc in ekycs_queryset: ekyc_dict = fetching_ekyc_details(each_ekyc) ekycList.append(ekyc_dict) ################# For Documents Data ################## documents_queryset = Documents.objects(status__in=[0,1]).order_by("-id") if document_search_element: documents_queryset = Countries.objects(name__icontains=document_search_element,status__in=[0,1]).order_by("-id") for each_document in documents_queryset: document_dict = fetching_document_details(each_document) documentsList.append(document_dict) ################# For KycGroups Data ################## groups_queryset = KycGroups.objects(status__in=[0,1]).order_by("-id") if group_search_element: groups_queryset = KycGroups.objects(name__icontains=group_search_element,status__in=[0,1]).order_by("-id") for each_group in groups_queryset: group_dict = fetching_groups_details(each_group) groupsList.append(group_dict) ################# For KYCGroupConfigs Data ################## configs_queryset = KYCGroupConfigs.objects(status__in=[0,1]).order_by("-id") if config_search_element: configs_queryset = KYCGroupConfigs.objects(name__icontains=config_search_element,status__in=[0,1]).order_by("-id") for each_config in configs_queryset: config_dict = fetching_config_details(each_config) configsList.append(config_dict) ################# For DocumentAttributes Data ################## attributes_queryset = DocumentAttributes.objects(status__in=[0,1]).order_by("-id") if attribute_search_element: attributes_queryset = DocumentAttributes.objects(attributeName__icontains=attribute_search_element,status__in=[0,1]).order_by("-id") for each_attribute in attributes_queryset: attribute_dict = fetching_attributes_details(each_attribute) attributesList.append(attribute_dict) return render_template("super_admin_templates/kyc_masters_list.html", ekycList=ekycList, documentsList=documentsList, groupsList=groupsList, configsList=configsList, attributesList=attributesList, attribute_search_element=attribute_search_element, config_search_element=config_search_element, group_search_element=group_search_element, document_search_element=document_search_element, ekyc_search_element=ekyc_search_element, redirectval=redirectval, ) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to fetch kyc masters details!!" return render_template("super_admin_templates/kyc_masters_list.html", error=error, documentsList=documentsList, groupsList=groupsList, configsList=configsList, ekycList=ekycList, attributesList=attributesList, attribute_search_element=attribute_search_element, document_search_element=document_search_element, group_search_element=group_search_element, config_search_element=config_search_element, ekyc_search_element=ekyc_search_element, redirectval=redirectval, ) @kyc_masters.route("/update_ekyc_status",methods=["POST","GET"]) @adminid_access_token_required def update_ekyc_status(): if not session.get("adminId"): return redirect("admin_login") ekycId = request.args.get("ekycId","") if ekycId: try: ekyc_queryset = Ekycs.objects(id=ekycId,status__nin=[2]).first() if ekyc_queryset: if ekyc_queryset.status == 0: ekyc_queryset.update(status=1) flash("Ekyc activated successfully!") elif ekyc_queryset.status == 1: ekyc_queryset.update(status=0) flash("Ekyc deactivated successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Ekyc")) else: flash("Invaild id!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Ekyc")) except Exception as e: app.logger.error(traceback.format_exc()) return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Ekyc")) else: return redirect(url_for("kyc_masters.kyc_masters_list")) # Update @kyc_masters.route("/update_ekyc",methods=["POST","GET"]) @adminid_access_token_required def update_ekyc(): if not session.get("adminId"): return redirect("admin_login") try: ekycId = request.args.get("ekycId","") if request.method == "POST": name = request.form.get("name","") code = request.form.get("code","") ekycType = request.form.get("ekycType","") ekyc_queryset = Ekycs.objects(id=ekycId).first() if ekyc_queryset: ekyc_queryset.update( name=name, code=code, ekycType=ekycType ) flash("Ekyc updated successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Ekyc")) else: flash("Invaild id!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Ekyc")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to update ekyc!!" return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Ekyc")) # Delete Ekyc @kyc_masters.route("/delete_ekyc",methods=["GET"]) @adminid_access_token_required def delete_ekyc(): try: if not session.get("adminId"): return redirect("admin_login") if request.method == "GET": ekycId = request.args.get("ekycId","") ekyc_queryset = Ekycs.objects(id=ekycId,status__in=[0,1]).first() ekyc_queryset.update(status=2) flash("Ekyc deleted successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Ekyc")) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to delete document!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Ekyc")) def fetching_csv_ekyc_details(ekyc_queryset): ekyc_dict = {} try: if ekyc_queryset.name: ekyc_dict["Ekyc Name"] = ekyc_queryset.name else: ekyc_dict["Ekyc Name"] = "" if ekyc_queryset.code: ekyc_dict["Ekyc Code"] = ekyc_queryset.code else: ekyc_dict["Ekyc Code"] = "" if ekyc_queryset.ekycType: ekyc_dict["Ekyc Type"] = ekyc_queryset.ekycType else: ekyc_dict["Ekyc Type"] = "" if ekyc_queryset.status == 1: ekyc_dict["Status"] = "Active" else: ekyc_dict["Status"] = "In-Active" except Exception as e: app.logger.error(traceback.format_exc()) return ekyc_dict @kyc_masters.route("/ekyc_csv_list",methods=["POST","GET"]) @adminid_access_token_required def ekyc_csv_list(): if not session.get("adminId"): return redirect("admin_login") ekycList = [] adminId = session.get("adminId") try: ekycs_queryset = Ekycs.objects(status__in=[0,1]).order_by("-id") for each_ekyc in ekycs_queryset: ekyc_dict = fetching_csv_ekyc_details(each_ekyc) ekycList.append(ekyc_dict) fieldnames = ['Ekyc Name','Ekyc Code', 'Ekyc Type', 'Status'] temp_csv_file_name = "/media/ekyc_csv_files/"+str(round(time.time() * 1000))+".csv" if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/ekyc_csv_files/")): os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/ekyc_csv_files/")) full_file_name = app.config['SITE_ROOT']+temp_csv_file_name with open(full_file_name, 'w', encoding='UTF8', newline='') as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(ekycList) save_csv_file = CSVFileExports( name=temp_csv_file_name, filters=[], createdOn=datetime.datetime.now(), status=1 ).save() csvFileId = str(save_csv_file.id) return send_file( app.config['SITE_ROOT']+temp_csv_file_name, mimetype='text/csv', download_name='Ekycs list.csv', as_attachment=True ) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to export CSV data!!") return redirect("kyc_masters.kyc_masters_list") ############################ Documents ################################ @kyc_masters.route("/add_document",methods=["POST","GET"]) @adminid_access_token_required def add_document(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") if request.method == "POST": name = request.form.get("name","") priority = request.form.get("priority",1) hasExpiry = request.form.get("hasExpiry",False) isKYC = request.form.get("isKYC",False) isMandatory = request.form.get("isMandatory",False) if name and priority: try: document_table = Documents( adminId=adminId, name = name, priority = priority, hasExpiry = hasExpiry, isMandatory = isMandatory, isKYC = isKYC, createdOn = datetime.datetime.now(), status = 1, ) save_table = document_table.save() flash("Document saved successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Document")) except Exception as e: flash("Unable to save document!!") app.logger.error(traceback.format_exc()) return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Document")) else: flash("Required fields are missing!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Document")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to save document!!" return render_template("super_admin_templates/kyc_masters_list.html",error=error) def fetching_document_details(document_queryset): document_dict = {} try: document_dict={ "id":str(document_queryset.id), "name":document_queryset.name, "priority":document_queryset.priority, "hasExpiry":document_queryset.hasExpiry, "isKYC":document_queryset.isKYC, "isMandatory":document_queryset.isMandatory } if document_queryset.status==1: document_dict["actionText"] = "Active" else: document_dict["actionText"] = "Deactive" if document_queryset.createdOn: document_dict["createdOn"] = document_queryset.createdOn.strftime("%m-%d-%Y") else: document_dict["createdOn"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return document_dict @kyc_masters.route("/update_document_status",methods=["POST","GET"]) @adminid_access_token_required def update_document_status(): if not session.get("adminId"): return redirect("admin_login") documentId = request.args.get("documentId","") if documentId: try: document_queryset = Documents.objects(id=documentId,status__nin=[2]).first() if document_queryset: if document_queryset.status == 0: document_queryset.update(status=1) flash("Document activated successfully!") elif document_queryset.status == 1: document_queryset.update(status=0) flash("Document deactivated successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Document")) else: flash("Invaild id!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Document")) except Exception as e: app.logger.error(traceback.format_exc()) return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Document")) else: return redirect(url_for("kyc_masters.kyc_masters_list")) # Update @kyc_masters.route("/update_document",methods=["POST","GET"]) @adminid_access_token_required def update_document(): if not session.get("adminId"): return redirect("admin_login") try: documentId = request.args.get("documentId","") if request.method == "POST": name = request.form.get("name","") priority = request.form.get("priority","") hasExpiry = request.form.get("hasExpiry") isKYC = request.form.get("isKYC") isMandatory = request.form.get("isMandatory") document_queryset = Documents.objects(id=documentId).first() if document_queryset: document_queryset.update( name=name, priority=priority, hasExpiry=hasExpiry, isMandatory=isMandatory, isKYC=isKYC ) flash("Documents updated successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Document")) else: flash("Invaild id!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Document")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to update document!!" return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Document")) # Delete Document @kyc_masters.route("/delete_document",methods=["GET"]) @adminid_access_token_required def delete_document(): try: if not session.get("adminId"): return redirect("admin_login") if request.method == "GET": documentId = request.args.get("documentId","") document_queryset = Documents.objects(id=documentId,status__in=[0,1]).first() document_queryset.update(status=2) flash("Documents deleted successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Document")) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to delete document!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="Document")) def fetching_csv_document_details(document_queryset): document_dict = {} try: if document_queryset.name: document_dict["Document Name"] = document_queryset.name else: document_dict["Document Name"] = "" if document_queryset.priority: document_dict["Priority"] = document_queryset.priority else: document_dict["Priority"] = "" if document_queryset.hasExpiry: document_dict["HasExpiry"] = document_queryset.hasExpiry else: document_dict["HasExpiry"] = False if document_queryset.isKYC: document_dict["IsKYC"] = document_queryset.isKYC else: document_dict["IsKYC"] = False if document_queryset.isMandatory: document_dict["IsMandatory"] = document_queryset.isMandatory else: document_dict["IsMandatory"] = False if document_queryset.status == 1: document_dict["Status"] = "Active" else: document_dict["Status"] = "In-Active" except Exception as e: app.logger.error(traceback.format_exc()) return document_dict @kyc_masters.route("/document_csv_list",methods=["POST","GET"]) @adminid_access_token_required def document_csv_list(): if not session.get("adminId"): return redirect("admin_login") documentsList = [] adminId = session.get("adminId") try: documents_queryset = Documents.objects(status__in=[0,1]).order_by("-id") for each_document in documents_queryset: document_dict = fetching_csv_document_details(each_document) documentsList.append(document_dict) fieldnames = ['Document Name','Priority', 'HasExpiry', 'IsKYC','IsMandatory' 'Status'] temp_csv_file_name = "/media/document_csv_files/"+str(round(time.time() * 1000))+".csv" if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/document_csv_files/")): os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/document_csv_files/")) full_file_name = app.config['SITE_ROOT']+temp_csv_file_name with open(full_file_name, 'w', encoding='UTF8', newline='') as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(documentsList) save_csv_file = CSVFileExports( name=temp_csv_file_name, filters=[], createdOn=datetime.datetime.now(), status=1 ).save() csvFileId = str(save_csv_file.id) return send_file( app.config['SITE_ROOT']+temp_csv_file_name, mimetype='text/csv', download_name='Documents list.csv', as_attachment=True ) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to export CSV data!!") return redirect("kyc_masters.kyc_masters_list") ############################ KYC Groups ############################## @kyc_masters.route("/add_kyc_group",methods=["POST","GET"]) @adminid_access_token_required def add_kyc_group(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") if request.method == "POST": name = request.form.get("name","") code = request.form.get("code","") description = request.form.get("description","") priority = request.form.get("priority") if name and code and description and priority: try: group_table = KycGroups( adminId=adminId, name = name, code = code, description = description, priority = priority, createdOn = datetime.datetime.now(), status = 1, ) save_table = group_table.save() flash("Ekyc group saved successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroup")) except Exception as e: flash("Unable to save ekyc group!!") app.logger.error(traceback.format_exc()) return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroup")) else: flash("Required fields are missing!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroup")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to save ekyc group!!" return render_template("super_admin_templates/kyc_masters_list.html",error=error,redirectTo="KycGroup") def fetching_groups_details(group_queryset): group_dict = {} try: group_dict={ "id":str(group_queryset.id), "name":group_queryset.name, "code":group_queryset.code, "description":group_queryset.description, "priority":group_queryset.priority } if group_queryset.status==1: group_dict["actionText"] = "Active" else: group_dict["actionText"] = "Deactive" if group_queryset.createdOn: group_dict["createdOn"] = group_queryset.createdOn.strftime("%m-%d-%Y") else: group_dict["createdOn"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return group_dict @kyc_masters.route("/update_group_status",methods=["POST","GET"]) @adminid_access_token_required def update_group_status(): if not session.get("adminId"): return redirect("admin_login") groupId = request.args.get("groupId","") if groupId: try: group_queryset = KycGroups.objects(id=groupId,status__nin=[2]).first() if group_queryset: if group_queryset.status == 0: group_queryset.update(status=1) flash("Ekyc group activated successfully!") elif group_queryset.status == 1: group_queryset.update(status=0) flash("Ekyc group deactivated successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroup")) else: flash("Invaild id!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroup")) except Exception as e: app.logger.error(traceback.format_exc()) return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroup")) else: return redirect(url_for("kyc_masters.kyc_masters_list")) @kyc_masters.route("/update_kyc_group",methods=["POST","GET"]) @adminid_access_token_required def update_kyc_group(): if not session.get("adminId"): return redirect("admin_login") try: groupId = request.args.get("groupId","") if request.method == "POST": name = request.form.get("name","") code = request.form.get("code","") description = request.form.get("description","") priority = request.form.get("priority") group_queryset = KycGroups.objects(id=groupId).first() if group_queryset: group_queryset.update( name=name, code=code, priority=priority, description=description ) flash("Ekyc group updated successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroup")) else: flash("Invaild id!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroup")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to update ekyc group!!" return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroup")) @kyc_masters.route("/delete_kyc_group",methods=["GET"]) @adminid_access_token_required def delete_kyc_group(): try: if not session.get("adminId"): return redirect("admin_login") if request.method == "GET": groupId = request.args.get("groupId","") group_queryset = KycGroups.objects(id=groupId,status__in=[0,1]).first() group_queryset.update(status=2) flash("Ekyc group deleted successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroup")) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to delete state!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroup")) def fetching_csv_kyc_group_details(group_queryset): group_dict = {} try: if group_queryset.name: group_dict["Kyc group Name"] = group_queryset.name else: group_dict["Kyc group Name"] = "" if group_queryset.code: group_dict["Code"] = group_queryset.code else: group_dict["Code"] = "" if group_queryset.description: group_dict["Description"] = group_queryset.description else: group_dict["Description"] = "" if group_queryset.priority: group_dict["Priority"] = group_queryset.priority else: group_dict["Priority"] = 0 if group_queryset.status == 1: group_dict["Status"] = "Active" else: group_dict["Status"] = "In-Active" except Exception as e: app.logger.error(traceback.format_exc()) return group_dict @kyc_masters.route("/groups_csv_list",methods=["POST","GET"]) @adminid_access_token_required def groups_csv_list(): if not session.get("adminId"): return redirect("admin_login") groupsList = [] adminId = session.get("adminId") try: groups_queryset = KycGroups.objects(status__in=[0,1]).order_by("-id") for each_group in groups_queryset: group_dict = fetching_csv_kyc_group_details(each_group) groupsList.append(group_dict) fieldnames = ['Kyc group Name', 'Code','Description','Priority','Status'] temp_csv_file_name = "/media/kyc_group_csv_files/"+str(round(time.time() * 1000))+".csv" if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/kyc_group_csv_files/")): os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/kyc_group_csv_files/")) full_file_name = app.config['SITE_ROOT']+temp_csv_file_name with open(full_file_name, 'w', encoding='UTF8', newline='') as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(statesList) save_csv_file = CSVFileExports( name=temp_csv_file_name, filters=[], createdOn=datetime.datetime.now(), status=1 ).save() csvFileId = str(save_csv_file.id) return send_file( app.config['SITE_ROOT']+temp_csv_file_name, mimetype='text/csv', download_name='KycGroups list.csv', as_attachment=True ) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to export CSV data!!") return redirect("kyc_masters.kyc_masters_list") ############################ KYCGroupConfigs ############################## @kyc_masters.route("/add_kyc_group_config",methods=["POST","GET"]) @adminid_access_token_required def add_kyc_group_config(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") if request.method == "POST": documentId = request.form.get("documentId","") groupId = request.form.get("groupId","") if documentId and groupId: try: config_table = KYCGroupConfigs( adminId=adminId, documentId = documentId, groupId = groupId, createdOn = datetime.datetime.now(), status = 1, ) save_table = config_table.save() flash("Config saved successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroupConfig")) except Exception as e: flash("Unable to save config!!") app.logger.error(traceback.format_exc()) return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroupConfig")) else: flash("Required fields are missing!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroupConfig")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to save config!!" return render_template("super_admin_templates/kyc_masters_list.html",error=error,redirectTo="KycGroupConfig") def fetching_config_details(config_queryset): config_dict = {} try: config_dict={ "id":str(config_queryset.id), "documentId":str(config_queryset.documentId.id), "documentName":config_queryset.documentId.name, "groupName":config_queryset.groupId.name, "groupId":str(config_queryset.groupId.id) } if config_queryset.status==1: config_dict["actionText"] = "Active" else: config_dict["actionText"] = "Deactive" if config_queryset.createdOn: config_dict["createdOn"] = config_queryset.createdOn.strftime("%m-%d-%Y") else: config_dict["createdOn"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return config_dict # Update config type status @kyc_masters.route("/update_config_status",methods=["POST","GET"]) @adminid_access_token_required def update_config_status(): if not session.get("adminId"): return redirect("admin_login") configId = request.args.get("configId","") if configId: try: config_queryset = KYCGroupConfigs.objects(id=configId,status__nin=[2]).first() if config_queryset: if config_queryset.status == 0: config_queryset.update(status=1) flash("Config activated successfully!") elif config_queryset.status == 1: config_queryset.update(status=0) flash("Config deactivated successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroupConfig")) else: flash("Invaild id!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroupConfig")) except Exception as e: app.logger.error(traceback.format_exc()) return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroupConfig")) else: return redirect(url_for("kyc_masters.kyc_masters_list")) # Update config @kyc_masters.route("/update_config",methods=["POST","GET"]) @adminid_access_token_required def update_config(): if not session.get("adminId"): return redirect("admin_login") try: configId = request.args.get("configId","") if request.method == "POST": documentId = request.form.get("documentId","") groupId = request.form.get("groupId","") config_queryset = KYCGroupConfigs.objects(id=configId).first() if config_queryset: config_queryset.update( documentId=ObjectId(documentId), groupId=ObjectId(groupId) ) flash("Config updated successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroupConfig")) else: flash("Invaild id!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroupConfig")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to update config!!" return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroupConfig")) # Delete config @kyc_masters.route("/delete_config",methods=["GET"]) @adminid_access_token_required def delete_config(): try: if not session.get("adminId"): return redirect("admin_login") if request.method == "GET": configId = request.args.get("configId","") config_queryset = KYCGroupConfigs.objects(id=configId,status__in=[0,1]).first() config_queryset.update(status=2) flash("Config deleted successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroupConfig")) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to delete config!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="KycGroupConfig")) def fetching_csv_config_details(config_queryset): config_dict = {} try: if config_queryset.documentId: config_dict["Document Name"] = config_queryset.documentId.name else: config_dict["Document Name"] = "" if config_queryset.groupId: config_dict["Group Name"] = config_queryset.groupId.name else: config_dict["Group Name"] = "" if config_queryset.status == 1: config_dict["Status"] = "Active" else: config_dict["Status"] = "In-Active" except Exception as e: app.logger.error(traceback.format_exc()) return config_dict @kyc_masters.route("/config_csv_list",methods=["POST","GET"]) @adminid_access_token_required def config_csv_list(): if not session.get("adminId"): return redirect("admin_login") configsList = [] adminId = session.get("adminId") try: configs_queryset = KYCGroupConfigs.objects(status__in=[0,1]).order_by("-id") for each_config in configs_queryset: config_dict = fetching_csv_config_details(each_config) configsList.append(config_dict) fieldnames = ['Document Name', 'Group Name','Status'] temp_csv_file_name = "/media/configs_csv_files/"+str(round(time.time() * 1000))+".csv" if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/configs_csv_files/")): os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/configs_csv_files/")) full_file_name = app.config['SITE_ROOT']+temp_csv_file_name with open(full_file_name, 'w', encoding='UTF8', newline='') as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(configsList) save_csv_file = CSVFileExports( name=temp_csv_file_name, filters=[], createdOn=datetime.datetime.now(), status=1 ).save() csvFileId = str(save_csv_file.id) return send_file( app.config['SITE_ROOT']+temp_csv_file_name, mimetype='text/csv', download_name='KYCGroupConfigs list.csv', as_attachment=True ) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to export CSV data!!") return redirect("kyc_masters.kyc_masters_list") ############################ DocumentAttributes ############################## @kyc_masters.route("/add_document_attribute",methods=["POST","GET"]) @adminid_access_token_required def add_document_attribute(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") if request.method == "POST": documentId = request.form.get("documentId","") attributeName = request.form.get("attributeName","") fieldType = request.form.get("fieldType","") supportedFileType = request.form.get("supportedFileType","") regex = request.form.get("regex","") mobileRegex = request.form.get("mobileRegex","") maxSize = request.form.get("maxSize","") minWidth = request.form.get("minWidth","") maxWidth = request.form.get("maxWidth","") label = request.form.get("label","") validationMessage = request.form.get("validationMessage","") priority = request.form.get("priority",1) isKYC = request.form.get("isKYC",False) ekycId = request.form.get("ekycId","") if documentId and attributeName and fieldType and supportedFileType: try: attribute_table = DocumentAttributes( adminId=adminId, documentId=documentId, attributeName = attributeName, fieldType = fieldType, regex = regex, mobileRegex = mobileRegex, supportedFileType = supportedFileType, maxSize = maxSize, minWidth = minWidth, maxWidth = maxWidth, label = label, validationMessage = validationMessage, priority = priority, isKYC = isKYC, createdOn = datetime.datetime.now(), status = 1, ) save_table = attribute_table.save() if ekycId: save_table.update(ekycId=ObjectId(ekycId)) else: save_table.update(ekycId=None) flash("Attribute saved successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="DocumentAttribute")) except Exception as e: flash("Unable to save attribute!!") app.logger.error(traceback.format_exc()) return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="DocumentAttribute")) else: flash("Required fields are missing!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="DocumentAttribute")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to save attribute!!" return render_template("super_admin_templates/kyc_masters_list.html",error=error) def fetching_attributes_details(attribute_queryset): attribute_dict = {} try: attribute_dict={ "id":str(attribute_queryset.id), "documentName":attribute_queryset.documentId.name, "attributeName":attribute_queryset.attributeName, "fieldType":attribute_queryset.fieldType, "supportedFileType":attribute_queryset.supportedFileType, "regex":attribute_queryset.regex, # "maxSize":attribute_queryset.maxSize, # "minWidth":attribute_queryset.minWidth, # "maxWidth":attribute_queryset.maxWidth, # "mobileRegex":attribute_queryset.mobileRegex, "webRegex":"-", "label":attribute_queryset.label, "validationMessage":attribute_queryset.validationMessage, "priority":attribute_queryset.priority, "isKYC":attribute_queryset.isKYC, } if attribute_queryset.ekycId: attribute_dict["ekycName"] = attribute_queryset.ekycId.name else: attribute_dict["ekycName"] = "-" if attribute_queryset.mobileRegex: attribute_dict["mobileRegex"] = attribute_queryset.mobileRegex else: attribute_dict["mobileRegex"] = "-" if attribute_queryset.maxSize: attribute_dict["maxSize"] = attribute_queryset.maxSize else: attribute_dict["maxSize"] = "-" if attribute_queryset.minWidth: attribute_dict["minWidth"] = attribute_queryset.minWidth else: attribute_dict["minWidth"] = "-" if attribute_queryset.maxWidth: attribute_dict["maxWidth"] = attribute_queryset.maxWidth else: attribute_dict["maxWidth"] = "-" if attribute_queryset.status==1: attribute_dict["actionText"] = "Active" else: attribute_dict["actionText"] = "Deactive" if attribute_queryset.createdOn: attribute_dict["createdOn"] = attribute_queryset.createdOn.strftime("%m-%d-%Y") else: attribute_dict["createdOn"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return attribute_dict # Update attribute status @kyc_masters.route("/update_attribute_status",methods=["POST","GET"]) @adminid_access_token_required def update_attribute_status(): if not session.get("adminId"): return redirect("admin_login") attributeId = request.args.get("attributeId","") if attributeId: try: attribute_queryset = DocumentAttributes.objects(id=attributeId,status__nin=[2]).first() if attribute_queryset: if attribute_queryset.status == 0: attribute_queryset.update(status=1) flash("Attribute activated successfully!") elif attribute_queryset.status == 1: attribute_queryset.update(status=0) flash("Attribute deactivated successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="DocumentAttribute")) else: flash("Invaild id!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="DocumentAttribute")) except Exception as e: app.logger.error(traceback.format_exc()) return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="DocumentAttribute")) else: return redirect(url_for("kyc_masters.kyc_masters_list")) # Update @kyc_masters.route("/update_attribute",methods=["POST","GET"]) @adminid_access_token_required def update_attribute(): if not session.get("adminId"): return redirect("admin_login") try: attributeId = request.args.get("attributeId","") if request.method == "POST": attributeId = request.form.get("attributeId","") documentId = request.form.get("documentId","") attributeName = request.form.get("attributeName","") fieldType = request.form.get("fieldType","") supportedFileType = request.form.get("supportedFileType","") regex = request.form.get("regex","") mobileRegex = request.form.get("mobileRegex","") maxSize = request.form.get("maxSize","") minWidth = request.form.get("minWidth","") maxWidth = request.form.get("maxWidth","") label = request.form.get("label","") validationMessage = request.form.get("validationMessage","") priority = request.form.get("priority") isKYC = request.form.get("isKYC") ekycId = request.form.get("ekycId") attribute_queryset = DocumentAttributes.objects(id=attributeId).first() if attribute_queryset: attribute_queryset.update( documentId=ObjectId(documentId), attributeName = attributeName, fieldType = fieldType, regex = regex, mobileRegex = mobileRegex, supportedFileType = supportedFileType, maxSize = maxSize, minWidth = minWidth, maxWidth = maxWidth, label = label, validationMessage = validationMessage, priority = priority, isKYC = isKYC ) if ekycId: attribute_queryset.update(ekycId=ObjectId(ekycId)) flash("Attribute updated successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="DocumentAttribute")) else: flash("Invaild id!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="DocumentAttribute")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to update attribute!!" return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="DocumentAttribute")) # Delete attribute @kyc_masters.route("/delete_attribute",methods=["GET"]) @adminid_access_token_required def delete_attribute(): try: if not session.get("adminId"): return redirect("admin_login") if request.method == "GET": attributeId = request.args.get("attributeId","") attribute_queryset = DocumentAttributes.objects(id=attributeId,status__in=[0,1]).first() attribute_queryset.update(status=2) flash("Attribute deleted successfully!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="DocumentAttribute")) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to delete attribute!!") return redirect(url_for("kyc_masters.kyc_masters_list",redirectTo="DocumentAttribute")) def fetching_csv_attribute_details(attribute_queryset): attribute_dict = {} try: if attribute_queryset.attributeName: attribute_dict["Attribute Name"] = attribute_queryset.attributeName else: attribute_dict["Attribute Name"] = "" if attribute_queryset.documentId: attribute_dict["Document Name"] = attribute_queryset.documentId.name else: attribute_dict["Document Name"] = "" if attribute_queryset.fieldType: attribute_dict["FieldType"] = attribute_queryset.fieldType else: attribute_dict["FieldType"] = "" if attribute_queryset.regex: attribute_dict["Regex"] = attribute_queryset.regex else: attribute_dict["Regex"] = "" if attribute_queryset.maxSize: attribute_dict["MaxSize"] = attribute_queryset.maxSize else: attribute_dict["MaxSize"] = "" if attribute_queryset.minWidth: attribute_dict["MinWidth"] = attribute_queryset.minWidth else: attribute_dict["MinWidth"] = "" if attribute_queryset.maxWidth: attribute_dict["MaxWidth"] = attribute_queryset.maxWidth else: attribute_dict["MaxWidth"] = "" if attribute_queryset.status == 1: attribute_dict["Status"] = "Active" else: attribute_dict["Status"] = "In-Active" except Exception as e: app.logger.error(traceback.format_exc()) return attribute_dict @kyc_masters.route("/attributes_csv_list",methods=["POST","GET"]) @adminid_access_token_required def attributes_csv_list(): if not session.get("adminId"): return redirect("admin_login") attributesList = [] adminId = session.get("adminId") try: attributes_queryset = DocumentAttributes.objects(status__in=[0,1]).order_by("-id") for each_attribute in attributes_queryset: attribute_dict = fetching_csv_attribute_details(each_attribute) attributesList.append(attribute_dict) fieldnames = ['Attribute Name', 'Document Name','FieldType','Regex','MaxSize','MinWidth','MaxWidth','Status'] temp_csv_file_name = "/media/attributes_csv_files/"+str(round(time.time() * 1000))+".csv" if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/attributes_csv_files/")): os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/attributes_csv_files/")) full_file_name = app.config['SITE_ROOT']+temp_csv_file_name with open(full_file_name, 'w', encoding='UTF8', newline='') as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(attributesList) save_csv_file = CSVFileExports( name=temp_csv_file_name, filters=[], createdOn=datetime.datetime.now(), status=1 ).save() csvFileId = str(save_csv_file.id) return send_file( app.config['SITE_ROOT']+temp_csv_file_name, mimetype='text/csv', download_name='DocumentAttributes list.csv', as_attachment=True ) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to export CSV data!!") return redirect("kyc_masters.kyc_masters_list")