from appservices.common.util import * from appservices.common.form_schemas import * user_configurations = Blueprint("user_configurations",__name__) # Add region @user_configurations.route("/add_user_type",methods=["POST","GET"]) @adminid_access_token_required @csrf_protect def add_user_type(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") csrf_token = request.form.get("csrf_token") print(csrf_token,"((((((((((((((csrf_token))))))))))))))") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get('User-Agent').split(";") browser = loginBrowseData[0] client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate=datetime.datetime.now() permissionsList = check_permissions(adminId,"userPermissionsList") if "add" in permissionsList: if request.method == "POST": action = request.form.get("action", "").strip() data = request.form.to_dict() # Handle OTP Generation # if action == "generate": # mail_type = data.get("mailType", "").strip() # print(mail_type,"((((((((((((((((mail_type))))))))))))))))") # if not mail_type: # return jsonify({"responseStatus": 0, "result": "mailType is required!"}), 400 # return jsonify(generate_otp_helper(mail_type)) # # Handle OTP Verification # elif action == "verify": # otp_check_id = request.form.get("otpCheckId","") # otp_code = request.form.get("otpCode","") # print(otp_check_id,"((((((((OTP CHECK ID????????????))))))))") # print(otp_code,"((((((((OTP CODE????????????))))))))") # if not otp_check_id or not otp_code: # return jsonify({"responseStatus": 0, "result": "Required fields are missing!"}) # return jsonify(verify_otp_helper(otp_check_id, otp_code)) # Ensure OTP is Verified Before Updating Service Charges # elif action == "update": # otp_check_id = request.form.get("otpCheckId", "") # otpcheck_queryset = OtpChecks.objects(adminId=adminId, id=str(otp_check_id), status=1).first() # if not otpcheck_queryset: # return jsonify({"responseStatus": 0, "result": "Invalid Request."}) # Update OTP status to 2 after verification # otpcheck_queryset.update(status=2) # return jsonify({"responseStatus": 1, "result": "OTP status successfully updated!"}) userType = request.form.get("userType","") code = request.form.get("code","") description = request.form.get("description","") userPermissionId = request.form.get("userPermissionId","") rank = request.form.get("rank") isUser = request.form.get("isUser",False) isAllow = request.form.get("isAllow",False) isCommission = request.form.get("isCommission",False) isSuperDistributor = request.form.get("isSuperDistributor",False) jsonData = request.form.to_dict(flat=True) requestData = [jsonData] updatedrequestData = [jsonData] if userType and rank and code and description: try: admin_queryset = SuperAdmin.objects(id=adminId,status=1).first() if admin_queryset: message=admin_queryset.userName+" "+userType+" User type created successfully!" save_admin_log_table = save_admin_logs_data(adminId,None,None,"add_user_type","create",actionDate,client_ip,browser,message,requestData,updatedrequestData) user_configuration_table = UserTypes( adminId=adminId, userType = userType, code = code, description = description, userPermissionId = userPermissionId, rank = rank, isUser = isUser, isAllow = isAllow, isCommission = isCommission, isSuperDistributor = isSuperDistributor, createdOn = datetime.datetime.now(), status = 1, ) save_table = user_configuration_table.save() userTypeId = str(save_table.id) flash("User type saved successfully!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserType")) except Exception as e: flash("Unable to save user type!!") app.logger.error(traceback.format_exc()) return redirect(url_for("user_configurations.user_types_list",redirectTo="UserType")) else: flash("Required fields are missing!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserType")) else: flash("Staff member does not have given create user type permissions!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserType")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to save user type!!" return render_template("super_admin_templates/user_configuration_list.html",error=error) @user_configurations.route("/user_types_list",methods=["POST","GET"]) @adminid_access_token_required def user_types_list(): if not session.get("adminId"): return redirect("admin_login") userTypesList = [] userEntityList = [] userNameFormatList = [] userTypeOperationList = [] operationList = [] serviceGroupingsList =[] userPermissionsList =[] userpermissionpagination="" pagination="" entityPagination="" form = SearchUserTypesList(request.args) adminId = session.get("adminId") permissionsList = check_permissions(adminId,"userPermissionsList") if "view" in permissionsList: try: redirectTo = request.args.get("redirectTo","UserType") if redirectTo: redirectval = redirectTo else: redirectval = "UserType" user_type_search_element = request.args.get("user_type_search_element","") user_entity_search_element = request.args.get("user_entity_search_element","") user_format_search_element = request.args.get("user_format_search_element","") user_operation_search_element = request.args.get("user_operation_search_element","") search_element = request.args.get("search_element","") ######################## User Types List Start ################################## if form.validate(): user_types_queryset = UserTypes.objects(status__in=[0,1]).order_by("-id") if user_type_search_element: user_types_queryset = user_types_queryset.filter(Q(userType__icontains=user_type_search_element)) # Get the current page from the query parameters page = request.args.get(get_page_parameter(), type=int, default=1) per_page = 20 # Number of items per page # Query the database for the current page's data total_count = user_types_queryset.count() start = (page - 1) * per_page end = min(start + per_page, total_count) user_types = user_types_queryset[start:end] snoCount = start for each_user_type in user_types: snoCount +=1 userTypeDict = fetching_user_type_details(each_user_type) userTypeDict["snoCount"]=snoCount userTypesList.append(userTypeDict) # Pagination object for rendering pagination controls in the template pagination = Pagination(page=page, total=total_count, per_page=per_page, alignment="right") ############################### User Types List End ################################## ######################## User Entities List Start ################################## user_entities_queryset = UserEntity.objects(status__in=[0,1]).order_by("-id") if user_entity_search_element: user_entities_queryset = user_entities_queryset.filter(Q(entityType__icontains=user_entity_search_element)) # Get the current page from the query parameters entityPage = request.args.get("entityPage", type=int, default=1) # Use a unique query parameter name entity_per_page = 20 # Number of items per page # Query the database for the current page's data total_entity_count = user_entities_queryset.count() startEntity = (entityPage - 1) * entity_per_page endEntity = min(startEntity + entity_per_page, total_entity_count) user_entities = user_entities_queryset[startEntity:endEntity] for each_user_entity in user_entities: userEntityDict = fetching_user_entity_details(each_user_entity) userEntityList.append(userEntityDict) # Pagination object for rendering pagination controls in the template entityPagination = Pagination(page=entityPage, total=total_entity_count, per_page=entity_per_page, alignment="right",page_parameter="entityPage",href="user_types_list?redirectTo=UserEntity&entityPage={0}") ############################### User Entitys List End ################################## user_name_formats_queryset = UserNameFormats.objects(status__in=[0,1]).order_by("-id") if user_format_search_element: user_name_formats_queryset = user_name_formats_queryset.filter(Q(userNameFormat__icontains=user_format_search_element)) for each_user_format in user_name_formats_queryset: userNameFormatDict = fetching_user_name_format_details(each_user_format) userNameFormatList.append(userNameFormatDict) user_type_operations_queryset = UserTypeOperations.objects(status__in=[0,1]).order_by("-id") if user_operation_search_element: user_type_operations_queryset = user_type_operations_queryset for each_user_type_operation in user_type_operations_queryset: userTypeOperationDict = fetching_user_type_operation_details(each_user_type_operation) userTypeOperationList.append(userTypeOperationDict) operations_queryset = Operations.objects(status__in=[0,1]).order_by("-id") for each_operation in operations_queryset: operationDict = fetching_operation_details(each_operation) operationList.append(operationDict) user_permissions_queryset = UserPermissions.objects(status__in=[0,1]).order_by("-id") if search_element: user_permissions_queryset = user_permissions_queryset.filter(Q(permissionName__icontains=search_element)) userpermissionpage = request.args.get(get_page_parameter(), type=int, default=1) per_page = 20 # Number of items per page # Query the database for the current page's data total_count = user_permissions_queryset.count() start = (userpermissionpage - 1) * per_page end = min(start + per_page, total_count) user_types = user_permissions_queryset[start:end] snoCount = start for each_user_permission in user_permissions_queryset: snoCount +=1 user_permission_dict = fetching_user_permissions_details(each_user_permission) user_permission_dict["snoCount"]=snoCount userPermissionsList.append(user_permission_dict) userpermissionpagination = Pagination(userpermissionpage=userpermissionpage, total=total_count, per_page=per_page, alignment="right") service_groupings_queryset = ServiceGrouping.objects(status__in=[1]).all() for each_service_grouping in service_groupings_queryset: serviceGroupingDict = fetching_service_groupings_details(each_service_grouping) serviceGroupingsList.append(serviceGroupingDict) print(serviceGroupingsList,"(((((((((((((((((((serviceGroupingsList serviceGroupingsList)))))))))))))))))))") # user_types_queryset = UserTypes.objects(status__in=[0,1]).order_by("-id") # if user_type_search_element: # user_types_queryset = user_types_queryset.filter(Q(userType__icontains=user_type_search_element)) # # Get the current page from the query parameters # page = request.args.get(get_page_parameter(), type=int, default=1) # per_page = 20 # Number of items per page # # Query the database for the current page's data # total_count = user_types_queryset.count() # start = (page - 1) * per_page # end = min(start + per_page, total_count) # user_types = user_types_queryset[start:end] # snoCount = start # for each_user_type in user_types: # snoCount +=1 # userTypeDict = fetching_user_type_details(each_user_type) # userTypeDict["snoCount"]=snoCount # userTypesList.append(userTypeDict) # # Pagination object for rendering pagination controls in the template # pagination = Pagination(page=page, total=total_count, per_page=per_page, alignment="right") # ############################### User Types List End ################################## # ######################## User Entities List Start ################################## # user_entities_queryset = UserEntity.objects(status__in=[0,1]).order_by("-id") # if user_entity_search_element: # user_entities_queryset = user_entities_queryset.filter(Q(entityType__icontains=user_entity_search_element)) # # Get the current page from the query parameters # entityPage = request.args.get("entityPage", type=int, default=1) # Use a unique query parameter name # entity_per_page = 20 # Number of items per page # # Query the database for the current page's data # total_entity_count = user_entities_queryset.count() # startEntity = (entityPage - 1) * entity_per_page # endEntity = min(startEntity + entity_per_page, total_entity_count) # user_entities = user_entities_queryset[startEntity:endEntity] # for each_user_entity in user_entities: # userEntityDict = fetching_user_entity_details(each_user_entity) # userEntityList.append(userEntityDict) # # Pagination object for rendering pagination controls in the template # entityPagination = Pagination(page=entityPage, # total=total_entity_count, per_page=entity_per_page, # alignment="right",page_parameter="entityPage",href="user_types_list?redirectTo=UserEntity&entityPage={0}") # ############################### User Entitys List End ################################## # user_name_formats_queryset = UserNameFormats.objects(status__in=[0,1]).order_by("-id") # if user_format_search_element: # user_name_formats_queryset = user_name_formats_queryset.filter(Q(userNameFormat__icontains=user_format_search_element)) # for each_user_format in user_name_formats_queryset: # userNameFormatDict = fetching_user_name_format_details(each_user_format) # userNameFormatList.append(userNameFormatDict) # user_type_operations_queryset = UserTypeOperations.objects(status__in=[0,1]).order_by("-id") # if user_operation_search_element: # user_type_operations_queryset = user_type_operations_queryset # for each_user_type_operation in user_type_operations_queryset: # userTypeOperationDict = fetching_user_type_operation_details(each_user_type_operation) # userTypeOperationList.append(userTypeOperationDict) # operations_queryset = Operations.objects(status__in=[0,1]).order_by("-id") # for each_operation in operations_queryset: # operationDict = fetching_operation_details(each_operation) # operationList.append(operationDict) # user_permissions_queryset = UserPermissions.objects(status__in=[0,1]).order_by("-id") # if search_element: # user_permissions_queryset = user_permissions_queryset.filter(Q(permissionName__icontains=search_element)) # userpermissionpage = request.args.get(get_page_parameter(), type=int, default=1) # per_page = 20 # Number of items per page # # Query the database for the current page's data # total_count = user_permissions_queryset.count() # start = (userpermissionpage - 1) * per_page # end = min(start + per_page, total_count) # user_types = user_permissions_queryset[start:end] # snoCount = start # for each_user_permission in user_permissions_queryset: # snoCount +=1 # user_permission_dict = fetching_user_permissions_details(each_user_permission) # user_permission_dict["snoCount"]=snoCount # userPermissionsList.append(user_permission_dict) # userpermissionpagination = Pagination(userpermissionpage=userpermissionpage, total=total_count, per_page=per_page, alignment="right") # service_groupings_queryset = ServiceGrouping.objects(status__in=[1]).all() # for each_service_grouping in service_groupings_queryset: # serviceGroupingDict = fetching_service_groupings_details(each_service_grouping) # serviceGroupingsList.append(serviceGroupingDict) return render_template("super_admin_templates/user_configuration_list.html", operationList=operationList, userTypesList=userTypesList, userEntityList=userEntityList, userNameFormatList=userNameFormatList, userTypeOperationList=userTypeOperationList, user_type_search_element=user_type_search_element, user_operation_search_element=user_operation_search_element, user_format_search_element=user_format_search_element, user_entity_search_element=user_entity_search_element, search_element=search_element, redirectval=redirectval, serviceGroupingsList=serviceGroupingsList, userPermissionsList=userPermissionsList, pagination=pagination, entityPagination=entityPagination, userpermissionpagination=userpermissionpagination, form = form ) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to fetch user types details!!" return render_template("super_admin_templates/user_configuration_list.html", error=error, operationList=operationList, userTypesList=userTypesList, userEntityList=userEntityList, userTypeOperationList=userTypeOperationList, userNameFormatList=userNameFormatList, user_format_search_element=user_format_search_element, user_operation_search_element=user_operation_search_element, user_type_search_element=user_type_search_element, user_entity_search_element=user_entity_search_element, search_element=search_element, redirectval=redirectval, serviceGroupingsList =serviceGroupingsList, userPermissionsList =userPermissionsList, pagination =pagination, entityPagination =entityPagination, userpermissionpagination=userpermissionpagination, form = form ) else: flash("The staff member does not have permission to view User type", "danger") return render_template("super_admin_templates/user_configuration_list.html") # Update regions type status @user_configurations.route("/update_user_type_status",methods=["POST","GET"]) @adminid_access_token_required def update_user_type_status(): if not session.get("adminId"): return redirect("admin_login") adminId=session.get("adminId") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get('User-Agent').split(";") browser = loginBrowseData[0] client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate=datetime.datetime.now() jsonData = request.form.to_dict(flat=True) existing_record = "" updatedrequestData = [jsonData] permissionsList = check_permissions(session.get("adminId"),"userPermissionsList") if "edit" in permissionsList: userTypeId = request.args.get("userTypeId","") if userTypeId: try: user_type_queryset = UserTypes.objects(id=userTypeId,status__nin=[2]).first() existing_record = user_type_queryset.to_json() requestData = [existing_record] if user_type_queryset: if user_type_queryset.status == 0: user_type_queryset.update(status=1) flash("User type activated successfully!") message=user_type_queryset.adminId.userName+" "+user_type_queryset.userType+" User type activated successfully!" elif user_type_queryset.status == 1: user_type_queryset.update(status=0) flash("User type deactivated successfully!") message=user_type_queryset.adminId.userName+" "+user_type_queryset.userType+" User type deactivated successfully!" save_admin_log_table = save_admin_logs_data(adminId,None,None,"update_user_type_status","updatestatus",actionDate,client_ip,browser,message,requestData,updatedrequestData) return redirect(url_for("user_configurations.user_types_list",redirectTo="UserType")) else: flash("Invaild id!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserType")) except Exception as e: app.logger.error(traceback.format_exc()) return redirect(url_for("user_configurations.user_types_list",redirectTo="UserType")) else: return redirect(url_for("user_configurations.user_types_list")) else: flash("Staff member does not have given update user type status permissions!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserType")) # Update region @user_configurations.route("/update_user_type",methods=["POST","GET"]) @adminid_access_token_required @csrf_protect def update_user_type(): print(f"Content length: {request.content_length} bytes") if not session.get("adminId"): return redirect("admin_login") adminId=session.get("adminId") csrf_token = request.form.get("csrf_token") print(csrf_token,"((((((((((((((csrf_token))))))))))))))") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get('User-Agent').split(";") browser = loginBrowseData[0] existing_record = "" client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate=datetime.datetime.now() permissionsList = check_permissions(session.get("adminId"),"userPermissionsList") if "edit" in permissionsList: try: userTypeId = request.args.get("userTypeId","") if request.method == "POST": action = request.form.get("action", "").strip() data = request.form.to_dict() # Handle OTP Generation # if action == "generate": # mail_type = data.get("mailType", "").strip() # print(mail_type,"((((((((((((((((mail_type))))))))))))))))") # if not mail_type: # return jsonify({"responseStatus": 0, "result": "mailType is required!"}), 400 # return jsonify(generate_otp_helper(mail_type)) # # Handle OTP Verification # elif action == "verify": # otp_check_id = request.form.get("otpCheckId","") # otp_code = request.form.get("otpCode","") # print(otp_check_id,"((((((((OTP CHECK ID????????????))))))))") # print(otp_code,"((((((((OTP CODE????????????))))))))") # if not otp_check_id or not otp_code: # return jsonify({"responseStatus": 0, "result": "Required fields are missing!"}) # return jsonify(verify_otp_helper(otp_check_id, otp_code)) # # Ensure OTP is Verified Before Updating Service Charges # elif action == "update": # otp_check_id = request.form.get("otpCheckId", "") # otpcheck_queryset = OtpChecks.objects(adminId=adminId, id=str(otp_check_id), status=1).first() # if not otpcheck_queryset: # return jsonify({"responseStatus": 0, "result": "Invalid Request."}) # # Update OTP status to 2 after verification # otpcheck_queryset.update(status=2) # return jsonify({"responseStatus": 1, "result": "OTP status successfully updated!"}) userType = request.form.get("userType","") code = request.form.get("code","") description = request.form.get("description","") userPermissionId = request.form.get("userPermissionId","") rank = request.form.get("rank") isUser = request.form.get("isUser") isAllow = request.form.get("isAllow") isCommission = request.form.get("isCommission") isSuperDistributor = request.form.get("isSuperDistributor") jsonData = request.form.to_dict(flat=True) user_type_queryset = UserTypes.objects(id=userTypeId).first() existing_record = user_type_queryset.to_json() message=user_type_queryset.adminId.userName+" "+userType+" User type updated successfully!" requestData=[existing_record] updatedrequestData=[jsonData] save_admin_log_table = save_admin_logs_data(adminId,None,None,"update_user_type","update",actionDate,client_ip,browser,message,requestData,updatedrequestData) if user_type_queryset: user_type_queryset.update( userType=userType, userPermissionId = ObjectId(userPermissionId), code = code, description = description, rank = rank ) if isUser == None: user_type_queryset.update(isUser=False) else: user_type_queryset.update(isUser=True) if isAllow == None: user_type_queryset.update(isAllow=False) else: user_type_queryset.update(isAllow=True) if isCommission == None: user_type_queryset.update(isCommission=False) else: user_type_queryset.update(isCommission=True) if isSuperDistributor == None: user_type_queryset.update(isSuperDistributor=False) else: user_type_queryset.update(isSuperDistributor=True) flash("User type updated successfully!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserType")) else: flash("Invaild id!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserType")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to update user type!!" return redirect(url_for("user_configurations.user_types_list",redirectTo="UserType")) else: flash("Staff member does not have given update user type permissions!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserType")) # Delete User type @user_configurations.route("/delete_user_type",methods=["GET"]) @adminid_access_token_required def delete_user_type(): try: existing_record = "" if not session.get("adminId"): return redirect("admin_login") adminId=session.get("adminId") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get("User-Agent").split(";") browser = loginBrowseData[0] client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate=datetime.datetime.now() jsonData = request.form.to_dict(flat=True) requestData = [existing_record] updatedrequestData = [jsonData] permissionsList = check_permissions(session.get("adminId"),"userPermissionsList") if "delete" in permissionsList: if request.method == "GET": admin_queryset = SuperAdmin.objects(id=adminId,status=1).first() userTypeId = request.args.get("userTypeId","") user_type_queryset = UserTypes.objects(id=userTypeId,status__in=[0,1]).first() existing_record = user_type_queryset.to_json() user_type_queryset.update(status=2) flash("User type deleted successfully!") message=user_type_queryset.adminId.userName+" User type deleted successfully!" save_admin_log_table = save_admin_logs_data(adminId,None,None,"delete_user_type","delete",actionDate,client_ip,browser,message,requestData,updatedrequestData) return redirect(url_for("user_configurations.user_types_list",redirectTo="UserType")) else: flash("Staff member does not have given delete user type permissions!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserType")) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to delete user type!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserType")) def fetching_csv_user_details(user_type_queryset): userTypeDict = {} try: if user_type_queryset.userType: userTypeDict["User type Name"] = user_type_queryset.userType else: userTypeDict["User type Name"] = "" if user_type_queryset.rank: userTypeDict["Rank"] = user_type_queryset.rank else: userTypeDict["Rank"] = "" if user_type_queryset.code: userTypeDict["Code"] = user_type_queryset.code else: userTypeDict["Code"] = "" if user_type_queryset.description: userTypeDict["Description"] = user_type_queryset.description else: userTypeDict["Description"] = "" if user_type_queryset.isUser: userTypeDict["Is User"] = user_type_queryset.isUser else: userTypeDict["Is User"] = False if user_type_queryset.isAllow: userTypeDict["Is Allow"] = user_type_queryset.isAllow else: userTypeDict["Is Allow"] = False if user_type_queryset.isCommission: userTypeDict["Is Commission"] = user_type_queryset.isCommission else: userTypeDict["Is Commission"] = False except Exception as e: app.logger.error(traceback.format_exc()) return userTypeDict # View All regions @user_configurations.route("/usertypes_csv_list",methods=["POST","GET"]) @adminid_access_token_required def usertypes_csv_list(): if not session.get("adminId"): return redirect("admin_login") userTypesList = [] adminId = session.get("adminId") try: user_types_queryset = UserTypes.objects(status__in=[0,1]).order_by("-id").all() for each_user_type in user_types_queryset: userTypeDict = fetching_csv_user_details(each_user_type) userTypesList.append(userTypeDict) fieldnames = ['User type Name', 'Rank', 'Code','Description','Is User','Is Allow','Is Commission'] temp_csv_file_name = "/media/user_types_csv_files/"+str(round(time.time() * 1000))+".csv" if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/user_types_csv_files/")): os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/user_types_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(userTypesList) 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='User types list.csv', as_attachment=True ) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to export CSV data!!") return redirect("user_configurations.user_types_list") ######################## Username Format Crud Operations Here ########################### @user_configurations.route("/add_user_name_format",methods=["POST","GET"]) @adminid_access_token_required def add_user_name_format(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") if request.method == "POST": userNameFormat = request.form.get("userNameFormat","") userTypeId = request.form.get("userTypeId","") suffix = request.form.get("suffix","") prefix = request.form.get("prefix","") if userNameFormat and userTypeId: try: user_name_format_table = UserNameFormats( adminId=adminId, userNameFormat = userNameFormat, userTypeId = userTypeId, suffix = suffix, prefix = prefix, createdOn = datetime.datetime.now(), status = 1, ) save_table = user_name_format_table.save() userNameFormatId = str(save_table.id) flash("Username format saved successfully!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UsernameFormat")) except Exception as e: flash("Unable to save username format!!") app.logger.error(traceback.format_exc()) return redirect(url_for("user_configurations.user_types_list",redirectTo="UsernameFormat")) else: flash("Required fields are missing!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UsernameFormat")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to save username format!!" return render_template("super_admin_templates/user_configuration_list.html",error=error) def fetching_user_name_format_details(user_name_format_queryset): userNameFormatDict = {} try: userNameFormatDict={ "id":str(user_name_format_queryset.id), "userTypeId":str(user_name_format_queryset.userTypeId.id), "userTypeName":user_name_format_queryset.userTypeId.userType, "userNameFormat":user_name_format_queryset.userNameFormat, "suffix":user_name_format_queryset.suffix, "prefix":user_name_format_queryset.prefix } if user_name_format_queryset.status==1: userNameFormatDict["actionText"] = "Active" else: userNameFormatDict["actionText"] = "Deactive" if user_name_format_queryset.createdOn: userNameFormatDict["createdOn"] = user_name_format_queryset.createdOn.strftime("%m-%d-%Y") else: userNameFormatDict["createdOn"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return userNameFormatDict @user_configurations.route("/update_user_type_format_status",methods=["POST","GET"]) @adminid_access_token_required def update_user_type_format_status(): if not session.get("adminId"): return redirect("admin_login") userNameFormatId = request.args.get("userNameFormatId","") if userNameFormatId: try: user_name_format_queryset = UserNameFormats.objects(id=userNameFormatId,status__nin=[2]).first() if user_name_format_queryset: if user_name_format_queryset.status == 0: user_name_format_queryset.update(status=1) flash("User name format activated successfully!") elif user_name_format_queryset.status == 1: user_name_format_queryset.update(status=0) flash("User name format deactivated successfully!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UsernameFormat")) else: flash("Invaild id!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UsernameFormat")) except Exception as e: app.logger.error(traceback.format_exc()) return redirect(url_for("user_configurations.user_types_list",redirectTo="UsernameFormat")) else: return redirect(url_for("user_configurations.user_types_list")) @user_configurations.route("/update_user_format",methods=["POST","GET"]) @adminid_access_token_required def update_user_format(): if not session.get("adminId"): return redirect("admin_login") try: userNameFormatId = request.args.get("userNameFormatId","") if request.method == "POST": userNameFormat = request.form.get("userNameFormat","") userTypeId = request.form.get("userTypeId","") prefix = request.form.get("prefix","") suffix = request.form.get("suffix") user_name_format_queryset = UserNameFormats.objects(id=userNameFormatId).first() if user_name_format_queryset: user_name_format_queryset.update( userNameFormat=userNameFormat, prefix = prefix, suffix = suffix ) if userTypeId: user_name_format_queryset.update(userTypeId=ObjectId(userTypeId)) flash("User name format updated successfully!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UsernameFormat")) else: flash("Invaild id!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UsernameFormat")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to update user name format!!" return redirect(url_for("user_configurations.user_types_list",redirectTo="UsernameFormat")) @user_configurations.route("/delete_user_name_format",methods=["GET"]) @adminid_access_token_required def delete_user_name_format(): try: if not session.get("adminId"): return redirect("admin_login") if request.method == "GET": userNameFormatId = request.args.get("userNameFormatId","") user_name_format_queryset = UserNameFormats.objects(id=userNameFormatId,status__in=[0,1]).first() user_name_format_queryset.update(status=2) flash("User name format deleted successfully!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UsernameFormat")) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to delete user name format!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UsernameFormat")) def fetching_csv_user_name_format_details(user_name_format_queryset): userNameFormatDict = {} try: if user_name_format_queryset.userNameFormat: userNameFormatDict["User Name Format"] = user_name_format_queryset.userNameFormat else: userNameFormatDict["User Name Format"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return userNameFormatDict @user_configurations.route("/username_format_csv_list",methods=["POST","GET"]) @adminid_access_token_required def username_format_csv_list(): if not session.get("adminId"): return redirect("admin_login") userNameFormatList = [] adminId = session.get("adminId") try: user_name_formats_queryset = UserNameFormats.objects(status__in=[0,1]).order_by("-id").all() for each_user_format in user_name_formats_queryset: userNameFormatDict = fetching_csv_user_name_format_details(each_user_format) userNameFormatList.append(userNameFormatDict) fieldnames = ['User Name Format'] temp_csv_file_name = "/media/user_name_format_csv_files/"+str(round(time.time() * 1000))+".csv" if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/user_name_format_csv_files/")): os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/user_name_format_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(userNameFormatList) 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='User name formats list.csv', as_attachment=True ) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to export CSV data!!") return redirect("user_configurations.user_types_list") @user_configurations.route("/add_user_entity",methods=["POST","GET"]) @adminid_access_token_required @csrf_protect def add_user_entity(): data_status={"responseStatus":0,"result":""} try: if not session.get("adminId"): flash("Session expired Please login again.","danger") data_status["responseStatus"]=4 return data_status adminId = session.get("adminId") csrf_token = request.form.get("csrf_token") print(csrf_token,"((((((((((((((csrf_token))))))))))))))") latitude = request.form.get("latitude", "") longitude = request.form.get("longitude", "") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get('User-Agent').split(";") browser = loginBrowseData[0] client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate=datetime.datetime.now() permissionsList = check_permissions(adminId,"kycEntityTypesPermissions") if "add" in permissionsList: if request.method == "POST": entityType = request.form.get("entityType","") isIndividual = request.form.get("isIndividual") isDefault = request.form.get("isDefault") jsonData = request.form.to_dict(flat=True) requestData = [jsonData] updatedrequestData = [jsonData] form = KycEntityForm() if form.validate_on_submit(): if entityType: try: isIndividual = True if isIndividual == "True" else False isDefault = True if isDefault == "True" else False admin_queryset = SuperAdmin.objects(id=adminId,status=1).first() if admin_queryset: message=admin_queryset.userName+" "+entityType+" User entity created successfully!" save_admin_log_table = save_admin_logs_data(adminId,None,None,"add_user_entity","create",actionDate,client_ip,browser,message,requestData,updatedrequestData,latitude,longitude) user_entity_table = UserEntity( adminId=adminId, entityType = entityType, isIndividual = isIndividual, isDefault =isDefault, createdOn = datetime.datetime.now(), status = 1, ) save_table = user_entity_table.save() entityTypeId = str(save_table.id) # if isIndividual: # user_entity_table = UserEntity.objects(id__ne=entityTypeId,status__in=[0,1], isIndividual=True).all() # if user_entity_table: # user_entity_table.update(isIndividual=False) if isDefault: user_entity_table = UserEntity.objects(id__ne=entityTypeId,status__in=[0,1], isDefault=True).all() if user_entity_table: user_entity_table.update(isDefault=False) flash("Entity saved successfully","success") data_status['responseStatus']=1 return data_status except Exception as e: flash("Unable to save Entity Details","danger") data_status['responseStatus']=4 return data_status else: data_status['responseStatus']=2 data_status['result']="Required Fields are Missing" return data_status else: data_status['result']=form.errors return data_status else: flash("Invalid Request.","danger") data_status['responseStatus']=4 return data_status else: flash("The staff member does not have permission to create a Entity types.", "danger") data_status['responseStatus']=4 return data_status except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to save entity details!!" flash("Unable to save entity type", "danger") data_status['responseStatus']=4 return data_status def fetching_user_entity_details(user_entity_queryset): userEntityDict = {} try: userEntityDict={ "id":str(user_entity_queryset.id), "entityType":user_entity_queryset.entityType, "isIndividual":user_entity_queryset.isIndividual, "isDefault" : user_entity_queryset.isDefault } if user_entity_queryset.status==1: userEntityDict["actionText"] = "Active" else: userEntityDict["actionText"] = "Deactive" except Exception as e: app.logger.error(traceback.format_exc()) return userEntityDict # Update regions type status @user_configurations.route("/update_user_entity_status",methods=["POST","GET"]) @adminid_access_token_required @csrf_protect def update_user_entity_status(): data_status={"responseStatus":0,"result":""} if not session.get("adminId"): return redirect("admin_login") adminId=session.get("adminId") csrf_token = request.form.get("csrf_token") print(csrf_token,"((((((((((((((csrf_token))))))))))))))") servicePage = request.args.get("servicePage") user_entity_search_element = request.args.get("user_entity_search_element", "") latitude = request.form.get("latitude", "") longitude = request.form.get("longitude", "") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get('User-Agent').split(";") browser = loginBrowseData[0] client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate=datetime.datetime.now() jsonData = request.form.to_dict(flat=True) existing_record = "" updatedrequestData = [jsonData] form=UpdateStatusRemarkForm() if form.validate_on_submit(): permissionsList = check_permissions(session.get("adminId"),"kycEntityTypesPermissions") if "edit" in permissionsList: entityTypeId = request.args.get("entityTypeId","") remark = request.form.get("remark") if entityTypeId: try: user_type_queryset = UserEntity.objects(id=entityTypeId,status__nin=[2]).first() existing_record = user_type_queryset.to_json() requestData = [existing_record] if user_type_queryset: if user_type_queryset.status == 0: user_type_queryset.update(status=1) flash("User entity activated successfully!","success") message=user_type_queryset.adminId.userName+" "+user_type_queryset.entityType+" User entity activated successfully!" elif user_type_queryset.status == 1: user_type_queryset.update(status=0) flash("User entity deactivated successfully!","success") message=user_type_queryset.adminId.userName+" "+user_type_queryset.entityType+" User entity deactivated successfully!" save_admin_log_table = save_admin_logs_data(adminId,None,None,"update_user_entity_status","updatestatus",actionDate,client_ip,browser,message,requestData,updatedrequestData,latitude,longitude) save_remarks_data=save_admin_remarks_data(entityTypeId,adminId,remark,"merchant") # return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity",servicePage=servicePage,user_entity_search_element=user_entity_search_element)) data_status['responseStatus']=1 return data_status else: flash("Invaild id","danger") # return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity",servicePage=servicePage,user_entity_search_element=user_entity_search_element)) data_status['responseStatus']=4 return data_status except Exception as e: flash("unable to update the entity status","danger") app.logger.error(traceback.format_exc()) # return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity",servicePage=servicePage,user_entity_search_element=user_entity_search_element)) data_status['responseStatus']=4 return data_status else: data_status['responseStatus']=2 data_status['result']="Required fields are missing!!" return data_status # return redirect(url_for("user_configurations.entity_types_list",servicePage=servicePage,user_entity_search_element=user_entity_search_element)) else: flash("The staff member does not have permission to update kyc entity type status.", "danger") data_status['responseStatus']=4 return data_status # return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity",servicePage=servicePage,user_entity_search_element=user_entity_search_element)) else: data_status['result']=form.errors return data_status # return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity",servicePage=servicePage,user_entity_search_element=user_entity_search_element)) # Update region @user_configurations.route("/update_user_entity",methods=["POST","GET"]) @adminid_access_token_required @csrf_protect def update_user_entity(): data_status={"responseStatus":0,"result":""} if not session.get("adminId"): flash("Session expired Please login again.","danger") data_status["responseStatus"]=4 return data_status adminId=session.get("adminId") csrf_token = request.form.get("csrf_token") print(csrf_token,"((((((((((((((csrf_token))))))))))))))") latitude = request.form.get("latitude", "") longitude = request.form.get("longitude", "") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get('User-Agent').split(";") browser = loginBrowseData[0] existing_record = "" client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate=datetime.datetime.now() try: permissionsList = check_permissions(session.get("adminId"),"kycEntityTypesPermissions") if "edit" in permissionsList: entityTypeId = request.args.get("entityTypeId","") if request.method == "POST": form = KycEntityForm(request.form, current_id=entityTypeId) if form.validate_on_submit(): entityType = request.form.get("entityType","") isIndividual = request.form.get("isIndividual") isDefault = request.form.get("isDefault") remark = request.form.get("remark") jsonData = request.form.to_dict(flat=True) print("********** Form Data **********",request.form.to_dict(flat=True)) if entityType and remark: # if entityType and not is_valid_alphanumeric(entityType): # flash("User entity must contain only alphanumeric characters, spaces, and specific special characters: @#$()+_-/.") # return redirect(url_for("user_configurations.entity_types_list")) # if remark and not is_valid_alphanumeric(remark): # flash("Remark must contain only alphanumeric characters, spaces, and specific special characters: @#$()+_-/.") # return redirect(url_for("user_configurations.entity_types_list")) # isIndividual = True if isIndividual == "True" else False # added for some reason user_entity_queryset = UserEntity.objects(id=entityTypeId).first() existing_record = user_entity_queryset.to_json() message=user_entity_queryset.adminId.userName+" "+entityType+" User entity updated successfully!" requestData=[existing_record] updatedrequestData=[jsonData] save_remarks_data=save_admin_remarks_data(entityTypeId,adminId,remark,"entityType") save_admin_log_table = save_admin_logs_data(adminId,None,None,"update_user_entity","update",actionDate,client_ip,browser,message,requestData,updatedrequestData,latitude,longitude) if isIndividual == "True": isIndividual = True else: isIndividual = False if isDefault == "True": isDefault = True else: isDefault = False print("((((((((((((((((((((((((((((((((((((((((((((((isIndividual))))))))))))))))))))))))))))))))))))))))))))))",isIndividual) if user_entity_queryset: user_entity_queryset.update(entityType=entityType,isIndividual=isIndividual,isDefault=isDefault) # if isIndividual: # user_entity_queryset = UserEntity.objects(id__ne=entityTypeId,status__in=[0,1], isIndividual=True).all() # if user_entity_queryset: # user_entity_queryset.update(isIndividual=False) if isDefault: user_entity_queryset = UserEntity.objects(id__ne=entityTypeId,status__in=[0,1], isDefault=True).all() if user_entity_queryset: user_entity_queryset.update(isDefault=False) flash("User entity updated successfully","success") data_status["responseStatus"]=1 return data_status else: flash("Invalid ID","danger") data_status["responseStatus"]=4 return data_status else: data_status["responseStatus"]=2 data_status["result"]="Required fields are missing." return data_status else: data_status["result"]=form.errors return data_status else: flash("Invalid request", "danger") data_status["responseStatus"]=4 return data_status else: flash("The staff member does not have permission to update KYC Entity.", "danger") data_status["responseStatus"]=4 return data_status except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to update Entity details","danger") data_status["responseStatus"]=4 return data_status # Delete User type @user_configurations.route("/delete_user_entity",methods=["GET"]) @adminid_access_token_required def delete_user_entity(): try: if not session.get("adminId"): return redirect("admin_login") adminId=session.get("adminId") servicePage=request.args.get("servicePage") user_entity_search_element=request.args.get("user_entity_search_element") redirectTo=url_for("user_configurations.entity_types_list",servicePage=servicePage,user_entity_search_element=user_entity_search_element) latitude = request.args.get("latitude", "") longitude = request.args.get("longitude", "") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get('User-Agent').split(";") browser = loginBrowseData[0] client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate=datetime.datetime.now() existing_record = "" jsonData = request.form.to_dict(flat=True) updatedrequestData = [jsonData] permissionsList = check_permissions(session.get("adminId"),"kycEntityTypesPermissions") if "delete" in permissionsList: if request.method == "GET": entityTypeId = request.args.get("entityTypeId","") user_entity_queryset = UserEntity.objects(id=entityTypeId,status__in=[0,1]).first() existing_record = user_entity_queryset.to_json() requestData = [existing_record] user_entity_queryset.update(status=2) flash("User entity deleted successfully!","success") message=user_entity_queryset.adminId.userName+" "+user_entity_queryset.entityType+" User entity deleted successfully!" save_admin_log_table = save_admin_logs_data(adminId,None,None,"delete_user_entity","delete",actionDate,client_ip,browser,message,requestData,updatedrequestData,latitude,longitude) return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity",servicePage=servicePage,user_entity_search_element=user_entity_search_element)) else: flash("The staff member does not have permission to delete kyc entity type.", "danger") return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity",servicePage=servicePage,user_entity_search_element=user_entity_search_element)) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to delete user entity!!") return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity")) def fetching_csv_user_entity_details(user_entity_queryset): userEntityDict = {} try: if user_entity_queryset.entityType: userEntityDict["User Entity Type"] = user_entity_queryset.entityType else: userEntityDict["User type Name"] = "" if user_entity_queryset.status==0: userEntityDict["Status"] = "Activate" else: userEntityDict["Status"] = "Deactivate" except Exception as e: app.logger.error(traceback.format_exc()) return userEntityDict # View All regions @user_configurations.route("/userentity_csv_list",methods=["POST","GET"]) @adminid_access_token_required def userentity_csv_list(): if not session.get("adminId"): return redirect("admin_login") userEntityList = [] adminId = session.get("adminId") try: user_entity_queryset = UserEntity.objects(status__in=[0,1]).order_by("-id").all() for each_user_entity in user_entity_queryset: userEntityDict = fetching_csv_user_entity_details(each_user_entity) userEntityList.append(userEntityDict) fieldnames = ['User Entity Type','Status'] temp_csv_file_name = "/media/user_entity_csv_files/"+str(round(time.time() * 1000))+".csv" if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/user_entity_csv_files/")): os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/user_entity_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(userEntityList) 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='User entity list.csv', as_attachment=True ) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to export CSV data!!") return redirect("user_configurations.user_types_list") @user_configurations.route("/add_user_type_operation",methods=["POST","GET"]) @adminid_access_token_required def add_user_type_operation(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") if request.method == "POST": operationId = request.form.get("operationId","") userTypeId = request.form.get("userTypeId","") isDateValidity = request.form.get("isDateValidity") startDate = request.form.get("startDate") endDate = request.form.get("endDate") if operationId and userTypeId: try: user_type_operation_table = UserTypeOperations( adminId=adminId, operationId = operationId, userTypeId = userTypeId, isDateValidity = isDateValidity, endDate = endDate, startDate = startDate, createdOn = datetime.datetime.now(), status = 1, ) save_table = user_type_operation_table.save() flash("User type operation saved successfully!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserTypeOperation")) except Exception as e: flash("Unable to save user type operation!!") app.logger.error(traceback.format_exc()) return redirect(url_for("user_configurations.user_types_list",redirectTo="UserTypeOperation")) else: flash("Required fields are missing!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserTypeOperation")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to save user type operation!!" return render_template("super_admin_templates/user_configuration_list.html",error=error) def fetching_user_type_operation_details(user_type_operation_queryset): userTypeOperationDict = {} try: userTypeOperationDict={ "id":str(user_type_operation_queryset.id), "userTypeId":str(user_type_operation_queryset.userTypeId.id), "userTypeName":user_type_operation_queryset.userTypeId.userType, "operationId":str(user_type_operation_queryset.operationId.id), "operationName":user_type_operation_queryset.operationId.operationName, "isDateValidity":user_type_operation_queryset.isDateValidity } if user_type_operation_queryset.status==1: userTypeOperationDict["actionText"] = "Active" else: userTypeOperationDict["actionText"] = "Deactive" if user_type_operation_queryset.createdOn: userTypeOperationDict["createdOn"] = user_type_operation_queryset.createdOn.strftime("%m-%d-%Y") else: userTypeOperationDict["createdOn"] = "" if user_type_operation_queryset.startDate: userTypeOperationDict["startDate"] = user_type_operation_queryset.startDate.strftime("%m-%d-%Y") else: userTypeOperationDict["startDate"] = "-" if user_type_operation_queryset.endDate: userTypeOperationDict["endDate"] = user_type_operation_queryset.endDate.strftime("%m-%d-%Y") else: userTypeOperationDict["endDate"] = "-" except Exception as e: app.logger.error(traceback.format_exc()) return userTypeOperationDict @user_configurations.route("/update_user_type_operation_status",methods=["POST","GET"]) @adminid_access_token_required def update_user_type_operation_status(): if not session.get("adminId"): return redirect("admin_login") userTypeOperationId = request.args.get("userTypeOperationId","") if userTypeOperationId: try: user_type_operation_queryset = UserTypeOperations.objects(id=userTypeOperationId,status__nin=[2]).first() if user_type_operation_queryset: if user_type_operation_queryset.status == 0: user_type_operation_queryset.update(status=1) flash("User type operation activated successfully!") elif user_type_operation_queryset.status == 1: user_type_operation_queryset.update(status=0) flash("User type operation deactivated successfully!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserTypeOperation")) else: flash("Invaild id!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserTypeOperation")) except Exception as e: app.logger.error(traceback.format_exc()) return redirect(url_for("user_configurations.user_types_list",redirectTo="UserTypeOperation")) else: return redirect(url_for("user_configurations.user_types_list")) @user_configurations.route("/update_user_type_operation",methods=["POST","GET"]) @adminid_access_token_required def update_user_type_operation(): if not session.get("adminId"): return redirect("admin_login") try: userTypeOperationId = request.args.get("userTypeOperationId","") if request.method == "POST": operationId = request.form.get("operationId","") userTypeId = request.form.get("userTypeId","") endDate = request.form.get("endDate") startDate = request.form.get("startDate") isDateValidity = request.form.get("isDateValidity") user_type_operation_queryset = UserTypeOperations.objects(id=userTypeOperationId).first() if user_type_operation_queryset: user_type_operation_queryset.update( # isDateValidity=isDateValidity, # endDate = datetime.datetime.strptime(endDate,"%Y-%m-%d"), endDate = endDate, startDate = startDate # startDate = datetime.datetime.strptime(startDate,"%Y-%m-%d") ) if isDateValidity: user_type_operation_queryset.update(isDateValidity=True) else: user_type_operation_queryset.update(isDateValidity=False) if operationId: user_type_operation_queryset.update(operationId=ObjectId(operationId)) if userTypeId: user_type_operation_queryset.update(userTypeId=ObjectId(userTypeId)) flash("User type operation updated successfully!") return redirect(url_for("user_configurations.user_types_list")) else: flash("Invaild id!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserTypeOperation")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to update user type operation!!" return redirect(url_for("user_configurations.user_types_list",redirectTo="UserTypeOperation")) @user_configurations.route("/delete_user_type_operation",methods=["GET"]) @adminid_access_token_required def delete_user_type_operation(): try: if not session.get("adminId"): return redirect("admin_login") if request.method == "GET": userTypeOperationId = request.args.get("userTypeOperationId","") user_type_operation_queryset = UserTypeOperations.objects(id=userTypeOperationId,status__in=[0,1]).first() user_type_operation_queryset.update(status=2) flash("User type operation deleted successfully!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserTypeOperation")) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to delete user type operation!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="UserTypeOperation")) def fetching_csv_user_type_operation_details(user_type_operation_queryset): userTypeOperationDict = {} try: if user_type_operation_queryset.operationId: userTypeOperationDict["User Type Operation"] = user_type_operation_queryset.operationId.operationName else: userTypeOperationDict["User Type Operation"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return userTypeOperationDict @user_configurations.route("/user_type_operation_csv_list",methods=["POST","GET"]) @adminid_access_token_required def user_type_operation_csv_list(): if not session.get("adminId"): return redirect("admin_login") userTypeOperationList = [] adminId = session.get("adminId") try: user_type_operations_queryset = UserTypeOperations.objects(status__in=[0,1]).order_by("-id").all() for each_user_type_operation in user_type_operations_queryset: userTypeOperationDict = fetching_csv_user_type_operation_details(each_user_type_operation) userTypeOperationList.append(userTypeOperationDict) fieldnames = ['User Type Operation'] temp_csv_file_name = "/media/user_type_operation_csv_files/"+str(round(time.time() * 1000))+".csv" if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/user_type_operation_csv_files/")): os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/user_type_operation_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(userTypeOperationList) 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='User type operation list.csv', as_attachment=True ) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to export CSV data!!") return redirect("user_configurations.user_types_list") def fetching_operation_details(operation_queryset): operationDict = {} try: operationDict={ "id":str(operation_queryset.id), "operationName":operation_queryset.operationName, "operationTypeName":operation_queryset.operationTypeId.name, "userTypeName":operation_queryset.userTypeId.userType, "channelName":operation_queryset.channelId.name, "operationAuth":operation_queryset.operationAuth, "code":operation_queryset.code, "description":operation_queryset.description, "isUI":operation_queryset.isUI, "isMakerChecker":operation_queryset.isMakerChecker } if operation_queryset.status==1: operationDict["actionText"] = "Active" else: operationDict["actionText"] = "Deactive" if operation_queryset.createdOn: operationDict["createdOn"] = operation_queryset.createdOn.strftime("%m-%d-%Y") else: operationDict["createdOn"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return operationDict @user_configurations.route("/add_permission",methods=["POST","GET"]) @adminid_access_token_required @csrf_protect def add_permission(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") csrf_token = request.form.get("csrf_token") print(csrf_token,"((((((((((((((csrf_token))))))))))))))") latitude = request.form.get("latitude", "") longitude = request.form.get("longitude", "") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get('User-Agent').split(";") browser = loginBrowseData[0] client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate=datetime.datetime.now() permissionsList = check_permissions(adminId,"userPermissionsList") if "add" in permissionsList: if request.method == "POST": otp_check_id = request.form.get("otpLogid", "") defaultVerificationId = request.form.get("defaultVerificationId", "") print("otp_check_id",otp_check_id) if not otp_check_id: flash("Invalid Request.") return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) otpcheck_queryset = OtpChecks.objects(adminId=adminId,defaultVerificationField=defaultVerificationId, id=str(otp_check_id), status=1).first() if not otpcheck_queryset: flash("Invalid Request.") return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) # Update OTP status to 2 after verification otpcheck_queryset.update(status=2) permissionName = request.form.get("permissionName","") # permissionsList = request.form.getlist("permissionsList[]") groupingPermissionsList = request.form.getlist("groupingPermissions[]") categoryPermissionsList = request.form.getlist("categoryPermissions[]") servicePermissionsList = request.form.getlist("servicePermissions[]") operatorPermissionsList = request.form.getlist("operatorPermissions[]") agents = request.form.get("Agents") # commissions = request.form.get("commissions") ccCustomer = request.form.get("ccCustomer") utilityKycCheck = request.form.get("utilityKycCheck") vendorPayment = request.form.get("vendorPayment") payRent = request.form.get("payRent") grievance = request.form.get("grievance") # retailer = request.form.get("retailer") jsonData = request.form.to_dict(flat=True) requestData = [jsonData] updatedrequestData = [jsonData] if permissionName and not is_valid_alphanumeric(permissionName): flash("permission Name must contain only alphanumeric characters, spaces, and specific special characters: @#$()+_-/.") return redirect(url_for("user_configurations.user_types_list", redirectTo="Permissions")) # areadistributor = "areadistributor" in request.form # masterdistributor = "masterdistributor" in request.form # retailer = "retailer" in request.form # otherPermissionsList = [not areadistributor, not masterdistributor, not retailer] # print(otherPermissionsList) # print("((((((((otherPermissionsList))))))))") # retailer = request.form.get("retailer") otherPermissionsList=[] data ={} if agents: data["Agents"]=True else: data["Agents"]=False if ccCustomer: data["ccCustomer"]=True else: data["ccCustomer"]=False if payRent: data["payRent"]=True else: data["payRent"]=False if vendorPayment: data["vendorPayment"]=True else: data["vendorPayment"]=False if utilityKycCheck: data["utilityKycCheck"]=True else: data["utilityKycCheck"]=False if grievance: data["grievance"]=True else: data["grievance"]=False # if retailer: # data["retailer"]=True # else: # data["retailer"]=False otherPermissionsList.append(data) if permissionName: try: admin_queryset = SuperAdmin.objects(id=adminId,status=1).first() if admin_queryset: message=admin_queryset.userName+" "+permissionName+" User permission added successfully!" save_admin_log_table = save_admin_logs_data(adminId,None,None,"add_permission","create",actionDate,client_ip,browser,message,requestData,updatedrequestData,latitude,longitude) print("(((((((((((((save_admin_log_table)))))))))))))",save_admin_log_table) user_permissions_table = UserPermissions( adminId=adminId, permissionName = permissionName, groupingPermissionsList = groupingPermissionsList, categoryPermissionsList = categoryPermissionsList, servicePermissionsList = servicePermissionsList, operatorPermissionsList = operatorPermissionsList, otherPermissionsList = otherPermissionsList, createdOn = datetime.datetime.now(), status = 1, ) save_table = user_permissions_table.save() flash("User permission added successfully!") return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) except Exception as e: flash("Unable to add user permission!!") app.logger.error(traceback.format_exc()) return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) else: flash("Required fields are missing!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) else: flash("The staff member does not have permission to create a User permissions.", "danger") return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to add user permission!!" return render_template("super_admin_templates/user_configuration_list.html",error=error) # def fetching_service_groupings_details(service_groupings_queryset): # categoriesList =[] # servicesList =[] # operatorsList =[] # try: # serviceGroupingsdata={ # "id":str(service_groupings_queryset.id), # "name":service_groupings_queryset.name, # } # categories_queryset = Categories.objects(serviceGroupingId=str(service_groupings_queryset.id),status__in=[1]).all() # for each_category in categories_queryset: # category = { # "id": str(each_category.id), # "categoryName": each_category.categoryName # } # categoriesList.append(category) # services_queryset = Service.objects(categoryId=str(each_category.id),status__in=[1]).all() # for each_service in services_queryset: # service = { # # "id": str(each_service.id), # "serviceName": each_service.serviceName, # } # servicesList.append(service) # operators_queryset = Operators.objects(serviceId=str(each_service.id),status__in=[1]).all() # for each_operator in operators_queryset: # operator = { # "id": str(each_operator.id), # "operatorName": each_operator.operatorName # } # operatorsList.append(operator) # serviceGroupingsdata["operatorsList"]=operatorsList # serviceGroupingsdata["servicesList"]= servicesList # serviceGroupingsdata["categoriesList"]= categoriesList # except: # app.logger.error(traceback.format_exc()) # return serviceGroupingsdata def fetching_user_permissions_details(user_permission_queryset): user_permission_dict = {} try: user_permission_dict={ "id":str(user_permission_queryset.id), "name":user_permission_queryset.permissionName, "groupingsIdsList":[each for each in user_permission_queryset.groupingPermissionsList], "categoryIdsList":[each for each in user_permission_queryset.categoryPermissionsList], "serviceIdsList":[each for each in user_permission_queryset.servicePermissionsList], "operatorIdsList":[each for each in user_permission_queryset.operatorPermissionsList], "otherServices":user_permission_queryset.otherPermissionsList[0] } if user_permission_queryset.status==1: user_permission_dict["actionText"] = "Active" else: user_permission_dict["actionText"] = "Deactive" if user_permission_queryset.createdOn: user_permission_dict["createdOn"] = user_permission_queryset.createdOn.strftime("%m-%d-%Y") else: user_permission_dict["createdOn"] = "" groupingsList =[] for each_id in user_permission_queryset.groupingPermissionsList: groupings_queryset = ServiceGrouping.objects(id=each_id).first() name = groupings_queryset.name groupingsList.append(name) categoriesList = [] for each_id in user_permission_queryset.categoryPermissionsList: category_queryset = Categories.objects(id=each_id).first() categoryName = category_queryset.categoryName categoriesList.append(categoryName) servicesList = [] for each_id in user_permission_queryset.servicePermissionsList: service_queryset = Service.objects(id=each_id).first() serviceName = service_queryset.serviceName servicesList.append(serviceName) operatorsList = [] for each_id in user_permission_queryset.operatorPermissionsList: operator_queryset = Operators.objects(id=each_id).first() operatorName = operator_queryset.operatorName operatorsList.append(operatorName) # print(operatorsList) # print("(((((((((((((operatorsList)))))))))))))") user_permission_dict["groupingsList"] = groupingsList user_permission_dict["categoriesList"] = categoriesList user_permission_dict["servicesList"] = servicesList user_permission_dict["operatorsList"] = operatorsList except Exception as e: app.logger.error(traceback.format_exc()) return user_permission_dict def fetching_service_groupings_details(service_groupings_queryset): serviceGroupingsdata = { "id": str(service_groupings_queryset.id), "name": service_groupings_queryset.name, "categoriesList": [] } try: categoriesList = [] categories_queryset=Categories.objects(serviceGroupingId=str(service_groupings_queryset.id), status__in=[1]).all() for each_category in categories_queryset: category = { "id": str(each_category.id), "categoryName": each_category.categoryName, "servicesList": [] } servicesList = [] services_queryset = Service.objects(categoryIdsList__in=[each_category.id], status__in=[1]).all() for each_service in services_queryset: service = { "id": str(each_service.id), "serviceName": each_service.serviceName, "operatorsList": [] } operatorsList = [] operators_queryset = Operators.objects(serviceId=str(each_service.id), status__in=[1]).all() for each_operator in operators_queryset: operator = { "id": str(each_operator.id), "operatorName": each_operator.operatorName } operatorsList.append(operator) service["operatorsList"] = operatorsList servicesList.append(service) category["servicesList"] = servicesList categoriesList.append(category) serviceGroupingsdata["categoriesList"] = categoriesList except Exception as e: app.logger.error(traceback.format_exc()) return serviceGroupingsdata # Update regions type status @user_configurations.route("/update_permission_status",methods=["POST","GET"]) @adminid_access_token_required def update_permission_status(): if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") latitude = request.args.get("latitude", "") longitude = request.args.get("longitude", "") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get("User-Agent").split(";") browser = loginBrowseData[0] client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate=datetime.datetime.now() jsonData = request.form.to_dict(flat=True) existing_record = "" updatedrequestData = [jsonData] permissionsList = check_permissions(session.get("adminId"),"userPermissionsList") if "edit" in permissionsList: permissionId = request.args.get("permissionId","") if permissionId: try: message="" user_permission_queryset = UserPermissions.objects(id=permissionId,status__nin=[2]).first() existing_record = user_permission_queryset.to_json() requestData = [existing_record] if user_permission_queryset: if user_permission_queryset.status == 0: user_permission_queryset.update(status=1) flash("User permission activated successfully!") message=user_permission_queryset.adminId.userName+" "+user_permission_queryset.permissionName+" User permission activated successfully!" elif user_permission_queryset.status == 1: user_permission_queryset.update(status=0) flash("User permission deactivated successfully!") message=user_permission_queryset.adminId.userName+" "+user_permission_queryset.permissionName+" User permission deactivated successfully!" save_admin_log_table = save_admin_logs_data(adminId,None,None,"update_permission_status","updatestatus",actionDate,client_ip,browser,message,requestData,updatedrequestData,latitude,longitude) print("((((((((((((((save_admin_log_table))))))))))))))",save_admin_log_table) return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) else: flash("Invaild id!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) except Exception as e: app.logger.error(traceback.format_exc()) return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) else: return redirect(url_for("user_configurations.user_types_list")) else: flash("The staff member does not have permission to update User permission status", "danger") return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) @user_configurations.route("/delete_user_permission",methods=["GET"]) @adminid_access_token_required def delete_user_permission(): try: if not session.get("adminId"): return redirect("admin_login") if request.method == "GET": permissionId = request.args.get("permissionId","") user_permission_queryset = UserPermissions.objects(id=permissionId,status__nin=[2]).first() user_permission_queryset.update(status=2) flash("User permission deleted successfully!") return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to delete user user_permission_queryset!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) @user_configurations.route("/update_user_permission",methods=["POST","GET"]) @adminid_access_token_required @csrf_protect def update_user_permission(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") csrf_token = request.form.get("csrf_token") print(csrf_token,"((((((((((((((csrf_token))))))))))))))") latitude = request.form.get("latitude", "") longitude = request.form.get("longitude", "") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get('User-Agent').split(";") browser = loginBrowseData[0] existing_record = "" client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate=datetime.datetime.now() print(f"Content length: {request.content_length} bytes") if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") permissionId = request.args.get("permissionId","s") if request.method == "GET": userpermissionDict={} patternsList =[] categorysList=[] subCategorysList=[] entityTypeId=[] statesList=[] citiesList=[] blocksList =[] pincodesList=[] masterIfscBanksList=[] user_permission_queryset = UserPermissions.objects(id=permissionId, status__in=[0,1]).first() userpermissionDict = fetching_user_permissions_details(user_permission_queryset) return render_template("super_admin_templates/update_permission_details.html",userpermissionDict=userpermissionDict) permissionsList = check_permissions(session.get("adminId"),"userPermissionsList") if "edit" in permissionsList: if request.method == "POST": otp_check_id = request.form.get("otpLogid", "") defaultVerificationId = request.form.get("defaultVerificationId", "") print("otp_check_id",otp_check_id) if not otp_check_id: flash("Invalid Request.") return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) otpcheck_queryset = OtpChecks.objects(adminId=adminId,defaultVerificationField=defaultVerificationId, id=str(otp_check_id), status=1).first() if not otpcheck_queryset: flash("Invalid Request.") return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) # Update OTP status to 2 after verification otpcheck_queryset.update(status=2) permissionId = request.args.get("permissionId","") permissionName = request.form.get("permissionName","") groupingPermissionsList = request.form.getlist("groupingPermissions[]") categoryPermissionsList = request.form.getlist("categoryPermissions[]") servicePermissionsList = request.form.getlist("servicePermissions[]") operatorPermissionsList = request.form.getlist("operatorPermissions[]") agents = request.form.get("Agents") ccCustomer = request.form.get("ccCustomer") vendorPayment = request.form.get("vendorPayment") utilityKycCheck = request.form.get("utilityKycCheck") payRent = request.form.get("payRent") grievance = request.form.get("grievance") # commissions = request.form.get("commissions") # payin = request.form.get("payin") # retailer = request.form.get("retailer") if permissionName and not is_valid_alphanumeric(permissionName): flash("permission Name must contain only alphanumeric characters, spaces, and specific special characters: @#$()+_-/.") return redirect(url_for("user_configurations.user_types_list", redirectTo="Permissions")) otherPermissionsList=[] data ={} if agents: data["Agents"]=True else: data["Agents"]=False # if commissions: # data["Commissions"]=True # else: # data["Commissions"]=False if ccCustomer: data["ccCustomer"]=True else: data["ccCustomer"]=False if vendorPayment: data["vendorPayment"]=True else: data["vendorPayment"]=False if utilityKycCheck: data["utilityKycCheck"]=True else: data["utilityKycCheck"]=False if payRent: data["payRent"]=True else: data["payRent"]=False if grievance: data["grievance"]=True else: data["grievance"]=False # if retailer: # data["retailer"]=True # else: # data["retailer"]=False otherPermissionsList.append(data) jsonData = request.form.to_dict(flat=True) if permissionName: try: user_permission_queryset = UserPermissions.objects(id=permissionId,status__nin=[2]).first() existing_record = user_permission_queryset.to_json() message=user_permission_queryset.adminId.userName+" "+permissionName+" User permissions updated successfully!" requestData=[user_permission_queryset] updatedrequestData=[jsonData] save_admin_log_table = save_admin_logs_data(adminId,None,None,"update_user_permission","update",actionDate,client_ip,browser,message,requestData,updatedrequestData,latitude,longitude) print("(((((((((((((save_admin_log_table)))))))))))))",save_admin_log_table) if user_permission_queryset: user_permission_queryset.update( permissionName = permissionName, groupingPermissionsList = groupingPermissionsList, categoryPermissionsList = categoryPermissionsList, servicePermissionsList = servicePermissionsList, operatorPermissionsList = operatorPermissionsList, otherPermissionsList = otherPermissionsList, ) flash("User permissions updated successfully!") return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) except Exception as e: flash("Unable to update user permissions!!") app.logger.error(traceback.format_exc()) return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) else: flash("Required fields are missing!!") return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) else: flash("The staff member does not have permission to update User permissions.", "danger") return redirect(url_for("user_configurations.user_types_list",redirectTo="Permissions")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to update user permissions!!" return render_template("super_admin_templates/user_configuration_list.html",error=error) @user_configurations.route("/entity_types_list",methods=["POST","GET"]) @adminid_access_token_required def entity_types_list(): if not session.get("adminId"): return redirect("admin_login") userEntityList = [] entityPagination = None adminId = session.get("adminId") user_entity_search_element="" snoCount=0 servicePage=None permissionsList = check_permissions(adminId,"kycEntityTypesPermissions") if "view" in permissionsList: try: redirectTo = request.args.get("redirectTo","UserEntity") if redirectTo: redirectval = redirectTo else: redirectval = "UserEntity" form = entityFormSearch(request.args) user_entity_search_element = request.args.get("user_entity_search_element", "") servicePage = request.args.get(get_page_parameter("servicePage"), type=int, default=1) per_page = 20 start = (servicePage - 1) * per_page total_count=0 filters = Q(status__in=[0, 1]) print(per_page,"((((((((((((((per_page))))))))))))))") if form.validate(): if user_entity_search_element: filters &= Q(entityType__icontains=user_entity_search_element) total_count = UserEntity.objects(filters).count() user_entity_queryset = ( UserEntity.objects(filters) .only("id", "entityType", "isIndividual","isDefault","status") .order_by("-id") .skip(start) .limit(per_page) ) userEntityList=list(user_entity_queryset) else: userEntityList=[] snoCount = start # user_entity_search_element = request.args.get("user_entity_search_element","") # ######################## User Entities List Start ################################## # user_entities_queryset = UserEntity.objects(status__in=[0,1]).order_by("-id") # if user_entity_search_element: # user_entities_queryset = user_entities_queryset.filter(Q(entityType__icontains=user_entity_search_element)) # # Get the current page from the query parameters # entityPage = request.args.get("entityPage", type=int, default=1) # Use a unique query parameter name # entity_per_page = 20 # Number of items per page # # Query the database for the current page's data # total_entity_count = user_entities_queryset.count() # startEntity = (entityPage - 1) * entity_per_page # endEntity = min(startEntity + entity_per_page, total_entity_count) # user_entities = user_entities_queryset[startEntity:endEntity] # for each_user_entity in user_entities: # userEntityDict = fetching_user_entity_details(each_user_entity) # userEntityList.append(userEntityDict) # # Pagination object for rendering pagination controls in the template # entityPagination = Pagination(page=entityPage, # total=total_entity_count, per_page=entity_per_page, # alignment="right",page_parameter="entityPage",href="entity_types_list?redirectTo=UserEntity&entityPage={0}") ############################### User Entitys List End ################################## entityPagination = Pagination(servicePage=servicePage, total=total_count, per_page=per_page,page_parameter ="servicePage", alignment="right", record_name="UserEntity",href=f"?user_entity_search_element={user_entity_search_element}&servicePage={{0}}") print("((((((((((((((((((((userEntityList))))))))))))))))))))",userEntityList) return render_template("super_admin_templates/entity_types_list.html", entityPagination=entityPagination, userEntityList=userEntityList, redirectval=redirectval, user_entity_search_element=user_entity_search_element, snoCount=snoCount, servicePage = servicePage, form=form, ) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to fetch entity types details" return render_template("super_admin_templates/entity_types_list.html", error=error, entityPagination=entityPagination, userEntityList=userEntityList, user_entity_search_element=user_entity_search_element, snoCount=snoCount, servicePage = servicePage, form=form, ) else: flash("The staff member does not have permission to view kyc entity type.", "danger") return render_template("super_admin_templates/entity_types_list.html")