# (c) cavaliba.com - home - views.py import re 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 app_user.aaa import start_view 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 .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"]) aaa = context["aaa"] 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)