import urllib import uuid import orjson from django.conf import settings from django.http import HttpResponse, JsonResponse from django.shortcuts import render from django.urls import reverse from django_tables2 import SingleTableView from rest_framework.parsers import FormParser from rest_framework.views import APIView from core.db.storage import db from core.lib.threshold import ( annotate_num_chans, annotate_num_users, get_chans, get_users, ) from core.views import helpers from core.views.ui.tables import DrilldownTable # from copy import deepcopy def parse_dates(dates): spl = dates.split(" - ") if all(spl): spl = [f"{x.replace(' ', 'T')}" for x in spl] if not len(spl) == 2: message = "Invalid dates" message_class = "danger" return {"message": message, "class": message_class} from_ts, to_ts = spl from_date, from_time = from_ts.split("T") to_date, to_time = to_ts.split("T") return { "from_date": from_date, "to_date": to_date, "from_time": from_time, "to_time": to_time, } def create_tags(query): """ Grab the tags out of the query and make a list we can add to the Bulma tags element when the page loads. """ spl = query.split("AND") spl = [x.strip() for x in spl if ":" in x] spl = [x.replace('"', "") for x in spl] tags = [f"{tag}: {elem}" for tag, elem in [x.split(":")[:2] for x in spl]] return tags def parse_tags(tags_pre): """ Parse the tags from the variable tags_pre. """ tags = [] tags_spl = tags_pre.split(",") if tags_spl: for tag in tags_spl: tag = tag.split(": ") if len(tag) == 2: key, val = tag tags.append({key: val}) return tags def make_table(context): table = DrilldownTable(context["object_list"]) context["table"] = table # del context["results"] return context def make_graph(results): graph = [] for index, item in enumerate(results): date = str(index) graph.append( { "text": item.get("words_noun", None) or item.get("msg", None) or item.get("id"), "nick": item.get("nick", None), "channel": item.get("channel", None), "net": item.get("net", None), "value": item.get("sentiment", None) or None, "date": date, } ) return orjson.dumps(graph).decode("utf-8") class DrilldownTableView(SingleTableView): table_class = DrilldownTable template_name = "wm/widget.html" window_content = "window-content/results.html" # htmx_partial = "partials/" paginate_by = settings.DRILLDOWN_RESULTS_PER_PAGE def common_request(self, request, **kwargs): extra_params = {} if request.user.is_anonymous: sizes = settings.MAIN_SIZES_ANON else: sizes = settings.MAIN_SIZES if request.GET: self.template_name = "index.html" # GET arguments in URL like ?query=xyz query_params = request.GET.dict() if request.htmx: if request.resolver_match.url_name == "search_partial": self.template_name = "partials/results_table.html" elif request.POST: query_params = request.POST.dict() else: self.template_name = "index.html" # No query, this is a fresh page load # Don't try to search, since there's clearly nothing to do params_with_defaults = {} helpers.add_defaults(params_with_defaults) context = { "sizes": sizes, "params": params_with_defaults, "unique": "results", "window_content": self.window_content, "title": "Results", } return render(request, self.template_name, context) # Merge everything together just in case tmp_post = request.POST.dict() tmp_get = request.GET.dict() tmp_post = {k: v for k, v in tmp_post.items() if v and not v == "None"} tmp_get = {k: v for k, v in tmp_get.items() if v and not v == "None"} query_params.update(tmp_post) query_params.update(tmp_get) # URI we're passing to the template for linking if "csrfmiddlewaretoken" in query_params: del query_params["csrfmiddlewaretoken"] # Parse the dates if "dates" in query_params: dates = parse_dates(query_params["dates"]) del query_params["dates"] if dates: if "message" in dates: return render(request, self.template_name, dates) query_params["from_date"] = dates["from_date"] query_params["to_date"] = dates["to_date"] query_params["from_time"] = dates["from_time"] query_params["to_time"] = dates["to_time"] # Remove null values if "query" in query_params: if query_params["query"] == "": del query_params["query"] # Remove null tags values if "tags" in query_params: if query_params["tags"] == "": del query_params["tags"] else: # Parse the tags and populate cast to pass to search function tags = parse_tags(query_params["tags"]) extra_params["tags"] = tags context = db.query_results(request, query_params, **extra_params) # Unique is for identifying the widgets. # We don't want a random one since we only want one results pane. context["unique"] = "results" context["window_content"] = self.window_content context["title"] = "Results" # Valid sizes context["sizes"] = sizes # Add any default parameters to the context params_with_defaults = dict(query_params) helpers.add_defaults(params_with_defaults) context["params"] = params_with_defaults # Remove anything that we or the user set to a default for # pretty URLs helpers.remove_defaults(query_params) url_params = urllib.parse.urlencode(query_params) context["client_uri"] = url_params # There's an error if "message" in context: response = render(request, self.template_name, context) # Still push the URL so they can share it to get assistance if request.GET: if request.htmx: response["HX-Push"] = reverse("home") + "?" + url_params elif request.POST: response["HX-Push"] = reverse("home") + "?" + url_params return response # Create data for chart.js sentiment graph graph = make_graph(context["object_list"]) context["data"] = graph # Create the table context = make_table(context) # URI we're passing to the template for linking, table fields removed table_fields = ["page", "sort"] clean_params = {k: v for k, v in query_params.items() if k not in table_fields} clean_url_params = urllib.parse.urlencode(clean_params) context["uri"] = clean_url_params # unique = str(uuid.uuid4())[:8] # self.context = context return context def get(self, request, *args, **kwargs): self.context = self.common_request(request) if isinstance(self.context, HttpResponse): return self.context self.object_list = self.context["object_list"] show = [] show = set().union(*(d.keys() for d in self.object_list)) allow_empty = self.get_allow_empty() if not allow_empty: # When pagination is enabled and object_list is a queryset, # it's better to do a cheap query than to load the unpaginated # queryset in memory. if self.get_paginate_by(self.object_list) is not None and hasattr( self.object_list, "exists" ): is_empty = not self.object_list.exists() # noqa else: is_empty = not self.object_list # noqa context = self.get_context_data() for k, v in self.context.items(): if k not in context: context[k] = v context["show"] = show # if request.htmx: # self.template_name = self.window_content # if request.method == "GET": # if not request.htmx: # self.template_name = "ui/drilldown/drilldown.html" response = self.render_to_response(context) # if not request.method == "GET": if "client_uri" in context: response["HX-Push"] = reverse("home") + "?" + context["client_uri"] return response def post(self, request, *args, **kwargs): return self.get(request, *args, **kwargs) # class Drilldown(View): # template_name = "ui/drilldown/drilldown.html" # plan_name = "drilldown" # def get(self, request): # return drilldown_search(request) # def post(self, request): # return drilldown_search(request) class DrilldownContextModal(APIView): parser_classes = [FormParser] plan_name = "drilldown" template_name = "modals/context.html" def post(self, request): if request.resolver_match.url_name == "modal_context_table": self.template_name = "partials/context_table.html" size = 15 nicks_sensitive = None query = False # Create the query params from the POST arguments mandatory = [ "net", "channel", "num", "source", "index", "nick", "type", "mtype", ] invalid = [None, False, "—", "None"] query_params = {k: v for k, v in request.data.items() if v} for key in query_params: if query_params[key] in invalid: query_params[key] = None for key in mandatory: if key not in query_params: query_params[key] = None # Lookup the hash values but don't disclose them to the user # if settings.HASHING: # if query_params["source"] not in settings.SAFE_SOURCES: # SAFE_PARAMS = deepcopy(query_params) # hash_lookup(request.user, SAFE_PARAMS) # else: # SAFE_PARAMS = deepcopy(query_params) # else: # SAFE_PARAMS = query_params type = None if request.user.is_superuser: if "type" in query_params: type = query_params["type"] if type == "znc": query_params["channel"] = "*status" # SAFE_PARAMS["channel"] = "*status" if type in ["query", "notice"]: nicks_sensitive = [ query_params["channel"], query_params["nick"], ] # UNSAFE # nicks = [query_params["channel"], query_params["nick"]] query = True if ( query_params["index"] == "internal" and query_params["mtype"] == "msg" and not type == "query" ): query_params["index"] = "main" # SAFE_PARAMS["index"] = "main" if query_params["type"] in ["znc", "auth"]: query = True if not request.user.is_superuser: query_params["index"] = "main" # SAFE_PARAMS["index"] = "main" query_params["sorting"] = "desc" # SAFE_PARAMS["sorting"] = "desc" # query_params["size"] = size annotate = False if query_params["source"] == "irc": if query_params["type"] not in ["znc", "auth"]: annotate = True # Create the query with the context helper if "num" in query_params: if query_params["num"]: if query_params["num"].isdigit(): query_params["num"] = int(query_params["num"]) else: return {"message": "Invalid num value", "class": "danger"} search_query = db.construct_context_query( query_params["index"], query_params["net"], query_params["channel"], query_params["source"], query_params["num"], size, type=type, nicks=nicks_sensitive, ) results = db.query_results( request, query_params, size=size, annotate=annotate, custom_query=search_query, reverse=True, dedup_fields=["net", "type", "msg"], ) if "message" in results: return render(request, self.template_name, results) unique = str(uuid.uuid4())[:8] context = { "net": query_params["net"], "channel": query_params["channel"], "source": query_params["source"], "ts": f"{query_params['date']} {query_params['time']}", "object_list": results["object_list"], "time": query_params["time"], "date": query_params["date"], "index": query_params["index"], "num": query_params["num"], "type": query_params["type"], "mtype": query_params["mtype"], "nick": query_params["nick"], "params": query_params, "unique": unique, } if request.user.is_superuser: if query: context["query"] = True return render(request, self.template_name, context) class ThresholdInfoModal(APIView): parser_classes = [FormParser] plan_name = "drilldown" template_name = "modals/drilldown.html" def post(self, request, type=None): # if not request.user.has_plan(self.plan_name): # return JsonResponse({"success": False}) if "net" not in request.data: return JsonResponse({"success": False}) if "nick" not in request.data: return JsonResponse({"success": False}) if "channel" not in request.data: return JsonResponse({"success": False}) if type == "window": self.template_name = "windows/drilldown.html" elif type == "widget": self.template_name = "widgets/drilldown.html" net = request.data["net"] nick = request.data["nick"] channel = request.data["channel"] channels = get_chans(net, [nick]) users = get_users(net, [nick]) num_users = annotate_num_users(net, channels) num_chans = annotate_num_chans(net, users) if channels: inter_users = get_users(net, channels) else: inter_users = [] if users: inter_chans = get_chans(net, users) else: inter_chans = [] unique = str(uuid.uuid4())[:8] context = { "net": net, "nick": nick, "channel": channel, "chans": channels, "users": users, "inter_chans": inter_chans, "inter_users": inter_users, "num_users": num_users, "num_chans": num_chans, "unique": unique, } return render(request, self.template_name, context)