from appservices.common.util import * from appservices.common.form_schemas import * bank_masters = Blueprint("bank_masters",__name__) ############### Master IFSC Bank ################ # @bank_masters.route("/add_master_ifsc_bank",methods=["POST","GET"]) # @adminid_access_token_required # @csrf_protect # def add_master_ifsc_bank(): # try: # if not session.get("adminId"): # return redirect("admin_login") # adminId = session.get("adminId") # csrf_token = request.form.get("csrf_token") # print(csrf_token,"((((((((((((((csrf_token))))))))))))))") # loginBrowser = request.headers.get("Sec-Ch-Ua") # if loginBrowser: # loginBrowseData = loginBrowser.split(";") # browser = loginBrowseData[0] # else: # loginBrowseData = request.headers.get('User-Agent').split(";") # browser = loginBrowseData[0] # client_ip=0 # # Extracting client IP address # if request.headers.getlist("X-Forwarded-For"): # client_ip = request.headers.getlist("X-Forwarded-For")[0] # else: # client_ip = request.remote_addr # actionDate=datetime.datetime.now() # permissionsList = check_permissions(adminId,"bankPermissions") # if "add" in permissionsList: # if request.method == "POST": # bankName = request.form.get("bankName","") # bankCode = request.form.get("bankCode","") # ifscCode = request.form.get("ifscCode","") # priority = request.form.get("priority","") # image = request.files.get("image","") # remark = request.files.get("remark","") # jsonData = request.form.to_dict(flat=True) # requestData = [jsonData] # updatedrequestData = [jsonData] # form = BankForm() # if form.validate_on_submit(): # # if bankName and not is_valid_alphanumeric(bankName): # # flash("Bank Name must contain only alphanumeric characters, spaces, and specific special characters:@#$()+_-/") # # return redirect(url_for("bank_masters.get_bank_masters_list", redirectTo="masterIfscBank")) # # if bankCode and not is_valid_alphanumeric_without_specialCharacters(bankCode): # # flash("Bank Code must contain only alphanumeric characters!") # # return redirect(url_for("bank_masters.get_bank_masters_list", redirectTo="masterIfscBank")) # # if ifscCode and ifscCode and not is_valid_ifsc_code(ifscCode): # # flash("IFSC Code must follow the format: 4 letters, 1 digit, and 6 alphanumeric characters!") # # return redirect(url_for("bank_masters.get_bank_masters_list", redirectTo="masterIfscBank")) # # if priority and not is_valid_numeric(priority): # # flash("Priority must contain only numeric characters!") # # return redirect(url_for("bank_masters.get_bank_masters_list", redirectTo="masterIfscBank")) # # if image: # # if not is_allowed_image(image.filename): # # flash("Invalid image format! Only JPG, JPEG, and PNG are allowed.") # # return redirect(url_for("bank_masters.get_bank_masters_list", redirectTo="masterIfscBank")) # # if remark and not is_valid_alphanumeric(remark): # # flash("Bank remark must contain only alphanumeric characters, spaces, and specific special characters:@#$()+_-/") # # return redirect(url_for("bank_masters.get_bank_masters_list", redirectTo="masterIfscBank")) # if bankName and bankCode and priority: # try: # admin_queryset = SuperAdmin.objects(id=adminId,status=1).first() # if admin_queryset: # message=admin_queryset.userName+" "+bankName+" bank created successfully!" # save_admin_log_table = save_admin_logs_data(adminId,None,None,"add_master_ifsc_bank","create",actionDate,client_ip,browser,message,requestData,updatedrequestData) # master_ifsc_table = MasterIFSCBank( # adminId=adminId, # bankName = bankName, # bankCode = bankCode, # ifscCode = ifscCode, # priority = priority, # remark = remark, # createdOn = datetime.datetime.now(), # status = 1, # ) # save_table = master_ifsc_table.save() # masterIfscBankId = str(save_table.id) # if image: # if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/masterIfscBank/")): # os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/masterIfscBank/")) # save_file = image.save(os.path.join(app.config['SITE_ROOT'], "media/masterIfscBank/"+masterIfscBankId+".png")) # save_table.update(image=save_file) # flash("Master bank saved successfully!") # return redirect(url_for("bank_masters.get_bank_masters_list",redirectTo="masterIfscBank")) # except Exception as e: # flash("Unable to save master bank details!!") # app.logger.error(traceback.format_exc()) # return redirect(url_for("bank_masters.get_bank_masters_list",redirectTo="masterIfscBank")) # else: # flash("Required fields are missing!!") # return redirect(url_for("bank_masters.get_bank_masters_list",redirectTo="masterIfscBank")) # else: # for fieldName, errorMessages in form.errors.items(): # print("errorMessages",form.errors) # for err in errorMessages: # flash(f"{fieldName}: {err}", "danger") # return redirect(url_for("bank_masters.get_bank_masters_list",redirectTo="masterIfscBank")) # else: # return redirect(url_for("bank_masters.get_bank_masters_list",redirectTo="masterIfscBank")) # else: # flash("Staff member does not have given create bank permissions!!") # return redirect(url_for("bank_masters.get_bank_masters_list",redirectTo="masterIfscBank")) # except Exception as e: # app.logger.error(traceback.format_exc()) # error = "Unable to save master bank details!!" # return render_template("super_admin_templates/bank_masters_list.html",error=error,redirectTo="masterIfscBank") @bank_masters.route("/add_master_ifsc_bank",methods=["POST","GET"]) @adminid_access_token_required @csrf_protect def add_master_ifsc_bank(): data_status={"responseStatus":0,"result":""} try: if not session.get("adminId"): flash("Session expired Please login again.","danger") data_status["responseStatus"]=4 return data_status adminId = session.get("adminId") csrf_token = request.form.get("csrf_token") print(csrf_token,"((((((((((((((csrf_token))))))))))))))") latitude = request.form.get("latitude", "") longitude = request.form.get("longitude", "") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get('User-Agent').split(";") browser = loginBrowseData[0] client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate=datetime.datetime.now() permissionsList = check_permissions(adminId,"bankPermissions") if "add" in permissionsList: if request.method == "POST": bankName = request.form.get("bankName","") bankCode = request.form.get("bankCode","") ifscCode = request.form.get("ifscCode","") priority = request.form.get("priority","") image = request.files.get("image","") remark = request.files.get("remark","") jsonData = request.form.to_dict(flat=True) requestData = [jsonData] updatedrequestData = [jsonData] form = BankForm() if form.validate_on_submit(): if bankName and bankCode and priority: try: admin_queryset = SuperAdmin.objects(id=adminId,status=1).first() if admin_queryset: message=admin_queryset.userName+" "+bankName+" bank created successfully!" save_admin_log_table = save_admin_logs_data(adminId,None,None,"add_master_ifsc_bank","create",actionDate,client_ip,browser,message,requestData,updatedrequestData,latitude,longitude) master_ifsc_table = MasterIFSCBank( adminId=adminId, bankName = bankName, bankCode = bankCode, ifscCode = ifscCode, priority = priority, remark = remark, createdOn = datetime.datetime.now(), status = 1, ) save_table = master_ifsc_table.save() masterIfscBankId = str(save_table.id) if image: if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/masterIfscBank/")): os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/masterIfscBank/")) save_file = image.save(os.path.join(app.config['SITE_ROOT'], "media/masterIfscBank/"+masterIfscBankId+".png")) save_table.update(image=save_file) flash("Master bank saved successfully", "success") data_status['responseStatus']=1 return data_status except Exception as e: flash("Unable to save master bank details.", "danger") data_status['responseStatus']=4 return data_status else: data_status['responseStatus']=2 data_status['result']="Required fields are missing!!" return data_status else: data_status['result']=form.errors return data_status else: flash("Invalid request.","danger") data_status['responseStatus']=4 return data_status else: flash("The staff member does not have permission to create a bank.", "danger") data_status['responseStatus']=4 return data_status except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to save master bank details!!" flash("Unable to save master bank details.", "danger") data_status['responseStatus']=4 return data_status #Update master IFSC bank @bank_masters.route("/update_master_ifsc_bank",methods=["POST","GET"]) @csrf_protect @adminid_access_token_required def update_master_ifsc_bank(): data_status={"responseStatus":0,"result":""} if not session.get("adminId"): flash("Session expired. Please log in again.", "danger") data_status["responseStatus"]=4 return data_status adminId=session.get("adminId") csrf_token = request.form.get("csrf_token") print(csrf_token,"((((((((((((((csrf_token))))))))))))))") latitude = request.form.get("latitude", "") longitude = request.form.get("longitude", "") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get('User-Agent').split(";") browser = loginBrowseData[0] existing_record = "" client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate=datetime.datetime.now() print(request.headers,"(((((((((((((((((request.headers)))))))))))))))))") try: permissionsList = check_permissions(session.get("adminId"),"bankPermissions") if "edit" in permissionsList: masterIfscBankId = request.args.get("masterIfscBankId","") if request.method == "POST": bankName = request.form.get("bankName","") bankCode = request.form.get("bankCode","") ifscCode = request.form.get("ifscCode","") priority = request.form.get("priority","") image = request.files.get("image","") remark = request.form.get("remark","") jsonData = request.form.to_dict(flat=True) print("(((((((((((((((((((((((((((bankName)))))))))))))))))))))))))))",bankName) # in your route form = BankForm(request.form, current_id=masterIfscBankId) if form.validate_on_submit(): if bankName and bankCode and priority: master_ifsc_queryset = MasterIFSCBank.objects(id=masterIfscBankId,status__in=[0,1]).first() existing_record = master_ifsc_queryset.to_json() message=master_ifsc_queryset.adminId.userName+" "+bankName+" bank updated successfully!" requestData=[master_ifsc_queryset] updatedrequestData=[jsonData] save_admin_log_table = save_admin_logs_data(adminId,None,None,"update_master_ifsc_bank","update",actionDate,client_ip,browser,message,requestData,updatedrequestData,latitude,longitude) if master_ifsc_queryset: master_ifsc_queryset.update( bankName = bankName, bankCode = bankCode, ifscCode = ifscCode, priority = priority, remark = remark, ) if image: if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/masterIfscBank/")): os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/masterIfscBank/")) save_file = image.save(os.path.join(app.config['SITE_ROOT'], "media/masterIfscBank/"+masterIfscBankId+".png")) master_ifsc_queryset.update(image=save_file) flash("Master bank updated successfully!", "success") data_status["responseStatus"]=1 return data_status else: flash("Invalid ID!", "danger") data_status["responseStatus"]=4 return data_status else: data_status["responseStatus"]=2 data_status["result"]="Required fields are missing." return data_status else: data_status["result"]=form.errors return data_status else: flash("Invalid request.", "danger") data_status["responseStatus"]=4 return data_status else: flash("The staff member does not have permission to update bank details.", "danger") data_status["responseStatus"]=4 return data_status except Exception as e: app.logger.error(traceback.format_exc()) flash("Unable to update master bank details.", "danger") data_status["responseStatus"]=4 return data_status # @bank_masters.route("/user_bank_csv_list",methods=["POST","GET"]) # def user_bank_csv_list(): # if not session.get("adminId"): # return redirect("admin_login") # companyBanksList = [] # adminId = session.get("adminId") # try: # company_banks_queryset = CompanyBankAccounts.objects(status__in=[0,1]).order_by("-id") # for each_company_bank in company_banks_queryset: # bank_dict = fetching_account_bank_details(each_company_bank) # companyBanksList.append(bank_dict) # fieldnames = ['Bank Name', 'Bank Code', 'Acc Name', 'Acc Number', 'IFSC Code', 'Branch'] # temp_csv_file_name = "/media/userBankCsvFiles/"+str(round(time.time() * 1000))+".csv" # if not os.path.exists(os.path.join(app.config['SITE_ROOT'], "media/userBankCsvFiles/")): # os.makedirs(os.path.join(app.config['SITE_ROOT'], "media/userBankCsvFiles/")) # 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(regionsList) # 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 Banks list.csv', # as_attachment=True # ) # except Exception as e: # app.logger.error(traceback.format_exc()) # flash("Unable to export CSV data!!") # return redirect("bank_masters.get_bank_masters_list") # def fetching_account_bank_details(bank_queryset): # bank_dict = {} # try: # if bank_queryset.bankName: # bank_dict["Bank Name"] = bank_queryset.bankName # else: # bank_dict["Bank Name"] = "" # if bank_queryset.bankCode: # bank_dict["Bank Code"] = bank_queryset.bankCode # else: # bank_dict["Bank Code"] = "" # if bank_queryset.accountName: # bank_dict["Acc Name"] = bank_queryset.accountName # else: # bank_dict["Acc Name"] = "" # if bank_queryset.accountNumber: # bank_dict["Acc Number"] = bank_queryset.accountNumber # else: # bank_dict["Acc Number"] = "" # if bank_queryset.ifscCode: # bank_dict["IFSC Code"] = bank_queryset.ifscCode # else: # bank_dict["IFSC Code"] = "" # if bank_queryset.branch: # bank_dict["Branch"] = bank_queryset.branch # else: # bank_dict["Branch"] = "" # except Exception as e: # app.logger.error(traceback.format_exc()) # return bank_dict @bank_masters.route("/get_bank_masters_list",methods=["POST","GET"]) @adminid_access_token_required def get_bank_masters_list(): if not session.get("adminId"): flash("Session expired. Please log in again.", "danger") return redirect(url_for("admin.admin_logout")) masterIfscBanksList = [] adminId = session.get("adminId") pagination = None search_element="" snoCount=0 bankPage=None permissionsList = check_permissions(session.get("adminId"),"bankPermissions") if "view" in permissionsList: try: redirectTo = request.args.get("redirectTo","masterIfscBank") if redirectTo: redirectval = redirectTo else: redirectval = "masterIfscBank" form = BankFormSearch(request.args) search_element = request.args.get("search_element", "").strip() # if search_element: # search_element = form.search_element.data.strip() # else: # if form.search_element.errors: # flash(" | ".join(form.search_element.errors), "danger") # search_element = "" bankPage = request.args.get(get_page_parameter("bankPage"), type=int, default=1) per_page = 20 start = (bankPage - 1) * per_page total_count=0 filters = Q(status__in=[0, 1]) print("(((((((((((((((((((search_element)))))))))))))))))))",search_element) if form.validate(): if search_element and search_element.strip(): filters &= Q(bankName__icontains=search_element.strip()) # start_time = datetime.datetime.now() total_count = MasterIFSCBank.objects(filters).count() # end_time = datetime.datetime.now() # print("Query Time:", (end_time - start_time).total_seconds(), "seconds") # print(MasterIFSCBank._get_collection().index_information()) # total_pages = math.ceil(total_count / per_page) if total_count else 1 # if total_count == 0 or bankPage > total_pages or bankPage < 1: # bankPage=1 # start =0 # else: # start = (bankPage - 1) * per_page master_ifsc_banks_queryset = ( MasterIFSCBank.objects(filters) .only("id", "bankName", "bankCode", "ifscCode", "status","priority") .order_by("-id") .skip(start) .limit(per_page) ) masterIfscBanksList=list(master_ifsc_banks_queryset) else: masterIfscBanksList=[] snoCount = start # print(listdata,"listdata") # 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 = master_ifsc_banks_queryset.count() # start = (page - 1) * per_page # end = min(start + per_page, total_count) # total_master_banks = master_ifsc_banks_queryset[start:end] # snoCount = start # for each_master_ifsc_bank in total_master_banks: # snoCount +=1 # master_ifsc_bank_dict = fetching_master_ifsc_bank_details(each_master_ifsc_bank) # master_ifsc_bank_dict["snoCount"]=snoCount # masterIfscBanksList.append(master_ifsc_bank_dict) # Pagination object for rendering pagination controls in the template pagination = Pagination(bankPage=bankPage, total=total_count, per_page=per_page,page_parameter ="bankPage", alignment="right", record_name="banks" ,href=f"?search_element={search_element}&bankPage={{0}}") return render_template("super_admin_templates/bank_masters_list.html", pagination=pagination, masterIfscBanksList=masterIfscBanksList, redirectval=redirectval, search_element=search_element, snoCount=snoCount, bankPage = bankPage, form=form, ) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to fetch bank master details." return render_template("super_admin_templates/bank_masters_list.html", error=error, pagination=pagination, masterIfscBanksList=masterIfscBanksList, search_element=search_element, snoCount=snoCount, bankPage = bankPage, form=form, ) else: flash("You do not have permission to view bank details.", "danger") return redirect(url_for("admin.dashboard")) # return render_template("super_admin_templates/bank_masters_list.html") @bank_masters.route("/update_bank_status",methods=["POST","GET"]) @adminid_access_token_required @csrf_protect def update_bank_status(): data_status={"responseStatus":0,"result":""} if not session.get("adminId"): return redirect("admin_login") adminId=session.get("adminId") csrf_token = request.form.get("csrf_token") print(csrf_token,"((((((((((((((csrf_token))))))))))))))") bankPage = request.args.get("bankPage") search_element = request.args.get("search_element", "").strip() latitude = request.form.get("latitude", "") longitude = request.form.get("longitude", "") loginBrowser = request.headers.get("Sec-Ch-Ua") if loginBrowser: loginBrowseData = loginBrowser.split(";") browser = loginBrowseData[0] else: loginBrowseData = request.headers.get('User-Agent').split(";") browser = loginBrowseData[0] client_ip=0 # Extracting client IP address if request.headers.getlist("X-Forwarded-For"): client_ip = request.headers.getlist("X-Forwarded-For")[0] else: client_ip = request.remote_addr actionDate=datetime.datetime.now() jsonData = request.form.to_dict(flat=True) existing_record = "" updatedrequestData = [jsonData] form=RemarkForm() if form.validate_on_submit(): permissionsList = check_permissions(session.get("adminId"),"bankPermissions") if "edit" in permissionsList: masterIfscBankId = request.args.get("masterIfscBankId","") remark = request.form.get("remark","") # if remark and not is_valid_alphanumeric(remark): # flash("Remark contain's only alphanumeric characters, spaces, and specific special characters:@#$()+_-/","danger") # return redirect(url_for("bank_masters.get_bank_masters_list", redirectTo="masterIfscBank")) if masterIfscBankId and remark: try: message="" master_ifsc_banks_queryset = MasterIFSCBank.objects(id=masterIfscBankId).first() if master_ifsc_banks_queryset: existing_record = master_ifsc_banks_queryset.to_json() requestData = [existing_record] if master_ifsc_banks_queryset.status == 0: master_ifsc_banks_queryset.update(status=1, remark=remark) flash("Bank activated successfully.", "success") message=master_ifsc_banks_queryset.adminId.userName+" "+master_ifsc_banks_queryset.bankName+" bank activated successfully!" elif master_ifsc_banks_queryset.status == 1: master_ifsc_banks_queryset.update(status=0,remark=remark) flash("Bank deactivated successfully.", "success") message=master_ifsc_banks_queryset.adminId.userName+" "+master_ifsc_banks_queryset.bankName+" bank deactivated successfully!" save_admin_log_table = save_admin_logs_data(adminId,None,None,"update_bank_status","updatestatus",actionDate,client_ip,browser,message,requestData,updatedrequestData,latitude,longitude) data_status['responseStatus']=1 return data_status else: flash("Invaild Id!","danger") data_status['responseStatus']=4 return data_status except Exception as e: flash("Unable to update the status", "danger") app.logger.error(traceback.format_exc()) data_status['responseStatus']=4 return data_status else: data_status['responseStatus']=2 data_status['result']="Required fields are missing!!" return data_status else: flash("The staff member does not have permission to update bank details.", "danger") data_status['responseStatus']=4 return data_status else: data_status['result']=form.errors return data_status # @bank_masters.route("/change_active_banks_beneficiaries",methods=["GET"]) # def change_active_banks_beneficiaries(): # data_status = {"responseStatus":0,"result":""} # try: # benificiaries_queryset = Beneficiaries.objects(status=1).order_by("-id") # for each_benificiary in benificiaries_queryset: # masterBankId = each_benificiary.masterBankId # beneCodeBankId = each_benificiary.beneCodeList[0]["bankId"] # print(masterBankId, "((((((masterBankId))))))") # print(beneCodeBankId, "((((((beneCodeBankId))))))") # if masterBankId and beneCodeBankId: # print("If condition is successfully passed!") # bank_queryset = MasterIFSCBank.objects(id=masterBankId).first() # if bank_queryset and bank_queryset.status == 2: # name_match_bank_queryset = MasterIFSCBank.objects( # bankName__iexact=str(bank_queryset.bankName), # ifscCode__exists=True, # Ensures the field exists # ifscCode__ne=None # Ensures it is not None # ).first() # if name_match_bank_queryset: # active_bank_queryset = MasterIFSCBank.objects(id=str(name_match_bank_queryset.id), status=1).first() # if active_bank_queryset: # activeBankId = str(active_bank_queryset.id) # # Update masterBankId and beneCodeList[0]["bankId"] # beneCodeList = each_benificiary.beneCodeList # beneCodeList[0]["bankId"] = str(activeBankId) # Update bankId # each_benificiary.update( # set__masterBankId=ObjectId(activeBankId), # set__beneCodeList=beneCodeList # ) # return "Success" # except Exception as e: # app.logger.error(traceback.format_exc()) # return "Unsucess!!" @bank_masters.route("/change_active_banks_beneficiaries", methods=["GET"]) @adminid_access_token_required def change_active_banks_beneficiaries(): data_status = {"responseStatus": 0, "result": ""} try: benificiaries_queryset = Beneficiaries.objects(status=1).order_by("-id") for each_benificiary in benificiaries_queryset: masterBankId = str(each_benificiary.masterBankId.id) print(masterBankId, "((((((masterBankId))))))") if masterBankId: print("If condition is successfully passed!") bank_queryset = MasterIFSCBank.objects(id=masterBankId).first() if bank_queryset and bank_queryset.status == 2: name_match_bank_queryset = MasterIFSCBank.objects( bankName__iexact=str(bank_queryset.bankName), ifscCode__exists=True, # Ensures the field exists ifscCode__ne=None # Ensures it is not None ).first() if name_match_bank_queryset: active_bank_queryset = MasterIFSCBank.objects(id=str(name_match_bank_queryset.id), status=1).first() if active_bank_queryset: activeBankId = str(active_bank_queryset.id) each_benificiary.update( set__masterBankId=ObjectId(activeBankId) ) return "Success" except Exception as e: app.logger.error(traceback.format_exc()) return "Unsuccess!!"