from appservices.common.util import * user_configurations = Blueprint("user_configurations",__name__) # Add region @user_configurations.route("/add_user_type",methods=["POST","GET"]) def add_user_type(): try: 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() permissionsList = check_permissions(adminId,"kycEntityTypesPermissions") if "add" in permissionsList: if request.method == "POST": action = request.form.get("action", "HI").strip() data = request.form.to_dict() # Step 1: 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)) # tep 2: Handle OTP Verification elif action == "verify": otp_check_id = request.form.get("otpCheckId","SAI") otp_code = request.form.get("otpCode","K") 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)) # Step 3: Ensure OTP is Verified Before Updating Service Charges elif action == "update": otp_check_id = request.form.get("otpCheckId", "") otp_record = OtpChecks.objects(id=otp_check_id, status=1).first() admin_id = request.form.get("adminId", "") otpcheck_queryset = OtpChecks.objects(adminId=str(admin_id), 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"]) def user_types_list(): if not session.get("adminId"): return redirect("admin_login") userTypesList = [] userEntityList = [] userNameFormatList = [] userTypeOperationList = [] operationList = [] serviceGroupingsList =[] userPermissionsList =[] userpermissionpagination="" adminId = session.get("adminId") permissionsList = check_permissions(adminId,"kycEntityTypesPermissions") if "view" in permissionsList: try: action = request.form.get("action", "HI").strip() data = request.form.to_dict() # Step 1: 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)) # tep 2: Handle OTP Verification elif action == "verify": otp_check_id = request.form.get("otpCheckId","SAI") otp_code = request.form.get("otpCode","K") 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)) # Step 3: Ensure OTP is Verified Before Updating Service Charges elif action == "update": otp_check_id = request.form.get("otpCheckId", "") otp_record = OtpChecks.objects(id=otp_check_id, status=1).first() admin_id = request.form.get("adminId", "") otpcheck_queryset = OtpChecks.objects(adminId=str(admin_id), 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!"}) redirectTo = request.args.get("redirectTo","UserType") if redirectTo: redirectval = redirectTo else: redirectval = "UserType" user_type_search_element = request.form.get("user_type_search_element","") user_entity_search_element = request.form.get("user_entity_search_element","") user_format_search_element = request.form.get("user_format_search_element","") user_operation_search_element = request.form.get("user_operation_search_element","") ######################## User Types List Start ################################## 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") 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, redirectval=redirectval, serviceGroupingsList=serviceGroupingsList, userPermissionsList=userPermissionsList, pagination=pagination, entityPagination=entityPagination, userpermissionpagination=userpermissionpagination, ) 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, redirectval=redirectval, serviceGroupingsList =serviceGroupingsList, userPermissionsList =userPermissionsList, pagination =pagination, entityPagination =entityPagination, userpermissionpagination=userpermissionpagination, ) else: flash("Staff member does not have given view kyc entity type permissions!!") return render_template("super_admin_templates/user_configuration_list.html") # Update regions type status @user_configurations.route("/update_user_type_status",methods=["POST","GET"]) 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"),"kycEntityTypesPermissions") 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"]) 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") 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"),"kycEntityTypesPermissions") if "edit" in permissionsList: try: userTypeId = request.args.get("userTypeId","") if request.method == "POST": action = request.form.get("action", "HI").strip() data = request.form.to_dict() # Step 1: 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)) # tep 2: Handle OTP Verification elif action == "verify": otp_check_id = request.form.get("otpCheckId","SAI") otp_code = request.form.get("otpCode","K") 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)) # Step 3: Ensure OTP is Verified Before Updating Service Charges elif action == "update": otp_check_id = request.form.get("otpCheckId", "") otp_record = OtpChecks.objects(id=otp_check_id, status=1).first() admin_id = request.form.get("adminId", "") otpcheck_queryset = OtpChecks.objects(adminId=str(admin_id), 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"]) def delete_user_type(): try: 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"),"kycEntityTypesPermissions") if "delete" in permissionsList: if request.method == "GET": userTypeId = request.args.get("userTypeId","") user_type_queryset = UserTypes.objects(id=userTypeId,status__in=[0,1]).first() existing_record = user_type_queryset.to_json() requestData = [existing_record] user_type_queryset.update(status=2) flash("User type deleted successfully!") message=user_type_queryset.adminId.userName+" "+user_type_queryset.userType+" 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"]) 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"]) 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"]) 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"]) 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"]) 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"]) 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"]) def add_user_entity(): try: 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() permissionsList = check_permissions(adminId,"kycEntityTypesPermissions") if "add" in permissionsList: if request.method == "POST": entityType = request.form.get("entityType","") isIndividual = request.form.get("isIndividual") jsonData = request.form.to_dict(flat=True) requestData = [jsonData] updatedrequestData = [jsonData] if entityType: try: 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) if isIndividual == "True": isIndividual = True else: isIndividual = False user_entity_table = UserEntity( adminId=adminId, entityType = entityType, isIndividual = isIndividual, createdOn = datetime.datetime.now(), status = 1, ) save_table = user_entity_table.save() entityTypeId = str(save_table.id) flash("User entity saved successfully!") return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity")) except Exception as e: flash("Unable to save user entity!!") app.logger.error(traceback.format_exc()) return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity")) else: flash("Required fields are missing!!") return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity")) else: flash("Staff member does not have given create kyc entity type permissions!!") return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity")) 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) 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 } 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"]) def update_user_entity_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"),"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!") 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!") 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) save_remarks_data=save_admin_remarks_data(entityTypeId,adminId,remark,"merchant") return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity")) else: flash("Invaild id!!") return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity")) except Exception as e: app.logger.error(traceback.format_exc()) return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity")) else: return redirect(url_for("user_configurations.entity_types_list")) else: flash("Staff member does not have given update kyc entity type permissions!!") return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity")) # Update region @user_configurations.route("/update_user_entity",methods=["POST","GET"]) def update_user_entity(): 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] 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"),"kycEntityTypesPermissions") if "edit" in permissionsList: try: entityTypeId = request.args.get("entityTypeId","") if request.method == "POST": entityType = request.form.get("entityType","") isIndividual = request.form.get("isIndividual") remark = request.form.get("remark") jsonData = request.form.to_dict(flat=True) 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) if isIndividual == "True": isIndividual = True else: isIndividual = False if user_entity_queryset: user_entity_queryset.update(entityType=entityType,isIndividual=isIndividual) flash("User entity updated successfully!") return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity")) else: flash("Invaild id!!") return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to update user entity!!" return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity")) else: flash("Staff member does not have given update kyc entity type permissions!!") return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity")) # Delete User type @user_configurations.route("/delete_user_entity",methods=["GET"]) def delete_user_entity(): try: 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() existing_record = "" updatedrequestData = [jsonData] jsonData = request.form.to_dict(flat=True) 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!") 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) return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity")) else: flash("Staff member does not have given delete kyc entity type permissions!!") return redirect(url_for("user_configurations.entity_types_list",redirectTo="UserEntity")) 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"]) 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"]) 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"]) 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"]) 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"]) 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"]) 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"]) def add_permission(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") if request.method == "POST": 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") # 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: 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")) 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(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, "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"]) def update_permission_status(): if not session.get("adminId"): return redirect("admin_login") permissionId = request.args.get("permissionId","") if permissionId: try: user_permission_queryset = UserPermissions.objects(id=permissionId,status__nin=[2]).first() if user_permission_queryset: if user_permission_queryset.status == 0: user_permission_queryset.update(status=1) flash("User permission activated successfully!") elif user_permission_queryset.status == 1: user_permission_queryset.update(status=0) flash("User permission deactivated successfully!") 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")) @user_configurations.route("/delete_user_permission",methods=["GET"]) 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"]) def update_user_permission(): try: 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) if request.method == "POST": 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") 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) if permissionName: try: user_permission_queryset = UserPermissions.objects(id=permissionId,status__nin=[2]).first() 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")) 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"]) def entity_types_list(): if not session.get("adminId"): return redirect("admin_login") userEntityList = [] entityPagination = "" adminId = session.get("adminId") permissionsList = check_permissions(adminId,"kycEntityTypesPermissions") if "view" in permissionsList: try: redirectTo = request.args.get("redirectTo","UserEntity") if redirectTo: redirectval = redirectTo else: redirectval = "UserEntity" user_entity_search_element = request.form.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 ################################## return render_template("super_admin_templates/entity_types_list.html", userEntityList=userEntityList, entityPagination=entityPagination, ) 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, userEntityList=userEntityList, entityPagination=entityPagination, ) else: flash("Staff member does not have given view kyc entity type permissions!!") return render_template("super_admin_templates/entity_types_list.html")