from appservices.common.util import * from appservices.common.form_schemas import * admin_logs = Blueprint("admin_logs",__name__) @admin_logs.route("/sms_logs_list",methods=["POST","GET"]) @adminid_access_token_required def sms_logs_list(): if not session.get("adminId"): return redirect("admin_login") smsLogsList = [] pagination = None form = SmsLogsForm(request.args) adminId = session.get("adminId") permissionsList = check_permissions(session.get("adminId"),"smslogsPermissions") if "view" in permissionsList: try: name = request.args.get('name', '').strip() phoneNumber = request.args.get('phoneNumber', '').strip() # Get the current page from the query parameters page = request.args.get("page", type=int, default=1) per_page = 20 # Number of items per page start = (page - 1) * per_page total_count = 0 filters = Q(status__in=[0,1]) if form.validate(): if phoneNumber: filters &= Q(phoneNumber__icontains=phoneNumber) if name: matching_users = Users.objects(fullName__icontains=name).only('id') user_ids = [user.id for user in matching_users] matching_admins = SuperAdmin.objects(userName__icontains=name).only('id') admin_ids = [user.id for user in matching_admins] filters&=Q(userId__in=user_ids) | Q(adminId__in=admin_ids) # if user_ids: # filters&=Q(userId__in=user_ids) # elif admin_ids: # filters&=Q(adminId__in=admin_ids) # else: # filters&=Q(userId__in=[]) # # if name: # matching_users = Users.objects(fullName__icontains=name).only('id') # matching_admins = SuperAdmin.objects(userName__icontains=name).only('id') total_count = OtpChecks.objects(filters).count() otp_checks_queryset = ( OtpChecks.objects(filters) .only("id","otpCode","phoneNumber","attempts", "otpReason","createdOn","userId","adminId") .order_by("-id") .skip(start) .limit(per_page) .select_related() ) snoCount = start for each_otp_check in otp_checks_queryset: snoCount +=1 smsLogDict = { "id":str(each_otp_check.id), "otpCode":each_otp_check.otpCode, "phoneNumber":each_otp_check.phoneNumber, "attempts":each_otp_check.attempts, "otpReason":each_otp_check.otpReason, "createdOn":each_otp_check.createdOn } try: if each_otp_check.userId: smsLogDict["name"]=each_otp_check.userId.fullName elif each_otp_check.adminId: smsLogDict["name"]=each_otp_check.adminId.userName else: smsLogDict["name"]="" except Exception as e: smsLogDict["name"]="" smsLogDict["snoCount"]=snoCount smsLogsList.append(smsLogDict) pagination = Pagination( page=page, total=total_count, record_name='sms_logs_list', per_page=per_page, alignment="right") else: print("form errors",form.errors) return render_template("super_admin_templates/sms_logs_list.html", pagination=pagination, smsLogsList=smsLogsList, name = name, phoneNumber = phoneNumber, form = form ) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to fetched sms logs data!!" return render_template("super_admin_templates/sms_logs_list.html", error=error, pagination=pagination, smsLogsList=smsLogsList, name = name, phoneNumber = phoneNumber, form = form ) else: flash("The staff member does not have permission to view sms logs.", "danger") return redirect(url_for("admin.dashboard")) @admin_logs.route("/merchant_logs_list",methods=["POST","GET"]) @adminid_access_token_required def merchant_logs_list(): if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") merchantLogsList = [] merchantsList = [] pagination = None OS= ["Android","IOS","Web"] form = MerchantLogsForm(request.args) snoCount = 0 permissionsList = check_permissions(session.get("adminId"),"merchantLogsPermissions") if "view" in permissionsList: try: name = request.args.get('name', '').strip() phoneNumber = request.args.get('phoneNumber', '').strip() selectOs = request.args.get('selectOs', '').strip() page = request.args.get("page", type=int, default=1) per_page = 20 # Number of items per page start = (page - 1) * per_page total_count = 0 filters = Q(status__in=[0,1]) merchants_queryset = Users.objects(status__nin=[2]).only("id","fullName","phoneNumber").order_by("-createdOn") merchantsList=list(merchants_queryset) if selectOs in ["Android", "IOS","Web"]: filters &= Q(os=selectOs) if form.validate(): if phoneNumber: matching_users = Users.objects(phoneNumber__icontains=phoneNumber).only('id') user_ids = [user.id for user in matching_users] filters&=Q(userId__in=user_ids) if name: filters &= Q(userId__in=[name]) total_count = LoginReportLogs.objects(filters).count() merchant_logs_queryset = ( LoginReportLogs.objects(filters) .only("id","userId","phoneNumber","otpCode","attempts","ipAddress","latitude","longitude","os","userAgent","loginBrowser","deviceName","createdOn") .order_by("-id") .skip(start) .limit(per_page) ) merchantLogsList = list(merchant_logs_queryset) snoCount = start pagination = Pagination(page=page, total=total_count, per_page=per_page, alignment="right", record_name="merchant_logs") else: print("form errors",form.errors) return render_template("super_admin_templates/merchant_logs_list.html", pagination=pagination, merchantLogsList=merchantLogsList, snoCount=snoCount, name = name, phoneNumber = phoneNumber, selectOs = selectOs, merchantsList = merchantsList, form = form, OS = OS ) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to fetched merchant logs data!!" return render_template("super_admin_templates/merchant_logs_list.html", error=error, pagination=pagination, merchantLogsList=merchantLogsList, snoCount=snoCount, name = name, phoneNumber = phoneNumber, selectOs = selectOs, merchantsList=merchantsList, form = form, OS = OS ) else: flash("The staff member does not have permission to view merchant logs.", "danger") return redirect(url_for("admin.dashboard")) @admin_logs.route("/admin_otp_logs_list",methods=["POST","GET"]) @adminid_access_token_required def admin_otp_logs_list(): if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") adminLogsList = [] form = AdminOtpLogsForm(request.args) pagination = None permissionsList = check_permissions(session.get("adminId"),"adminOtpLogsPermissions") if "view" in permissionsList: try: name = request.args.get('name', '').strip() page = request.args.get("page", type=int, default=1) per_page = 20 # Number of items per page start = (page - 1) * per_page total_count = 0 filters = Q(status__in=[0,1]) if form.validate(): if name: matching_admins = SuperAdmin.objects(userName__icontains=name).only('id') admin_ids = [user.id for user in matching_admins] filters&=Q(adminId__in=admin_ids) total_count = SuperAdminLogs.objects(filters).count() admin_logs_queryset = ( SuperAdminLogs.objects(filters) .only("id","otpCode","attempts","ipAddress","userAgent","loginBrowser","createdOn","adminId") .order_by("-id") .skip(start) .limit(per_page) ) snoCount = start for each_admin_log in admin_logs_queryset: snoCount +=1 adminLogDict = { "id":str(each_admin_log.id), # "adminName":each_admin_log.adminId.userName, "otpCode":each_admin_log.otpCode, "attempts":each_admin_log.attempts, "ipAddress":each_admin_log.ipAddress, "userAgent":each_admin_log.userAgent, "loginBrowser":each_admin_log.loginBrowser, "createdOn":each_admin_log.createdOn.astimezone(ist_timezone).strftime("%d-%m-%Y %I:%M %p") } try: if each_admin_log.adminId: adminLogDict["adminName"]=each_admin_log.adminId.userName else: adminLogDict["adminName"]="" except Exception as e: adminLogDict["adminName"]="" adminLogDict["snoCount"]=snoCount adminLogsList.append(adminLogDict) # Pagination object for rendering pagination controls in the template pagination = Pagination(page=page, total=total_count, per_page=per_page, alignment="right", record_name="admin_logs") else: print("form errors",form.errors) return render_template("super_admin_templates/admin_otp_logs_list.html", pagination=pagination, adminLogsList=adminLogsList, name = name, form = form ) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to fetched admin logs data!!" return render_template("super_admin_templates/admin_otp_logs_list.html", error=error, pagination=pagination, adminLogsList=adminLogsList, name = name, form = form ) else: flash("The staff member does not have permission to view admin otp logs.", "danger") return redirect(url_for("admin.dashboard")) @admin_logs.route("/api_logs_list",methods=["POST","GET"]) @adminid_access_token_required def api_logs_list(): if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") apiLogsList = [] pagination = None form = ApiLogsForm(request.args) permissionsList = check_permissions(session.get("adminId"),"apiLogsPermissions") if "view" in permissionsList: try: walletType = request.args.get('walletType', '').strip() page = request.args.get("page", type=int, default=1) per_page = 20 # Number of items per page start = (page - 1) * per_page total_count = 0 filters = Q(status__in=[0,1]) if walletType in ["payin", "payout"]: filters &= Q(walletType=walletType) if form.validate(): total_count = ApiLogs.objects(filters).count() api_logs_queryset = ( ApiLogs.objects(filters) .only("id","walletType","apiCallingType","apiCallingUrl","transactionAPIId","requestData","responseData","clientIp","platformType","createdOn","requestDate","responseDate") .order_by("-id") .skip(start) .limit(per_page) ) snoCount = start for each_api_logs in api_logs_queryset: snoCount +=1 apiLogDict = { "id":str(each_api_logs.id), "walletType":each_api_logs.walletType, "transactionAPIId":each_api_logs.transactionAPIId, "apiCallingType":each_api_logs.apiCallingType, "apiCallingUrl":each_api_logs.apiCallingUrl, "requestData":each_api_logs.requestData, "responseData":each_api_logs.responseData, "clientIp":each_api_logs.clientIp, "platformType":each_api_logs.platformType, "createdOn":each_api_logs.createdOn.astimezone(ist_timezone).strftime("%d-%m-%Y %I:%M %p") } requestList = [] for each_dict in each_api_logs.requestData: dict_items_data = each_dict.items() for each_key,each_value in dict_items_data: requestDict = { "keyName":each_key, "valueName":each_value } requestList.append(requestDict) apiLogDict["requestList"]=requestList responseList = [] for each_dict in each_api_logs.responseData: for each_key in each_dict.keys(): responseDict ={ "keyName":each_key, "valueName":each_dict[each_key] } responseList.append(responseDict) apiLogDict["responseList"]=responseList if each_api_logs.requestDate: apiLogDict["requestDate"] = each_api_logs.requestDate.astimezone(ist_timezone).strftime("%d-%m-%Y %I:%M %p") else: apiLogDict["requestDate"] = "" if each_api_logs.responseDate: apiLogDict["responseDate"] = each_api_logs.responseDate.astimezone(ist_timezone).strftime("%d-%m-%Y %I:%M %p") else: apiLogDict["responseDate"] = "" apiLogDict["snoCount"]=snoCount apiLogsList.append(apiLogDict) # Pagination object for rendering pagination controls in the template pagination = Pagination(page=page, total=total_count, per_page=per_page, alignment="right", record_name="api_logs") else: print("form errors",form.errors) return render_template("super_admin_templates/api_logs_list.html", pagination=pagination, apiLogsList=apiLogsList, form = form, walletType = walletType ) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to fetched api logs data!!" return render_template("super_admin_templates/api_logs_list.html", error=error, pagination=pagination, apiLogsList=apiLogsList, form = form, walletType = walletType ) else: flash("The staff member does not have permission to view api logs.", "danger") return redirect(url_for("admin.dashboard")) @admin_logs.route("/client_logs_list",methods=["POST","GET"]) @adminid_access_token_required def client_logs_list(): if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") clientLogsList = [] pagination = None form = ClientLogsForm(request.args) permissionsList = check_permissions(session.get("adminId"),"clientPermissions") if "view" in permissionsList: try: walletType = request.args.get('walletType', '').strip() # Get the current page from the query parameters page = request.args.get("page", type=int, default=1) per_page = 20 # Number of items per page start = (page - 1) * per_page total_count = 0 filters = Q(status__in=[0,1]) if walletType in ["payin", "payout"]: filters &= Q(walletType=walletType) if form.validate(): total_count = ClientLogs.objects(filters).count() client_logs_queryset = ( ClientLogs.objects(filters) .only("id","walletType","apiCallingType","apiCallingUrl","transactionAPIId","requestData","responseData","clientIp","platformType","createdOn","requestDate","responseDate") .order_by("-id") .skip(start) .limit(per_page) ) snoCount = start for each_client_logs in client_logs_queryset: snoCount +=1 clientLogDict = { "id":str(each_client_logs.id), "walletType":each_client_logs.walletType, "apiCallingType":each_client_logs.apiCallingType, "apiCallingUrl":each_client_logs.apiCallingUrl, "requestData":each_client_logs.requestData, "responseData":each_client_logs.responseData, "clientIp":each_client_logs.clientIp, "platformType":each_client_logs.platformType, "createdOn":each_client_logs.createdOn.astimezone(ist_timezone) } requestList = [] for each_dict in each_client_logs.requestData: dict_items_data = each_dict.items() for each_key,each_value in dict_items_data: requestDict = { "keyName":each_key, "valueName":each_value } requestList.append(requestDict) clientLogDict["requestList"]=requestList responseList = [] for each_dict in each_client_logs.responseData: dict_items_data = each_dict.items() for each_key,each_value in dict_items_data: responseDict = { "keyName":each_key, "valueName":each_value } responseList.append(responseDict) clientLogDict["responseList"]=responseList try: if each_client_logs.transactionAPIId: clientLogDict["transactionAPIId"] = str(each_client_logs.transactionAPIId.id) else: clientLogDict["transactionAPIId"] = "" except Exception as e: clientLogDict["transactionAPIId"] = "" if each_client_logs.requestDate: clientLogDict["requestDate"] = each_client_logs.requestDate.astimezone(ist_timezone).strftime("%d-%m-%Y %I:%M %p") else: clientLogDict["requestDate"] = "" if each_client_logs.responseDate: clientLogDict["responseDate"] = each_client_logs.responseDate.astimezone(ist_timezone).strftime("%d-%m-%Y %I:%M %p") else: clientLogDict["responseDate"] = "" clientLogDict["snoCount"]=snoCount clientLogsList.append(clientLogDict) # Pagination object for rendering pagination controls in the template pagination = Pagination(page=page, total=total_count, per_page=per_page, alignment="right", record_name="client_logs") else: print("form errors", form.errors) return render_template("super_admin_templates/client_logs_list.html", pagination=pagination, clientLogsList=clientLogsList, walletType = walletType, form = form ) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to fetched client logs data!!" return render_template("super_admin_templates/client_logs_list.html", error=error, pagination=pagination, clientLogsList=clientLogsList, walletType = walletType, form = form ) else: flash("Staff member does not have given view client logs permissions!!") flash("The staff member does not have permission to view client logs.", "danger") return redirect(url_for("admin.dashboard")) # @admin_logs.route("/admin_logs_list",methods=["POST","GET"]) # def admin_logs_list(): # if not session.get("adminId"): # return redirect("admin_login") # adminId = session.get("adminId") # adminsLogsList = [] # pagination = None # permissionsList = check_permissions(session.get("adminId"),"adminLogsPermissions") # if "view" in permissionsList: # try: # if not session.get("adminId"): # return redirect("admin_login") # adminId = session.get("adminId") # admin_logs_queryset = AdminLogs.objects(status__in=[0,1]).order_by("-id").all() # # 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 = admin_logs_queryset.count() # start = (page - 1) * per_page # end = min(start + per_page, total_count) # total_admin_logs = admin_logs_queryset[start:end] # snoCount = start # for each_admin_log in total_admin_logs: # snoCount +=1 # adminLogDict = { # "id":str(each_admin_log.id), # "adminName":each_admin_log.adminId.userName, # "message":each_admin_log.message, # "createdOn":each_admin_log.createdOn.astimezone(ist_timezone).strftime("%d-%m-%Y %I:%M %p") # } # adminLogDict["snoCount"]=snoCount # adminsLogsList.append(adminLogDict) # # Pagination object for rendering pagination controls in the template # pagination = Pagination(page=page, total=total_count, per_page=per_page, alignment="right", record_name="admin_logs") # return render_template("super_admin_templates/admin_logs_list.html", # pagination=pagination, # adminsLogsList=adminsLogsList # ) # except Exception as e: # app.logger.error(traceback.format_exc()) # error = "Unable to fetched admin logs data!!" # return render_template("super_admin_templates/admin_logs_list.html", # error=error, # pagination=pagination, # adminsLogsList=adminsLogsList # ) # else: # flash("Staff member does not have given view admin logs permissions!!") # return redirect(url_for("admin.dashboard")) @admin_logs.route("/admin_logs_list", methods=["POST", "GET"]) @adminid_access_token_required def admin_logs_list(): if not session.get("adminId"): return redirect("admin_login") adminId = session.get("adminId") adminsLogsList = [] pagination = None form = AdminLogsForm(request.args) permissionsList = check_permissions(session.get("adminId"), "adminLogsPermissions") if "view" in permissionsList: try: name = request.args.get('name', '').strip() # Get the current page from the query parameters page = request.args.get("page", type=int, default=1) per_page = 20 # Number of items per page start = (page - 1) * per_page total_count = 0 filters = Q(status__in=[0,1]) if form.validate(): if name: matching_admins = SuperAdmin.objects(userName__icontains=name).only('id') admin_ids = [user.id for user in matching_admins] filters&=Q(adminId__in=admin_ids) total_count = AdminLogs.objects(filters).count() admin_logs_queryset = ( AdminLogs.objects(filters) .only("id","message","clientIp","latitude","longitude","attempts","createdOn","adminId","liveLocation") .order_by("-id") .skip(start) .limit(per_page) ) snoCount = start for each_admin_log in admin_logs_queryset: snoCount += 1 adminLogDict = { "id": str(each_admin_log.id), # "adminName": each_admin_log.adminId.userName if each_admin_log.adminId else "Unknown Admin", "message": each_admin_log.message, "ipAddress": each_admin_log.clientIp, "latitude": each_admin_log.latitude, "longitude": each_admin_log.longitude, "attempts": each_admin_log.attempts, # "createdOn": each_admin_log.createdOn.astimezone(ist_timezone).strftime("%d-%m-%Y %I:%M %p") "createdOn": each_admin_log.createdOn.astimezone(ist_timezone).strftime("%d-%m-%Y %I:%M %p") } try: if each_admin_log.adminId: adminLogDict["adminName"]=each_admin_log.adminId.userName else: adminLogDict["adminName"]="" except Exception as e: adminLogDict["adminName"]="" try: if each_admin_log.liveLocation: adminLogDict["liveLocation"]=each_admin_log.liveLocation else: adminLogDict["liveLocation"]="" except Exception as e: adminLogDict["liveLocation"]="" adminLogDict["snoCount"] = snoCount adminsLogsList.append(adminLogDict) # Pagination object for rendering pagination controls in the template pagination = Pagination(page=page, total=total_count, per_page=per_page, alignment="right", record_name="admin_logs") else: print("form errors", form.errors) return render_template("super_admin_templates/admin_logs_list.html", pagination=pagination, adminsLogsList=adminsLogsList, name = name, form = form ) except Exception as e: app.logger.error(traceback.format_exc()) error = "Unable to fetch admin logs data!!" return render_template("super_admin_templates/admin_logs_list.html", error=error, pagination=pagination, adminsLogsList=adminsLogsList, name = name, form = form ) else: flash("The staff member does not have permission to view admin logs.", "danger") return redirect(url_for("admin.dashboard")) # @admin_logs.route("/admin_logs_list", methods=["POST", "GET"]) # def admin_logs_list(): # if not session.get("adminId"): # return redirect("admin_login") # adminId = session.get("adminId") # adminsLogsList = [] # pagination = None # permissionsList = check_permissions(session.get("adminId"), "adminLogsPermissions") # if "view" in permissionsList: # try: # admin_logs_queryset = AdminLogs.objects(status__in=[0, 1]).order_by("-id").all() # page = request.args.get(get_page_parameter(), type=int, default=1) # per_page = 20 # total_count = admin_logs_queryset.count() # start = (page - 1) * per_page # end = min(start + per_page, total_count) # total_admin_logs = admin_logs_queryset[start:end] # for snoCount, each_admin_log in enumerate(total_admin_logs, start=start + 1): # adminLogDict = { # "id": str(each_admin_log.id), # "adminName": each_admin_log.adminId.userName if each_admin_log.adminId else "Unknown Admin", # "message": each_admin_log.message, # "ipAddress": each_admin_log.clientIp, # "latitude": each_admin_log.latitude if each_admin_log.latitude else "Not Available", # "longitude": each_admin_log.longitude if each_admin_log.longitude else "Not Available", # "attempts": each_admin_log.attempts if each_admin_log.attempts is not None else 0, # "createdOn": each_admin_log.createdOn.astimezone(ist_timezone).strftime("%d-%m-%Y %I:%M %p"), # "snoCount": snoCount # } # adminsLogsList.append(adminLogDict) # # Pagination object for rendering pagination controls in the template # pagination = Pagination(page=page, total=total_count, per_page=per_page, alignment="right", record_name="admin_logs") # return render_template("super_admin_templates/admin_logs_list.html", # pagination=pagination, # adminsLogsList=adminsLogsList) # except Exception as e: # app.logger.error(traceback.format_exc()) # error = "Unable to fetch admin logs data!!" # return render_template("super_admin_templates/admin_logs_list.html", # error=error, # pagination=pagination, # adminsLogsList=adminsLogsList) # else: # flash("Staff member does not have permission to view admin logs!!") # return redirect(url_for("admin.dashboard"))