# (c) cavaliba.com - home - views.py import re from django.contrib import messages from django.db.models import Q from django.http import HttpResponse from django.shortcuts import redirect, render from django.utils.translation import gettext as _ from app_data.aaa import start_view from app_data.data import update_bigset from app_data.eav import task_eav_purge, task_eav_refresh from app_data.models import DataRevision from .configuration import get_configuration, get_initial_form, get_post_form, save_form from .home import get_app_by_name, get_applist, get_cavaliba_apps from .log import DEBUG, INFO, WARNING, log, purge from .models import CavalibaLog # ----------------------------------------- # status page # ----------------------------------------- def status(request): return HttpResponse("OK") # ----------------------------------------- # welcome # ----------------------------------------- # def welcome(request): # context = start_view(request, app="home", view="welcome", noauth="app_sirene:index", perm="p_home_access") # aaa = context["aaa"] # if context["redirect"]: # return redirect(context["redirect"]) # ----------------------------------------- # private - private_page # ----------------------------------------- def private(request): context = start_view( request, app="home", view="private", noauth="app_sirene:index", perm="p_home_access", noauthz="app_sirene:index", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] apps = get_applist(aaa) for x in apps: if x.keyname == "home": apps.remove(x) break # page/order for UI # [ [page1, [item1, item22, ...] , [ page2, [...] ] , ... ] paginated = [] pagelist = [] index = {} # page => [class1, class2] default_name = get_configuration("home", "GLOBAL_APPNAME") for element in apps: # order = element.order if not element.is_allowed: continue section = element.dashboard_section if not section: section = default_name if len(section) == 0: section = default_name if section not in index: index[section] = [] pagelist.append(section) index[section].append(element) default_name = section for p in pagelist: paginated.append([p, index[p]]) log(DEBUG, aaa=aaa, app="home", view="dash", action="view", status="OK") context["paginated"] = paginated return render(request, "app_home/private.html", context) # ----------------------------------------- # configuration # ----------------------------------------- def configuration(request, appname=None): # context = start_view(request, app="home", view="private", noauth="app_sirene:index", perm="p", noauthz="app_sirene:index") context = start_view( request, app="home", view="configuration", noauth="app_home:index", perm="p_conf_admin", noauthz="app_sirene:index", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if appname: appobj = get_app_by_name(appname) if not appobj: return redirect("app_home:index") if request.method == "POST": form = get_post_form(request, appname=appname) if form.is_valid(): save_form(form, appname=appname) messages.add_message(request, messages.SUCCESS, _("Configuration updated")) log( INFO, aaa=aaa, app="home", view="configuration", action="update", status="OK", data=f"app {appname}", ) return redirect("app_home:configuration", appname) else: messages.add_message(request, messages.ERROR, _("Invalid configuration")) else: form = get_initial_form(appname) if appname: context["appname"] = appobj.keyname context["apps"] = get_cavaliba_apps() context["form"] = form return render(request, "app_home/configuration.html", context) # ----------------------------------------- # log # ----------------------------------------- def logview(request, level="info"): context = start_view( request, app="home", view="log", noauth="app_sirene:index", perm="p_log_view", noauthz="app_home:index", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] # purge ? if request.method == "POST": if request.POST.get("purge"): if "p_log_admin" in aaa["perms"]: count = purge(aaa=aaa) messages.add_message(request, messages.SUCCESS, _("Logs purged: ") + str(count)) log( WARNING, aaa=aaa, app="home", view="log", action="purge", status="OK", data=f"{count} removed", ) return redirect("app_home:log") # bigset or not count = CavalibaLog.objects.count() bigset = True context["bigset"] = bigset context["count"] = count # Filter POST (form) # ------------------- # query = request.GET.get("q", "") query = "" if request.method == "POST": if request.POST.get("query"): query = request.POST.get("query") if request.method == "GET": if request.GET.get("query"): query = request.GET.get("query") if query: m = re.compile(r"[a-zA-Z0-9()_/.-]*$") if not m.match(query): query = "" context["query"] = query # partial query + paginate if bigset try: size = int(request.GET.get("size", 100)) page = int(request.GET.get("page", 1)) except Exception: return redirect("app_home:log") if page < 1 or size > 10000 or size < 1: return redirect("app_home:log") offset = (page - 1) * size limit = offset + size if len(query) > 0: if level not in ["debug", "info", "warning", "error", "critical"]: level = "debug" if level == "info": level_display = _("INFO") logs = ( CavalibaLog.objects.all() .order_by("-id") .filter(level__in=["INFO", "WARNING", "ERROR", "CRITICAL"]) .filter( Q(app__icontains=query) | Q(username__icontains=query) | Q(view__icontains=query) | Q(action__icontains=query) | Q(data__icontains=query) | Q(status__icontains=query) | Q(user_ip__icontains=query) )[offset:limit] ) elif level == "warning": level_display = _("WARNING") logs = ( CavalibaLog.objects.all() .order_by("-id") .filter(level__in=["WARNING", "ERROR", "CRITICAL"]) .filter( Q(app__icontains=query) | Q(username__icontains=query) | Q(view__icontains=query) | Q(action__icontains=query) | Q(data__icontains=query) | Q(status__icontains=query) | Q(user_ip__icontains=query) )[offset:limit] ) elif level == "error": level_display = _("ERROR") logs = ( CavalibaLog.objects.all() .order_by("-id") .filter(level__in=["ERROR", "CRITICAL"]) .filter( Q(app__icontains=query) | Q(username__icontains=query) | Q(view__icontains=query) | Q(action__icontains=query) | Q(data__icontains=query) | Q(status__icontains=query) | Q(user_ip__icontains=query) )[offset:limit] ) elif level == "critical": level_display = _("CRITICAL") logs = ( CavalibaLog.objects.all() .order_by("-id") .filter(level="CRITICAL") .filter( Q(app__icontains=query) | Q(username__icontains=query) | Q(view__icontains=query) | Q(action__icontains=query) | Q(data__icontains=query) | Q(status__icontains=query) | Q(user_ip__icontains=query) )[offset:limit] ) else: level_display = _("DEBUG") logs = ( CavalibaLog.objects.all() .order_by("-id") .filter( Q(app__icontains=query) | Q(username__icontains=query) | Q(view__icontains=query) | Q(action__icontains=query) | Q(data__icontains=query) | Q(status__icontains=query) | Q(user_ip__icontains=query) )[offset:limit] ) else: if level not in ["debug", "info", "warning", "error", "critical"]: level = "info" if level == "info": level_display = _("INFO") logs = ( CavalibaLog.objects.all() .order_by("-id") .filter(level__in=["INFO", "WARNING", "ERROR", "CRITICAL"])[offset:limit] ) elif level == "warning": level_display = _("WARNING") logs = ( CavalibaLog.objects.all() .order_by("-id") .filter(level__in=["WARNING", "ERROR", "CRITICAL"])[offset:limit] ) elif level == "error": level_display = _("ERROR") logs = ( CavalibaLog.objects.all() .order_by("-id") .filter(level__in=["ERROR", "CRITICAL"])[offset:limit] ) elif level == "critical": level_display = _("CRITICAL") logs = CavalibaLog.objects.all().order_by("-id").filter(level="CRITICAL")[offset:limit] else: level_display = _("DEBUG") logs = CavalibaLog.objects.all().order_by("-id")[offset:limit] # PREV | FIRST || CURRENT or ... || LAST | NEXT context["size"] = size context["page"] = page page_last = int(count / size) + 1 if page > 1: context["page_prev"] = page - 1 else: context["page_prev"] = page if page == 1: context["page_first"] = True else: context["page_first"] = False if page > 1 and page < page_last: context["page_current"] = True else: context["page_current"] = False context["page_last"] = page_last if page < page_last: context["page_last_active"] = False else: context["page_last_active"] = True if page < page_last: context["page_next"] = page + 1 else: context["page_next"] = page context["level"] = level_display context["logs"] = logs return render(request, "app_home/log.html", context) # ---------------------------------------------------------------------------------------------------- # admin tools MAIN # ----------------------------------------------------------------------------------------------------- def admintools(request): context = start_view( request, app="admintools", view="main", noauth="app_home:index", perm="p_admintools", noauthz="app_sirene:index", ) if context["redirect"]: return redirect(context["redirect"]) return render(request, "app_home/admintools.html", context) # ----------------------------------------- # admin tools - update bigset # ----------------------------------------- def admintools_updatebigset(request): context = start_view( request, app="admintools", view="update_bigset", noauth="app_home:index", perm="p_admintools", noauthz="app_sirene:index", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if request.method == "POST": messages.add_message(request, messages.SUCCESS, _("Task started...")) update_bigset(aaa=aaa) messages.add_message(request, messages.SUCCESS, _("Task done.")) log( WARNING, aaa=aaa, app="home", view="admintools", action="updatebigset", status="OK", data="done", ) return redirect("app_home:admintools") # ----------------------------------------- # admin tools - EAV refresh # ----------------------------------------- def admintools_eavrefresh(request): context = start_view( request, app="admintools", view="eav_refresh", noauth="app_home:index", perm="p_admintools", noauthz="app_sirene:index", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if request.method == "POST": messages.add_message(request, messages.SUCCESS, _("EAV Refresh - long task started...")) log( WARNING, aaa=aaa, app="home", view="admintools", action="eav_refresh", status="OK", data="started...", ) task_eav_refresh.delay(dryrun=False, force=False) return redirect("app_home:admintools") # ----------------------------------------- # admin tools - EAV purge # ----------------------------------------- def admintools_eavpurge(request): context = start_view( request, app="admintools", view="eav_purge", noauth="app_home:index", perm="p_admintools", noauthz="app_sirene:index", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if request.method == "POST": messages.add_message(request, messages.SUCCESS, _("EAV Purge - long task started...")) log( WARNING, aaa=aaa, app="home", view="admintools", action="eav_purge", status="OK", data="started...", ) task_eav_purge.delay(dryrun=False) return redirect("app_home:admintools") # ----------------------------------------- # admin tools - Revision # ----------------------------------------- def admintools_revision(request): context = start_view( request, app="admintools", view="revision", noauth="app_home:index", perm="p_admintools", noauthz="app_sirene:index", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] # Get pagination parameters from GET request try: page = int(request.GET.get("page", 1)) size = int(request.GET.get("size", 1000)) except Exception: page = 1 size = 1000 # Calculate offset for pagination offset = (page - 1) * size # Retrieve revisions ordered by date descending with pagination revisions = DataRevision.objects.all().order_by("-date")[offset : offset + size] context["revisions"] = revisions context["count"] = DataRevision.objects.count() context["page"] = page context["size"] = size log( DEBUG, aaa=aaa, app="home", view="admintools", action="revision", status="OK", data="view accessed", ) return render(request, "app_home/revision.html", context)