gestioCOF/shared/views.py

import base64
from collections import namedtuple
from typing import Any

from dal import autocomplete
from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.exceptions import ImproperlyConfigured
from django.http import Http404, HttpResponse
from django.views.generic import TemplateView, View

from shared.autocomplete import ModelSearch

User = get_user_model()


class SympaListView(View):
    realm = "sympa"

    username = settings.SYMPA_USERNAME
    password = settings.SYMPA_PASSWORD

    filters: dict[str, Any] = {}

    def dispatch(self, request, *args, **kwargs):
        if "HTTP_AUTHORIZATION" in request.META:
            auth = request.META["HTTP_AUTHORIZATION"].split()

            if len(auth) == 2 and auth[0].lower() == "basic":
                name, passwd = base64.b64decode(auth[1]).split(b":")

                if name == self.username and passwd == self.password:
                    return self.render_to_response(request, *args, **kwargs)

        return HttpResponse(
            status=401, headers={"WWW-Authenticate": f'Basic realm="{self.realm}"'}
        )

    def render_to_response(self, request, *args, **kwargs):
        """
        Renders a list of emails in a text response.
        """

        users = User.objects.filter(**self.filters)

        return HttpResponse(
            b"\n".join(u.email.encode("utf-8") for u in users if u.email),
            content_type="text/plain",
        )


class Select2QuerySetView(ModelSearch, autocomplete.Select2QuerySetView):
    """Compatibility layer between ModelSearch and Select2QuerySetView."""

    paginate_by = None

    def get_queryset(self):
        keywords = self.q.split()
        return super().search(keywords)


Section = namedtuple("Section", ("name", "verbose_name", "entries"))
Entry = namedtuple("Entry", ("verbose_name", "link"))


class AutocompleteView(TemplateView):
    template_name = "shared/search_results.html"
    search_composer = None

    def get_search_composer(self):
        if self.search_composer is None:
            raise ImproperlyConfigured("Please specify a search composer")
        return self.search_composer

    def get_context_data(self, **kwargs):
        ctx = super().get_context_data(**kwargs)
        if "q" not in self.request.GET:
            raise Http404
        q = self.request.GET["q"]
        ctx["q"] = q
        ctx["results"] = self.search(keywords=q.split())
        return ctx

    def search(self, keywords):
        search_composer = self.get_search_composer()
        raw_results = search_composer.search(keywords)
        sections = []
        for name, unit in search_composer.search_units:
            entries = [
                Entry(
                    verbose_name=unit.result_verbose_name(res),
                    link=unit.result_link(res),
                )
                for res in raw_results[name]
            ]
            if entries:
                sections.append(
                    Section(name=name, verbose_name=unit.verbose_name, entries=entries)
                )
        return sections