# (c) cavaliba.conf - app_sirene - views.py from app_data.data import Instance from app_data.fieldtypes.field_enumerate import get_enumerate from app_data.models import DataInstance from app_home.configuration import get_configuration from app_home.log import DEBUG, ERROR, INFO, WARNING, log from app_sirene.forms import MessageUpdateForm # create_notification in data/views HOOK from app_sirene.notify import ( aaa_message_allowed, archive_all, archive_expired, archive_message, count_userlist, create_update, get_message_by_id, get_message_enabled, get_public_active, get_public_default, get_updates, message_from_request, message_notify, reopen_message, ) from app_user.aaa import get_aaa, start_ajax, start_view from app_user.models import SireneGroup, SireneUser from django.contrib import messages from django.http import HttpResponse from django.shortcuts import redirect, render from django.utils.translation import gettext as _ from django.views.decorators.csrf import csrf_protect #----------------------------------------- # public #----------------------------------------- def index(request): context = start_view(request, app="sirene", view="public") if context["redirect"]: return redirect(context["redirect"]) max_display = int(get_configuration("sirene", "PUBLIC_MAX_ITEMS")) #sort_order = get_configuration("sirene", "PUBLIC_SORT_ORDER") # skip public page, if # - trusted ip / or authenticated # - configuration requires skip # - no toggle in GET URL skip_public = get_configuration("sirene", "PUBLIC_SKIP_TO_TRUSTED") keep_public = request.GET.get('public', 'no') if keep_public != "yes": if skip_public == "yes": aaa = get_aaa(request) if (aaa['is_trusted_ip'] or aaa['is_authenticated']): return redirect("app_sirene:anonymous") instances = get_public_active() if len(instances) == 0: instances = get_public_default() messages = [] for instance in instances: if not instance.is_enabled: continue # no private (restricted) message for Anonymous if instance.is_field_true('is_restricted'): continue # full structure with datapoints ui_dict = instance.get_dict_for_ui_detail(skip_external=True, skip_injected=False, skip_enumerate=False) # extract relevant value message = {} try: message['title'] = ui_dict['displayname'] message['severity'] = ui_dict['severity']['value'] # [{ value:'xxx', widget:'' }] message['bgcolor'] = ui_dict['bgcolor']['value'] message['fgcolor'] = ui_dict['fgcolor']['value'] message['body'] = ui_dict['content']['value'] messages.append(message) except Exception: continue context["pages"] = messages[:max_display] return render(request, 'app_sirene/public.html', context) #----------------------------------------- # anonymous #----------------------------------------- def anon_list(request): context = start_view(request, app="sirene", view="anonymous_list") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if not (aaa["is_trusted_ip"] or aaa["is_authenticated"] or aaa["is_visitor"]): return redirect("app_sirene:index") instances = get_message_enabled() messages = [] for instance in instances: # no private (restricted) message for Anonymous if instance.is_field_true('is_restricted'): continue # full structure with datapoints ui_dict = instance.get_dict_for_ui_detail(skip_external=True, skip_injected=False, skip_enumerate=False) # extract relevant value from (flat) Datapoints message = {} try: message['id'] = ui_dict['id'] except Exception: pass try: message['displayname'] = ui_dict['displayname'] message['category'] = ui_dict['category']['value'] # [{ value:'xxx', widget:'' }] message['severity'] = ui_dict['severity']['value'] # [{ value:'xxx', widget:'' }] except Exception: continue try: message['created_at'] = ui_dict['created_at']['value'] except Exception: message['created_at'] = '' try: message['updated_at'] = ui_dict['updated_at']['value'] except Exception: message['updated_at'] = '' messages.append(message) log(DEBUG, aaa=aaa, app="sirene", view="anonymous_list", action="get", status="OK", data="") context["messages"] = messages return render(request, 'app_sirene/anonymous.html', context) def anon_detail(request, id=None): ''' display private message details for anonymous users''' context = start_view(request, app="sirene", view="anonymous_detail") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] aaa = get_aaa(request) if not (aaa['is_trusted_ip'] or aaa['is_authenticated'] or aaa["is_visitor"]): log(DEBUG, aaa=aaa, app="sirene", view="anonymous_detail", action="get", status="KO", data="access denied") return redirect("app_sirene:index") if not id: log(DEBUG, aaa=aaa, app="sirene", view="anonymous_detail", action="get", status="KO", data="no message") return redirect("app_sirene:anonymous") instance = get_message_by_id(id=id) if not instance: log(DEBUG, aaa=aaa, app="sirene", view="anonymous_detail", action="get", status="KO", data="no message") return redirect("app_sirene:anonymous") # exclude private (restricted) message from anonymous if instance.is_field_true('is_restricted'): return redirect("app_sirene:anonymous") # acive message only for anonymous if not instance.is_enabled: return redirect("app_sirene:anonymous") ui_dict = instance.get_dict_for_ui_detail(skip_external=True, skip_injected=False, skip_enumerate=False) # extract relevant value from (flat) Datapoints message = {} message['id'] = ui_dict['id'] try: message['title'] = ui_dict['displayname'] message['body'] = ui_dict['content']['value'] message['category'] = ui_dict['category']['value'] # [{ value:'xxx', widget:'' }] message['severity'] = ui_dict['severity']['value'] # [{ value:'xxx', widget:'' }] message['bgcolor'] = ui_dict['severity__bgcolor']['value'] except Exception: return redirect("app_sirene:anonymous") try: message['created_at'] = ui_dict['created_at']['value'] except Exception: message['created_at'] = '' try: message['created_by'] = ui_dict['created_by']['value'][0] except Exception: message['created_by'] = '' try: message['updated_at'] = ui_dict['updated_at']['value'] except Exception: message['updated_at'] = '' log(DEBUG, aaa=aaa, app="sirene", view="anonymous_detail", action="get", status="OK", data=f"{message['title']}") context["message"] = message return render(request, 'app_sirene/anonymous_detail.html', context) # ------------------------------------------- # private message list : name = 'private' # ------------------------------------------- def private_list(request): context = start_view(request, app="sirene", view="message_list", noauth="app_sirene:index", perm="p_sirene_access", noauthz="app_sirene:index", # explicitly allow visitor visitor = True ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] instances = get_message_enabled() pages = [] for instance in instances: # SireneUser ORM object r = aaa_message_allowed(instance=instance, aaa=aaa) if not r: continue # full structure with datapoints ui_dict = instance.get_dict_for_ui_detail(skip_external=True, skip_injected=False, skip_enumerate=False) # extract relevant value from (flat) Datapoints page = {} page['id'] = ui_dict['id'] page['displayname'] = ui_dict['displayname'] try: page['category'] = ui_dict['category']['value'][0]['value'] # [{ value:'xxx', widget:'' }] except Exception: continue try: page['severity'] = ui_dict['severity']['value'][0]['value'] # [{ value:'xxx', widget:'' }] except Exception as e: print(e) continue try: page['bgcolor'] = ui_dict['severity__bgcolor']['value'] except Exception: continue try: page['created_at'] = ui_dict['created_at']['value'] except Exception: continue try: page['updated_at'] = ui_dict['updated_at']['value'] except Exception: page['updated_at'] = '' if instance.is_field_true('is_restricted'): page['is_restricted'] = True else: page['is_restricted'] = False pages.append(page) log(DEBUG, aaa=aaa, app="sirene", view="message", action="list", status="OK", data="") context["pages"] = pages return render(request, 'app_sirene/private.html', context) #----------------------------------------- # template list AJAX #----------------------------------------- def template_list_ajax(request): """AJAX endpoint for loading templates in modal via HTMX""" context = start_view(request, app="sirene", view="template_list_ajax", noauth="app_sirene:index", perm="p_sirene_new", noauthz="app_sirene:private") if context["redirect"]: return redirect(context["redirect"]) templates = [] for instance in Instance.iterate_classname(classname="sirene_template", enabled="yes"): try: hint = instance.fields["hint"].value[0] except Exception: hint = "" try: category = instance.fields['category'].value[0] except Exception: category = "?" try: severity = instance.fields['severity'].value[0] except Exception: severity = "?" item = { "id": instance.id, "title": instance.displayname, "category": category, "severity": severity, "hint": hint } templates.append(item) context["templates"] = templates return render(request, 'app_sirene/_template_list.html', context) #----------------------------------------- # notify summary AJAX (stub) #----------------------------------------- @csrf_protect def notify_summary_ajax(request): """AJAX endpoint for updating notification target counters""" response = HttpResponse("{}", content_type='text/json') context = start_ajax(request) if not context: return response if request.method == "POST": # Extract POST data from HTMX request notify_user = request.POST.getlist('notify_user') notify_group = request.POST.getlist('notify_group') has_email = request.POST.get('has_email') == 'on' or request.POST.get('has_email') == 'true' has_sms = request.POST.get('has_sms') == 'on' or request.POST.get('has_sms') == 'true' # Create a temp/fake message temp_instance = Instance(classname="sirene_message") temp_instance.fields['has_email'].value = [has_email] temp_instance.fields['has_sms'].value = [has_sms] temp_instance.fields['notify_user'].value = notify_user temp_instance.fields['notify_group'].value = notify_group # generic notify_to conf_notify = get_configuration('sirene','NOTIFY_FIELDS') notify_fields = conf_notify.split() context["notify_generic"] = {} for fieldname in notify_fields: if fieldname in ["user", 'group']: continue # TODO check fieldname is an existing schema + form field try: temp_instance.fields['notify_'+fieldname].value = request.POST.getlist('notify_'+fieldname) except Exception: pass user_max = int(get_configuration("sirene","USER_MAX_NOTIFICATION")) people_count, email_count, sms_count = count_userlist(instance=temp_instance, maxcount=user_max) context = { 'people_count': people_count, 'email_count': email_count, 'sms_count': sms_count, } return render(request, 'app_sirene/_notify_summary.html', context) return render(request, 'app_sirene/_notify_summary.html', {'people_count': 0, 'email_count': 0, 'sms_count': 0}) # ------------------------------------------- # detail # ------------------------------------------- def private_detail(request, id=None): ''' GET id - display details + admin tools''' context = start_view(request, app="sirene", view="detail", noauth="app_sirene:index", perm="p_sirene_detail", noauthz="app_sirene:private", visitor=True) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] archive_expired(aaa=aaa) if not id: messages.add_message(request, messages.ERROR, _("Not found")) log(ERROR, aaa=aaa, app="sirene", view="message", action="detail", status="KO", data="no ID") return redirect("app_sirene:private") instance = get_message_by_id(id=id) if not instance: messages.add_message(request, messages.ERROR, _("Not found")) log(ERROR, aaa=aaa, app="sirene", view="message", action="detail", status="KO", data=f"Not found {id}") return redirect("app_sirene:private") # SireneUser ORM object r = aaa_message_allowed(instance=instance, aaa=aaa) if not r: return redirect("app_sirene:private") ui_dict = instance.get_dict_for_ui_detail(skip_external=True, skip_injected=False, skip_enumerate=False) #instance.print() # extract relevant value from (flat) Datapoints page = {} page['id'] = instance.id page['is_enabled'] = instance.is_enabled page['keyname'] = instance.keyname try: page['title'] = ui_dict['displayname'] except Exception: page['title'] = "" try: page['body'] = ui_dict['content']['value'] except Exception: page['body'] = "" try: if instance.is_field_true('has_sms'): page['sms_body'] = ui_dict['sms_content']['value'] else: page['sms_body'] = None except Exception: page['sms_body'] = None try: page['category'] = ui_dict['category']['value'][0]['value'] except Exception: page['category'] = "" try: page['severity'] = ui_dict['severity']['value'][0]['value'] except Exception: page['severity'] = "" try: page['bgcolor'] = ui_dict['severity__bgcolor']['value'] except Exception: page['bgcolor'] = "success" try: page['created_at'] = ui_dict['created_at']['value'] except Exception: page['created_at'] = '' try: page['created_by'] = ui_dict['created_by']['value'][0]['key'] except Exception: page['created_by'] = '' try: page['updated_at'] = ui_dict['updated_at']['value'] except Exception: page['updated_at'] = '' try: page['updated_by'] = ui_dict['updated_by']['value'][0]['key'] except Exception: page['updated_by'] = '' try: page['removed_at'] = ui_dict['removed_at']['value'] except Exception: page['removed_at'] = '' try: page['removed_by'] = ui_dict['removed_by']['value'][0]['key'] except Exception: page['removed_by'] = '' if instance.is_field_true('is_restricted'): page['is_restricted'] = True else: page['is_restricted'] = False try: page['publicpage'] = ui_dict['public_page']['value'][0]['display'] except Exception: page['publicpage'] = "" try: page['email_count'] = int(ui_dict['email_count']['value']) except Exception: page['email_count'] = 0 try: page['sms_count'] = int(ui_dict['sms_count']['value']) except Exception: page['email_count'] = 0 page['sms_count'] = 0 try: page['notified_username'] = ', '.join(instance.fields['notified_username'].value) page['notified_username_count'] = len(instance.fields['notified_username'].value) except Exception: pass # notify_* fields: # [{'key': 'testuser01', 'display': 'Test User 01'}] # ['testgroup02'] # [{'key': 'testapp01', 'display': 'Test APP 01', 'id': id}] page['notified_targets'] = [] for fieldname in ui_dict: if fieldname.startswith('notify_'): for target in ui_dict[fieldname]['value']: if type(target) is dict: v = target.get('display', target.get('key', '?')) elif type(target) is list: v = target[0] elif type(target) is str: v = target else: v= 'n/a' page['notified_targets'].append(v) try: page['template'] = ui_dict['template']['value'] except Exception: page['template'] = "N/A" log(DEBUG, aaa=aaa, app="sirene", view="message", action="detail", status="OK", data=f"{page['title']}") # updates updates = get_updates(instance=instance)[::-1] if len(updates) > 0: context['updates'] = updates # total email / sms page["total_email_count"] = int(page["email_count"]) page["total_sms_count"] = int(page["sms_count"]) if len(updates) > 0: for update in updates: page["total_email_count"] += int(update["email_count"]) page["total_sms_count"] += int(update["sms_count"]) context["page"] = page return render(request, 'app_sirene/detail.html', context) #----------------------------------------- # new_message #----------------------------------------- def new_message(request, id=None): ''' Create a new sirene_message GET => form / POST => submit IN: id : sirene_template id to create new message from ; if None, blank form OUT: GET: HTML Form POST: sirene_message created, notify called ''' context = start_view(request, app="sirene", view="new_message", noauth="app_sirene:index", perm="p_sirene_new", noauthz="app_sirene:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] # POST => Submit => create DB mesasge => notify if request.method == "POST": # Instance() message = message_from_request(request=request, aaa=aaa) if message: if message.is_valid(): if message_notify(message=message, aaa=aaa): email_count = message.fields["email_count"].value[0] sms_count = message.fields["sms_count"].value[0] messages.add_message(request, messages.SUCCESS, f"OK - {email_count} emails / {sms_count} SMS") log(INFO, aaa=aaa, app="sirene", view="message", action="notify", status="OK", data=f"#{message.id} new - {message.displayname} - queued {email_count} emails, {sms_count} SMS") return redirect("app_sirene:private") messages.add_message(request, messages.ERROR, _("Failed")) log(WARNING, aaa=aaa, app="sirene", view="message", action="notify", status="KO", data="failed to post") # exit POST REQUEST in all case return redirect("app_sirene:private") # GET => provide edit form , blank or from template id context["message_title"] = "" context["message_template"] = None context["message_severity"] = get_enumerate("sirene_severity") context["message_category"] = get_enumerate("sirene_category") context["message_restricted"] = False context["message_public"] = [] context["message_user"] = [] context["message_group"] = [] context["message_email"] = False context["message_email_content"] = None context["message_sms"] = False context["message_sms_content"] = None # generic notify_to conf_notify = get_configuration('sirene','NOTIFY_FIELDS') notify_fields = conf_notify.split() context["notify_generic"] = {} for fieldname in notify_fields: if fieldname in ["user", 'group']: continue # TODO check fieldname is an existing schema + form field context["notify_generic"][fieldname] = [] if id: # get template template = Instance.from_id(id) if not template: messages.add_message(request, messages.ERROR, _("Not found")) log(ERROR, aaa=aaa, app="sirene", view="message", action="new", status="KO", data=f"Invalid template {id}") return redirect("app_sirene:private") if not (template.classname == "sirene_template" and template.is_enabled): log(ERROR, aaa=aaa, app="sirene", view="message", action="new", status="KO", data=f"Invalid template {id}") messages.add_message(request, messages.ERROR, _("Not found")) return redirect("app_sirene:private") # copy template # -------------- context["message_template"] = template.keyname context["message_title"] = template.displayname # severity severities = [] for i in context["message_severity"]: if "is_enabled" in i: if not i["is_enabled"]: continue if i["value"] in template.fields['severity'].value: i["selected"] = True else: i["selected"] = False severities.append(i) context["message_severity"] = severities # category categories = [] for i in context["message_category"]: if "is_enabled" in i: if not i["is_enabled"]: continue if i["value"] in template.fields['category'].value: i["selected"] = True else: i["selected"] = False categories.append(i) context["message_category"] = categories # restricted if template.is_field_true('is_restricted'): context["message_restricted"] = True else: context["message_restricted"] = False # public page context["message_public"] = [] instances = DataInstance.objects.filter(classname="sirene_public", is_enabled=True) for i in instances: if i.keyname in template.fields['public_page'].value: selected = True else: selected = False item = { "key":i.keyname, "display":i.displayname, "selected":selected } context["message_public"].append( item ) # user: NEXT : make optional #if "user" in notify_fields: context["message_user"] = [] users = SireneUser.objects.filter(login__in=template.fields['notify_user'].value, is_enabled=True) for i in users: selected = True item = { "key":i.login, "display":i.displayname, "selected":selected} context["message_user"].append(item) # group # if "group" in notify_fields: context["message_group"] = [] allgroups = SireneGroup.objects.filter(is_role=False).all() for i in allgroups: if i.keyname in template.fields['notify_group'].value: selected = True else: selected = False item = { "key":i.keyname, "selected":selected} context["message_group"].append( item ) # generic notify_ fields #context["notify_generic"] = {} for fieldname in notify_fields: if fieldname in ["user", 'group']: continue # TODO check fieldname is an existing schema + form field if 'notify_'+fieldname not in template.fields: continue #context["notify_generic"][fieldname] = [] instances = DataInstance.objects.filter( classname=fieldname, keyname__in=template.fields['notify_'+fieldname].value,is_enabled=True) for i in instances: selected = True item = { "key":i.keyname, "display":i.displayname, "selected":selected} context["notify_generic"][fieldname].append(item) # sms + sms_content if template.is_field_true('has_sms'): context["message_sms"] = True try: context["message_sms_content"] = template.fields['sms_content'].value[0] except Exception: pass else: context["message_sms"] = False # email + email_content if template.is_field_true('has_email'): context["message_email"] = True try: context["message_email_content"] = template.fields['content'].value[0] except Exception: pass else: context["message_email"] = False else: # blank form with external lists # public_page context["message_public"] = [] instances = DataInstance.objects.filter(classname="sirene_public", is_enabled=True) for i in instances: item = { "key":i.keyname, "display":i.displayname, "selected":False } context["message_public"].append( item ) # group (no ajax) context["message_group"] = [] allgroups = SireneGroup.objects.filter(is_role=False).all() for i in allgroups: item = { "key":i.keyname, "selected":False} context["message_group"].append( item ) return render(request, 'app_sirene/message.html', context) #----------------------------------------- # update #----------------------------------------- def update(request, id=None): ''' Edit MessageUpdateForm - GET/POST''' context = start_view(request, app="sirene", view="update", noauth="app_sirene:index", perm="p_sirene_update", noauthz="app_sirene:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if not id: messages.add_message(request, messages.ERROR, _("Not found")) log(ERROR, aaa=aaa, app="sirene", view="update", action="message_id", status="KO", data="No ID") return redirect("app_sirene:private") instance = get_message_by_id(id) if not instance: messages.add_message(request, messages.ERROR, _("Not found")) log(ERROR, aaa=aaa, app="sirene", view="update", action="message", status="KO", data=f"{id} not found") return redirect("app_sirene:private") # SireneUser ORM object r = aaa_message_allowed(instance=instance, aaa=aaa) if not r: return redirect("app_sirene:private") if request.method == "POST": form = MessageUpdateForm(request.POST) if form.is_valid(): cd = form.cleaned_data mu = {} mu['content'] = cd["content"] mu['sms_content'] = cd["sms_content"] mu['has_email'] = cd['has_email'] mu['has_sms'] = cd['has_sms'] action = 'update' err, email_count, sms_count = create_update(instance=instance, aaa=aaa, mu=mu) if err: messages.add_message(request, messages.ERROR, _("Failed")) log(WARNING, aaa=aaa, app="sirene", view="update", action=action, status="KO", data=f"{instance.id} - {instance.displayname} - {err}") else: messages.add_message(request, messages.SUCCESS, f"OK - {email_count} emails / {sms_count} SMS") log(INFO, aaa=aaa, app="sirene", view="update", action=action, status="OK", data=f"#{instance.id} update - queued {email_count} emails - {sms_count} sms") return redirect("app_sirene:detail", instance.id) else: messages.add_message(request, messages.ERROR, _("Invalid form")) log(ERROR, aaa=aaa, app="sirene", view="update", action="valid", status="KO", data=f"invalid form - {instance.id} - {instance.displayname}") else: form = MessageUpdateForm() # extract relevant value from (flat) Datapoints message = {} message['id'] = instance.id context["message"] = message context["form"] = form return render(request, 'app_sirene/update.html', context) #----------------------------------------- # close (to archive) #----------------------------------------- def close(request, id=None): ''' Close an active message ''' context = start_view(request, app="sirene", view="close", noauth="app_sirene:index", perm="p_sirene_archive", noauthz="app_sirene:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] # messages.add_message(request, messages.ERROR, _("Not found")) if not id: log(ERROR, aaa=aaa, app="sirene", view="close", action="id", status="KO", data="No ID") return redirect("app_sirene:private") instance = get_message_by_id(id) if not instance: log(ERROR, aaa=aaa, app="sirene", view="close", action="get", status="KO", data=f"{id} not found") return redirect("app_sirene:private") if not instance.is_enabled: log(ERROR, aaa=aaa, app="sirene", view="close", action="enabled", status="KO", data=f"{id} not enabled") return redirect("app_sirene:private") if request.method != 'POST': log(ERROR, aaa=aaa, app="sirene", view="close", action="method", status="KO", data=f"{id} invalid method") return redirect("app_sirene:private") result = archive_message(id=id, aaa=aaa) log(INFO, aaa=aaa, app="sirene", view="close", action="archive", status="OK", data=f"#{id} closed") messages.add_message(request, messages.SUCCESS, _("OK")) return redirect("app_sirene:private") #----------------------------------------- # close_all #----------------------------------------- def close_all(request): ''' POST with CSRF > close all is_visble messages''' context = start_view(request, app="sirene", view="close_all", noauth="app_sirene:index", perm="p_sirene_flushall", noauthz="app_sirene:index") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if request.method != 'POST': log(ERROR, aaa=aaa, app="sirene", view="close_all", action="method", status="KO", data="method not supported") return redirect("app_sirene:index") count = archive_all(aaa=aaa) log(INFO, aaa=aaa, app="sirene", view="close_all", action="post", status="OK", data=f"close_all {count} entries") messages.add_message(request, messages.SUCCESS, _("OK")) return redirect("app_sirene:private") #----------------------------------------- # reopen #----------------------------------------- def reopen(request, id=None): ''' Reopen archived message''' context = start_view(request, app="sirene", view="reopen", noauth="app_sirene:index", perm="p_sirene_update", noauthz="app_sirene:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if not id: log(ERROR, aaa=aaa, app="sirene", view="reopen", action="id", status="KO", data="No ID") return redirect("app_sirene:private") instance = get_message_by_id(id) if not instance: log(ERROR, aaa=aaa, app="sirene", view="reopen", action="instance", status="KO", data=f"{id} not found") return redirect("app_sirene:private") # SireneUser ORM object r = aaa_message_allowed(instance=instance, aaa=aaa) if not r: return redirect("app_sirene:private") result = reopen_message(instance=instance) if not result: messages.add_message(request, messages.ERROR, _("Failed")) log(WARNING, aaa=aaa, app="sirene", view="reopen", action="reopen", status="KO", data=f"{instance.id} - {instance.displayname}") else: messages.add_message(request, messages.SUCCESS, _("Reopened")) log(INFO, aaa=aaa, app="sirene", view="reopen", action="reopen", status="OK", data=f"{instance.id} - {instance.displayname}") return redirect("app_sirene:detail", instance.id) #----------------------------------------- # display archive #----------------------------------------- def archive(request): context = start_view(request, app="sirene", view="archive", noauth="app_sirene:index", perm="p_sirene_history", noauthz="app_sirene:private", visitor=True) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] bigset_size = int(get_configuration("data", "DATA_BIGSET_SIZE")) context["bigset"] = True # Filter POST (form) query = "" if request.method == "POST": if request.POST.get('query'): query = request.POST.get('query').strip() context["query"] = query try: size = int(request.GET.get("size", bigset_size)) page = int(request.GET.get("page", 1)) except Exception: size = bigset_size page = 1 if page < 1 or size > 10000 or size < 1: size = bigset_size page = 1 # build full list all_pages = [] for instance in Instance.iterate_classname(classname="sirene_message", first=0, last=5000, enabled="no", expand=True): if not aaa_message_allowed(instance=instance, aaa=aaa): continue ui_dict = instance.get_dict_for_ui_detail(skip_external=True, skip_injected=False, skip_enumerate=False) page_item = {} page_item['id'] = ui_dict['id'] try: page_item['displayname'] = ui_dict['displayname'] except Exception: page_item['displayname'] = "" try: page_item['category'] = ui_dict['category']['value'][0]['value'] except Exception: page_item['category'] = "" try: page_item['severity'] = ui_dict['severity']['value'][0]['value'] except Exception: page_item['severity'] = "" try: page_item['bgcolor'] = ui_dict['severity__bgcolor']['value'] except Exception: page_item['bgcolor'] = "" page_item['is_enabled'] = False if instance.is_field_true('is_restricted'): page_item['is_restricted'] = True else: page_item['is_restricted'] = False try: page_item['publicpage'] = ui_dict['public_page']['value'][0]['display'] except Exception: page_item['publicpage'] = "" try: page_item['created_at'] = ui_dict['created_at']['value'] except Exception: page_item['created_at'] = '' all_pages.append(page_item) # search filter if query: q = query.lower() all_pages = [ p for p in all_pages if q in p['displayname'].lower() or q in p['category'].lower() or q in p['severity'].lower() ] # sort by date descending all_pages.sort(key=lambda x: x['created_at'] or '', reverse=True) # paginate count = len(all_pages) offset = (page - 1) * size pages = all_pages[offset: offset + size] # paginator context (same pattern as user_list / group_list) page_last = int(count / size) + 1 context["count"] = count context["size"] = size context["page"] = page context["page_prev"] = page - 1 if page > 1 else page context["page_first"] = page == 1 context["page_current"] = 1 < page < page_last context["page_last"] = page_last context["page_last_active"] = page >= page_last context["page_next"] = page + 1 if page < page_last else page log(DEBUG, aaa=aaa, app="sirene", view="archive", action="list", status="OK", data="") context["pages"] = pages return render(request, 'app_sirene/archive.html', context)