from appservices.common.util import * system_configurations = Blueprint("system_configurations",__name__) @system_configurations.route("/add_channel",methods=["POST","GET"]) def add_channel(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") if request.method == "POST": name = request.form.get("name","") if name: try: channel_table = Channels( adminId=adminId, name = name, createdOn = datetime.datetime.now(), status = 1, ) save_table = channel_table.save() flash("Channel saved successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Channel")) except Exception as e: flash("Unable to save channel!!") app.logger.error(traceback.format_exc()) return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Channel")) else: flash("Required fields are missing!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Channel")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to save channel!!" return render_template("super_admin_templates/system_configuration_list.html",error=error) def fetching_channel_details(channel_queryset): channelDict = {} try: channelDict={ "id":str(channel_queryset.id), "name":channel_queryset.name } if channel_queryset.status==1: channelDict["actionText"] = "Active" else: channelDict["actionText"] = "Deactive" if channel_queryset.createdOn: channelDict["createdOn"] = channel_queryset.createdOn.strftime("%m-%d-%Y") else: channelDict["createdOn"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return channelDict @system_configurations.route("/system_configurations_list",methods=["POST","GET"]) def system_configurations_list(): if not session.get("adminId"): return redirect("admin_login") channelList = [] transactionTypeList = [] operationTypeList = [] userTypesList = [] operationList = [] errorList = [] adminId = session.get("adminId") try: redirectTo = request.args.get("redirectTo","Channel") if redirectTo: redirectVal = redirectTo else: redirectVal = "Channel" channel_search_element = request.form.get("channel_search_element","") transaction_search_element = request.form.get("transaction_search_element","") operation_type_search_element = request.form.get("operation_type_search_element","") operation_search_element = request.form.get("operation_search_element","") error_search_element = request.form.get("error_search_element","") channels_queryset = Channels.objects(status__in=[0,1]).order_by("-id") if channel_search_element: channels_queryset = channels_queryset.filter(Q(name__icontains=channel_search_element)) for each_channel in channels_queryset: channelDict = fetching_channel_details(each_channel) channelList.append(channelDict) transactions_type_queryset = TransactionTypes.objects(status__in=[0,1]).order_by("-id") if transaction_search_element: transactions_type_queryset = transactions_type_queryset.filter(Q(transactionType__icontains=transaction_search_element)) for each_transaction in transactions_type_queryset: transactionDict = fetching_transaction_details(each_transaction) transactionTypeList.append(transactionDict) operations_type_queryset = OperationTypes.objects(status__in=[0,1]).order_by("-id") if operation_type_search_element: operations_type_queryset = operations_type_queryset.filter(Q(name__icontains=operation_type_search_element)) for each_operation_type in operations_type_queryset: operationTypeDict = fetching_operation_type_details(each_operation_type) operationTypeList.append(operationTypeDict) user_types_queryset = UserTypes.objects(status__in=[0,1]).order_by("-id") for each_user_type in user_types_queryset: userTypeDict = fetching_user_type_details(each_user_type) userTypesList.append(userTypeDict) operations_queryset = Operations.objects(status__in=[0,1]).order_by("-id") if operation_search_element: operations_queryset = operations_queryset.filter(Q(operationName__icontains=operation_search_element)) for each_operation in operations_queryset: operationDict = fetching_operation_details(each_operation) operationList.append(operationDict) error_definations_queryset = ErrorDefinations.objects(status__in=[0,1]).order_by("-id") if error_search_element: error_definations_queryset = error_definations_queryset.filter(Q(errorCode__icontains=error_search_element)) for each_error in error_definations_queryset: errorDict = fetching_error_defination_details(each_error) errorList.append(errorDict) return render_template("super_admin_templates/system_configuration_list.html", channelList=channelList, errorList=errorList, userTypesList=userTypesList, operationList=operationList, transactionTypeList=transactionTypeList, operationTypeList=operationTypeList, transaction_search_element=transaction_search_element, operation_search_element=operation_search_element, operation_type_search_element=operation_type_search_element, error_search_element=error_search_element, channel_search_element=channel_search_element, redirectVal = redirectVal ) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to fetch system configurations details!!" return render_template("super_admin_templates/system_configuration_list.html", error=error, channelList=channelList, errorList=errorList, userTypesList=userTypesList, operationTypeList=operationTypeList, transactionTypeList=transactionTypeList, operationList=operationList, error_search_element=error_search_element, transaction_search_element=transaction_search_element, operation_search_element=operation_search_element, operation_type_search_element=operation_type_search_element, channel_search_element=channel_search_element, redirectVal = redirectVal ) @system_configurations.route("/update_channel_status",methods=["POST","GET"]) def update_channel_status(): if not session.get("adminId"): return redirect("admin_login") channelId = request.args.get("channelId","") if channelId: try: channel_queryset = Channels.objects(id=channelId,status__nin=[2]).first() if channel_queryset: if channel_queryset.status == 0: channel_queryset.update(status=1) flash("Channel activated successfully!") elif channel_queryset.status == 1: channel_queryset.update(status=0) flash("Channel deactivated successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Channel")) else: flash("Invaild id!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Channel")) except Exception as e: app.logger.error(traceback.format_exc()) return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Channel")) else: return redirect(url_for("system_configurations.system_configurations_list")) @system_configurations.route("/update_channel",methods=["POST","GET"]) def update_channel(): if not session.get("adminId"): return redirect("admin_login") try: channelId = request.args.get("channelId","") if request.method == "POST": name = request.form.get("name","") channel_queryset = Channels.objects(id=channelId).first() if channel_queryset: channel_queryset.update( name=name ) flash("Channel updated successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Channel")) else: flash("Invaild id!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Channel")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to update channel!!" return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Channel")) @system_configurations.route("/delete_channel",methods=["GET"]) def delete_channel(): try: if not session.get("adminId"): return redirect("admin_login") if request.method == "GET": channelId = request.args.get("channelId","") channel_queryset = Channels.objects(id=channelId,status__in=[0,1]).first() channel_queryset.update(status=2) flash("Channel deleted successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Channel")) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to delete channel!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Channel")) def fetching_csv_channel_details(channel_queryset): channelDict = {} try: if channel_queryset.name: channelDict["Name"] = channel_queryset.name else: channelDict["Name"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return channelDict @system_configurations.route("/channels_csv_list",methods=["POST","GET"]) def channels_csv_list(): if not session.get("adminId"): return redirect("admin_login") channelList = [] adminId = session.get("adminId") try: channels_queryset = Channels.objects(status__in=[0,1]).order_by("-id").all() for each_channel in channels_queryset: channelDict = fetching_csv_channel_details(each_channel) channelList.append(channelDict) fieldnames = ['Name'] temp_csv_file_name = "/media/channel_csv_files/"+str(round(time.time() * 1000))+".csv" if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/channel_csv_files/")): os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/channel_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(channelList) 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='Channel list.csv', as_attachment=True ) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to export CSV data!!") return redirect("system_configurations.system_configurations_list") @system_configurations.route("/add_transaction_type",methods=["POST","GET"]) def add_transaction_type(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") if request.method == "POST": transactionType = request.form.get("transactionType","") code = request.form.get("code","") if transactionType and code: try: transaction_type_table = TransactionTypes( adminId=adminId, transactionType = transactionType, code = code, createdOn = datetime.datetime.now(), status = 1, ) save_table = transaction_type_table.save() flash("Transaction type saved successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="TransactionType")) except Exception as e: flash("Unable to save transaction type!!") app.logger.error(traceback.format_exc()) return redirect(url_for("system_configurations.system_configurations_list",redirectTo="TransactionType")) else: flash("Required fields are missing!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="TransactionType")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to save transaction type!!" return render_template("super_admin_templates/system_configuration_list.html",error=error) def fetching_transaction_details(transaction_type_queryset): transactionDict = {} try: transactionDict={ "id":str(transaction_type_queryset.id), "transactionType":transaction_type_queryset.transactionType, "code":transaction_type_queryset.code } if transaction_type_queryset.status==1: transactionDict["actionText"] = "Active" else: transactionDict["actionText"] = "Deactive" if transaction_type_queryset.createdOn: transactionDict["createdOn"] = transaction_type_queryset.createdOn.strftime("%m-%d-%Y") else: transactionDict["createdOn"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return transactionDict @system_configurations.route("/update_transaction_type_status",methods=["POST","GET"]) def update_transaction_type_status(): if not session.get("adminId"): return redirect("admin_login") transactionTypeId = request.args.get("transactionTypeId","") if transactionTypeId: try: transaction_type_queryset = TransactionTypes.objects(id=transactionTypeId,status__nin=[2]).first() if transaction_type_queryset: if transaction_type_queryset.status == 0: transaction_type_queryset.update(status=1) flash("Transaction type activated successfully!") elif transaction_type_queryset.status == 1: transaction_type_queryset.update(status=0) flash("Transaction type deactivated successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="TransactionType")) else: flash("Invaild id!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="TransactionType")) except Exception as e: app.logger.error(traceback.format_exc()) return redirect(url_for("system_configurations.system_configurations_list",redirectTo="TransactionType")) else: return redirect(url_for("system_configurations.system_configurations_list")) @system_configurations.route("/update_transaction_type",methods=["POST","GET"]) def update_transaction_type(): if not session.get("adminId"): return redirect("admin_login") try: transactionTypeId = request.args.get("transactionTypeId","") if request.method == "POST": transactionType = request.form.get("transactionType","") code = request.form.get("code","") transaction_type_queryset = TransactionTypes.objects(id=transactionTypeId).first() if transaction_type_queryset: transaction_type_queryset.update( transactionType=transactionType, code=code ) flash("Transaction type updated successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="TransactionType")) else: flash("Invaild id!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="TransactionType")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to update transaction type!!" return redirect(url_for("system_configurations.system_configurations_list",redirectTo="TransactionType")) @system_configurations.route("/delete_transaction_type",methods=["GET"]) def delete_transaction_type(): try: if not session.get("adminId"): return redirect("admin_login") if request.method == "GET": transactionTypeId = request.args.get("transactionTypeId","") transaction_type_queryset = TransactionTypes.objects(id=transactionTypeId,status__in=[0,1]).first() transaction_type_queryset.update(status=2) flash("Transaction type deleted successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="TransactionType")) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to delete transaction type!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="TransactionType")) def fetching_csv_transaction_type_details(transaction_type_queryset): transactionDict = {} try: if transaction_type_queryset.transactionType: transactionDict["Transaction Type"] = transaction_type_queryset.transactionType else: transactionDict["Transaction Type"] = "" if transaction_type_queryset.code: transactionDict["Code"] = transaction_type_queryset.code else: transactionDict["Code"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return transactionDict @system_configurations.route("/transaction_type_csv_list",methods=["POST","GET"]) def transaction_type_csv_list(): if not session.get("adminId"): return redirect("admin_login") transactionTypeList = [] adminId = session.get("adminId") try: transactions_type_queryset = TransactionTypes.objects(status__in=[0,1]).order_by("-id").all() for each_transaction in transactions_type_queryset: transactionDict = fetching_csv_transaction_type_details(each_transaction) transactionTypeList.append(transactionDict) fieldnames = ['Transaction Type','Code'] temp_csv_file_name = "/media/transaction_type_csv_files/"+str(round(time.time() * 1000))+".csv" if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/transaction_type_csv_files/")): os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/transaction_type_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(transactionTypeList) save_csv_file = CSVFileExports( transactionType=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='Transaction type list.csv', as_attachment=True ) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to export CSV data!!") return redirect("system_configurations.system_configurations_list") @system_configurations.route("/add_operation_type",methods=["POST","GET"]) def add_operation_type(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") if request.method == "POST": name = request.form.get("name","") if name: try: operation_table = OperationTypes( adminId=adminId, name = name, createdOn = datetime.datetime.now(), status = 1, ) save_table = operation_table.save() flash("Operation type saved successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="OperationType")) except Exception as e: flash("Unable to save operation type!!") app.logger.error(traceback.format_exc()) return redirect(url_for("system_configurations.system_configurations_list",redirectTo="OperationType")) else: flash("Required fields are missing!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="OperationType")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to save operation type!!" return render_template("super_admin_templates/system_configuration_list.html",error=error) def fetching_operation_type_details(operation_type_queryset): operationTypeDict = {} try: operationTypeDict={ "id":str(operation_type_queryset.id), "name":operation_type_queryset.name } if operation_type_queryset.status==1: operationTypeDict["actionText"] = "Active" else: operationTypeDict["actionText"] = "Deactive" if operation_type_queryset.createdOn: operationTypeDict["createdOn"] = operation_type_queryset.createdOn.strftime("%m-%d-%Y") else: operationTypeDict["createdOn"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return operationTypeDict @system_configurations.route("/update_operation_type_status",methods=["POST","GET"]) def update_operation_type_status(): if not session.get("adminId"): return redirect("admin_login") operationTypeId = request.args.get("operationTypeId","") if operationTypeId: try: operation_type_queryset = OperationTypes.objects(id=operationTypeId,status__nin=[2]).first() if operation_type_queryset: if operation_type_queryset.status == 0: operation_type_queryset.update(status=1) flash("Operation type activated successfully!") elif operation_type_queryset.status == 1: operation_type_queryset.update(status=0) flash("Operation type deactivated successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="OperationType")) else: flash("Invaild id!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="OperationType")) except Exception as e: app.logger.error(traceback.format_exc()) return redirect(url_for("system_configurations.system_configurations_list",redirectTo="OperationType")) else: return redirect(url_for("system_configurations.system_configurations_list")) @system_configurations.route("/update_operation_type",methods=["POST","GET"]) def update_operation_type(): if not session.get("adminId"): return redirect("admin_login") try: operationTypeId = request.args.get("operationTypeId","") if request.method == "POST": name = request.form.get("name","") operation_type_queryset = OperationTypes.objects(id=operationTypeId).first() if operation_type_queryset: operation_type_queryset.update( name=name ) flash("Operation type updated successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="OperationType")) else: flash("Invaild id!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="OperationType")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to update operation type!!" return redirect(url_for("system_configurations.system_configurations_list",redirectTo="OperationType")) @system_configurations.route("/delete_operation_type",methods=["GET"]) def delete_operation_type(): try: if not session.get("adminId"): return redirect("admin_login") if request.method == "GET": operationTypeId = request.args.get("operationTypeId","") operation_type_queryset = OperationTypes.objects(id=operationTypeId,status__in=[0,1]).first() operation_type_queryset.update(status=2) flash("Operation type deleted successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="OperationType")) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to delete operation type!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="OperationType")) def fetching_csv_operation_type_details(operation_type_queryset): operationTypeDict = {} try: if operation_type_queryset.name: operationTypeDict["Name"] = operation_type_queryset.name else: operationTypeDict["Name"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return operationTypeDict @system_configurations.route("/operation_type_csv_list",methods=["POST","GET"]) def operation_type_csv_list(): if not session.get("adminId"): return redirect("admin_login") operationTypeList = [] adminId = session.get("adminId") try: operations_type_queryset = OperationTypes.objects(status__in=[0,1]).order_by("-id").all() for each_operation_type in operations_type_queryset: operationTypeDict = fetching_csv_operation_type_details(each_operation_type) operationTypeList.append(operationTypeDict) fieldnames = ['Name'] temp_csv_file_name = "/media/operation_type_csv_files/"+str(round(time.time() * 1000))+".csv" if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/operation_type_csv_files/")): os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/operation_type_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(operationTypeList) 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='Operation type list.csv', as_attachment=True ) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to export CSV data!!") return redirect("system_configurations.system_configurations_list") @system_configurations.route("/add_operation",methods=["POST","GET"]) def add_operation(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") if request.method == "POST": operationName = request.form.get("operationName","") operationTypeId = request.form.get("operationTypeId","") userTypeId = request.form.get("userTypeId","") channelId = request.form.get("channelId","") operationAuth = request.form.get("operationAuth","") code = request.form.get("code","") description = request.form.get("description","") isUI = request.form.get("isUI",False) isMakerChecker = request.form.get("isMakerChecker",False) if operationName and operationTypeId: try: operation_table = Operations( adminId=adminId, operationName = operationName, operationTypeId = operationTypeId, userTypeId = userTypeId, channelId = channelId, operationAuth = operationAuth, code = code, description = description, isUI = isUI, isMakerChecker = isMakerChecker, createdOn = datetime.datetime.now(), status = 1, ) save_table = operation_table.save() flash("Operation saved successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Operation")) except Exception as e: flash("Unable to save operation!!") app.logger.error(traceback.format_exc()) return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Operation")) else: flash("Required fields are missing!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Operation")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to save operation!!" return render_template("super_admin_templates/system_configuration_list.html",error=error) 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 @system_configurations.route("/update_operation_status",methods=["POST","GET"]) def update_operation_status(): if not session.get("adminId"): return redirect("admin_login") operationId = request.args.get("operationId","") if operationId: try: operation_queryset = Operations.objects(id=operationId,status__nin=[2]).first() if operation_queryset: if operation_queryset.status == 0: operation_queryset.update(status=1) flash("Operation activated successfully!") elif operation_queryset.status == 1: operation_queryset.update(status=0) flash("Operation deactivated successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Operation")) else: flash("Invaild id!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Operation")) except Exception as e: app.logger.error(traceback.format_exc()) return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Operation")) else: return redirect(url_for("system_configurations.system_configurations_list")) @system_configurations.route("/update_operation",methods=["POST","GET"]) def update_operation(): if not session.get("adminId"): return redirect("admin_login") try: operationId = request.args.get("operationId","") if request.method == "POST": operationName = request.form.get("operationName","") operationName = request.form.get("operationName","") operationTypeId = request.form.get("operationTypeId","") userTypeId = request.form.get("userTypeId","") channelId = request.form.get("channelId","") operationAuth = request.form.get("operationAuth","") code = request.form.get("code","") description = request.form.get("description","") isUI = request.form.get("isUI") isMakerChecker = request.form.get("isMakerChecker") operation_queryset = Operations.objects(id=operationId).first() if operation_queryset: operation_queryset.update( operationName = operationName, operationTypeId = ObjectId(operationTypeId), userTypeId = ObjectId(userTypeId), channelId = ObjectId(channelId), operationAuth = operationAuth, code = code, description = description, isUI = isUI, isMakerChecker = isMakerChecker, ) flash("Operation updated successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Operation")) else: flash("Invaild id!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Operation")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to update operation!!" return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Operation")) @system_configurations.route("/delete_operation",methods=["GET"]) def delete_operation(): try: if not session.get("adminId"): return redirect("admin_login") if request.method == "GET": operationId = request.args.get("operationId","") operation_queryset = Operations.objects(id=operationId,status__in=[0,1]).first() operation_queryset.update(status=2) flash("Operation deleted successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Operation")) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to delete operation!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="Operation")) def fetching_csv_operation_type_details(operation_queryset): operationDict = {} try: if operation_queryset.operationName: operationDict["Name"] = operation_queryset.operationName else: operationDict["Name"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return operationDict @system_configurations.route("/operation_csv_list",methods=["POST","GET"]) def operation_csv_list(): if not session.get("adminId"): return redirect("admin_login") operationList = [] adminId = session.get("adminId") try: operations_type_queryset = Operations.objects(status__in=[0,1]).order_by("-id").all() for each_operation in operations_type_queryset: operationDict = fetching_csv_operation_type_details(each_operation) operationList.append(operationDict) fieldnames = ['Name'] temp_csv_file_name = "/media/operation_csv_files/"+str(round(time.time() * 1000))+".csv" if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/operation_csv_files/")): os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/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(operationList) save_csv_file = CSVFileExports( operationName=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='Operation list.csv', as_attachment=True ) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to export CSV data!!") return redirect("system_configurations.system_configurations_list") @system_configurations.route("/add_error_defination",methods=["POST","GET"]) def add_error_defination(): try: if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") if request.method == "POST": typeId = request.form.get("typeId","") errorCode = request.form.get("errorCode","") errorMessage = request.form.get("errorMessage","") resolution = request.form.get("resolution","") description = request.form.get("description","") if typeId and errorCode and errorMessage and resolution: try: error_def_table = ErrorDefinations( adminId=adminId, typeId = typeId, errorCode = errorCode, errorMessage = errorMessage, resolution = resolution, description = description, createdOn = datetime.datetime.now(), status = 1, ) save_table = error_def_table.save() flash("Error defination saved successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="ErrorDefination")) except Exception as e: flash("Unable to save error defination!!") app.logger.error(traceback.format_exc()) return redirect(url_for("system_configurations.system_configurations_list",redirectTo="ErrorDefination")) else: flash("Required fields are missing!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="ErrorDefination")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to save error defination!!" return render_template("super_admin_templates/system_configuration_list.html",error=error) def fetching_error_defination_details(error_defination_queryset): errorDict = {} try: errorDict={ "id":str(error_defination_queryset.id), "typeId":error_defination_queryset.typeId, "errorCode":error_defination_queryset.errorCode, "errorMessage":error_defination_queryset.errorMessage, "resolution":error_defination_queryset.resolution, "description":error_defination_queryset.description } if error_defination_queryset.status==1: errorDict["actionText"] = "Active" else: errorDict["actionText"] = "Deactive" if error_defination_queryset.createdOn: errorDict["createdOn"] = error_defination_queryset.createdOn.strftime("%m-%d-%Y") else: errorDict["createdOn"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return errorDict @system_configurations.route("/update_error_defination_status",methods=["POST","GET"]) def update_error_defination_status(): if not session.get("adminId"): return redirect("admin_login") errorDefinationId = request.args.get("errorDefinationId","") if errorDefinationId: try: error_defination_queryset = ErrorDefinations.objects(id=errorDefinationId,status__nin=[2]).first() if error_defination_queryset: if error_defination_queryset.status == 0: error_defination_queryset.update(status=1) flash("Error defination activated successfully!") elif error_defination_queryset.status == 1: error_defination_queryset.update(status=0) flash("Error defination deactivated successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="ErrorDefination")) else: flash("Invaild id!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="ErrorDefination")) except Exception as e: app.logger.error(traceback.format_exc()) return redirect(url_for("system_configurations.system_configurations_list",redirectTo="ErrorDefination")) else: return redirect(url_for("system_configurations.system_configurations_list")) @system_configurations.route("/update_error_defination",methods=["POST","GET"]) def update_error_defination(): if not session.get("adminId"): return redirect("admin_login") try: errorDefinationId = request.args.get("errorDefinationId","") if request.method == "POST": typeId = request.form.get("typeId","") errorCode = request.form.get("errorCode","") errorMessage = request.form.get("errorMessage","") resolution = request.form.get("resolution","") description = request.form.get("description","") error_defination_queryset = ErrorDefinations.objects(id=errorDefinationId).first() if error_defination_queryset: error_defination_queryset.update( typeId = typeId, errorCode = errorCode, errorMessage = errorMessage, resolution = resolution, description = description ) flash("Error defination updated successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="ErrorDefination")) else: flash("Invaild id!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="ErrorDefination")) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to update error defination!!" return redirect(url_for("system_configurations.system_configurations_list",redirectTo="ErrorDefination")) @system_configurations.route("/delete_error_defination",methods=["GET"]) def delete_error_defination(): try: if not session.get("adminId"): return redirect("admin_login") if request.method == "GET": errorDefinationId = request.args.get("errorDefinationId","") error_defination_queryset = ErrorDefinations.objects(id=errorDefinationId,status__in=[0,1]).first() error_defination_queryset.update(status=2) flash("Error defination deleted successfully!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="ErrorDefination")) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to delete error defination!!") return redirect(url_for("system_configurations.system_configurations_list",redirectTo="ErrorDefination")) def fetching_csv_error_defination_type_details(error_defination_queryset): errorDict = {} try: if error_defination_queryset.typeId: errorDict["Name"] = error_defination_queryset.typeId else: errorDict["Name"] = "" except Exception as e: app.logger.error(traceback.format_exc()) return errorDict @system_configurations.route("/error_defination_csv_list",methods=["POST","GET"]) def error_defination_csv_list(): if not session.get("adminId"): return redirect("admin_login") errorList = [] adminId = session.get("adminId") try: error_definations_queryset = ErrorDefinations.objects(status__in=[0,1]).order_by("-id").all() for each_error in error_definations_queryset: errorDict = fetching_csv_error_defination_type_details(each_error) errorList.append(errorDict) fieldnames = ['Name'] temp_csv_file_name = "/media/error_defination_csv_files/"+str(round(time.time() * 1000))+".csv" if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/error_defination_csv_files/")): os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/error_defination_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(errorList) save_csv_file = CSVFileExports( typeId=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='Error defination list.csv', as_attachment=True ) except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to export CSV data!!") return redirect("system_configurations.system_configurations_list")