# (c) cavaliba.conf - app_sirene - views.py 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 from app_data.aaa import get_aaa, start_ajax, start_view 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 from app_sirene.notify import ( aaa_message_allowed, archive_all, archive_expired, archive_message, count_userlist, get_message_enabled, get_public_active, get_public_default, get_updates, message_from_request, message_notify, message_update, reopen_message, set_placeholder_load, ) # ----------------------------------------- # 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 list # ----------------------------------------- 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) # ----------------------------------------- # anonymous list # ----------------------------------------- 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 = Instance.from_id(id, expand=True) 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"]: # logout request.session.flush() return redirect(context["redirect"]) aaa = context["aaa"] instances = get_message_enabled() pages = [] for instance in instances: if not aaa_message_allowed(instance=instance, aaa=aaa): 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: 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 = Instance.from_id(id, expand=True) 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") if not aaa_message_allowed(instance=instance, aaa=aaa): 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) # ----------------------------------------- # template_picker - V4.0 # ----------------------------------------- def template_picker(request, instance_id=None): """Display a list of sirene_template instances; clicking one navigates to new_message with the selected template id and source instance id.""" context = start_view( request, app="sirene", view="template_picker", noauth="app_sirene:index", perm="p_sirene_new", noauthz="app_sirene:private", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if not instance_id: return redirect("app_sirene:private") 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, "keyname": instance.keyname, "title": instance.displayname, "category": category, "severity": severity, "hint": hint, } templates.append(item) log( DEBUG, aaa=aaa, app="sirene", view="template_picker", action="get", status="OK", data=f"instance={instance_id}", ) context["templates"] = templates context["instance_id"] = instance_id return render(request, "app_sirene/template_picker.html", context) # ----------------------------------------- # new_message # ----------------------------------------- def new_message(request, tid=None, iid=None): """ Create a new sirene_message GET => form ; POST => create/save to DB/notify IN: tid : sirene_template id; if None, blank form iid : source instance id for placeholder substitution (requires tid) 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: context["notify_generic"][fieldname] = [] if tid: # get template to fill blank message template = Instance.from_id(tid) 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 {tid}", ) 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 {tid}", ) 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 context["message_restricted"] = template.is_field_true("is_restricted") # public page context["message_public"] = [] instances = DataInstance.objects.filter(classname="sirene_public", is_enabled=True) for i in instances: selected = i.keyname in template.fields["public_page"].value context["message_public"].append( {"key": i.keyname, "display": i.displayname, "selected": selected} ) # notify_ fields for fieldname in notify_fields: notifyname = "notify_" + fieldname if notifyname not in template.fields: continue instances = DataInstance.objects.filter( classname=fieldname, keyname__in=template.fields[notifyname].value, is_enabled=True ) for i in instances: context["notify_generic"][fieldname].append( {"key": i.keyname, "display": i.displayname, "selected": True} ) # 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 if iid: # load source instance and apply placeholder substitution source_instance = Instance.from_id(iid, expand=True) if not (source_instance and isinstance(source_instance, Instance)): messages.add_message(request, messages.ERROR, _("Instance not found")) log( ERROR, aaa=aaa, app="sirene", view="message", action="new", status="KO", data=f"Invalid instance {iid}", ) return redirect("app_sirene:private") # append source instance to matching notify_generic bucket if source_instance.classname in context["notify_generic"]: sn = source_instance.classname kn = source_instance.keyname dn = source_instance.displayname existing_keys = {item["key"] for item in context["notify_generic"][sn]} if kn not in existing_keys: context["notify_generic"][sn].append( {"key": kn, "display": dn, "selected": True} ) context["message_title"] = set_placeholder_load( context["message_title"], template, source_instance=source_instance ) if context["message_email_content"]: context["message_email_content"] = set_placeholder_load( context["message_email_content"], template, source_instance=source_instance ) if context["message_sms_content"]: context["message_sms_content"] = set_placeholder_load( context["message_sms_content"], template, source_instance=source_instance ) log( DEBUG, aaa=aaa, app="sirene", view="message", action="new", status="OK", data=f"template={tid} instance={iid}", ) else: # blank form with external lists context["message_public"] = [] instances = DataInstance.objects.filter(classname="sirene_public", is_enabled=True) for i in instances: context["message_public"].append( {"key": i.keyname, "display": i.displayname, "selected": False} ) 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 = Instance.from_id(id, expand=True) 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") if not aaa_message_allowed(instance=instance, aaa=aaa): 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 = message_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 = Instance.from_id(id, expand=True) 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 = Instance.from_id(id, expand=True) 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") if not aaa_message_allowed(instance=instance, aaa=aaa): 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)