kadenios/elections/utils.py

import csv
import io
import random

import networkx as nx
import numpy as np
from networkx.algorithms.dag import ancestors, descendants

from django.contrib.auth.hashers import make_password
from django.core.exceptions import ValidationError
from django.core.mail import EmailMessage, get_connection
from django.core.validators import validate_email
from django.template.loader import render_to_string
from django.urls import reverse
from django.utils.translation import gettext_lazy as _

# #############################################################################
#  Fonctions universelles
# #############################################################################


def choices_length(choices):
    """Renvoie la longueur maximale des choix de choices"""
    m = 0
    for c in choices:
        m = max(m, len(c[0]))
    return m


# #############################################################################
#  Classes pour différencier les différents types de questions
# #############################################################################


class CastFunctions:
    """Classe pour enregistrer les votes"""

    def cast_select(user, vote_form):
        """On enregistre un vote classique"""
        selected, n_selected = [], []
        for v in vote_form:
            if v.cleaned_data["selected"]:
                selected.append(v.instance)
            else:
                n_selected.append(v.instance)

        user.votes.add(*selected)
        user.votes.remove(*n_selected)

    def cast_rank(user, vote_form):
        """On enregistre un vote par classement"""
        from .models import Rank, Vote

        # On enregistre les votes pour pouvoir créér les classements
        options = [v.instance for v in vote_form]
        user.votes.add(*options)

        votes = {
            v.option: v for v in Vote.objects.filter(user=user, option__in=options)
        }
        ranks_create = []
        ranks_update = []

        for v in vote_form:
            vote = votes[v.instance]
            if hasattr(vote, "rank"):
                vote.rank.rank = v.cleaned_data["rank"]
                ranks_update.append(vote.rank)
            else:
                ranks_create.append(Rank(vote=vote, rank=v.cleaned_data["rank"]))

        Rank.objects.bulk_update(ranks_update, ["rank"])
        Rank.objects.bulk_create(ranks_create)


class TallyFunctions:
    """Classe pour gérer les dépouillements"""

    def tally_select(question):
        """On dépouille un vote classique"""
        from .models import Option

        max_votes = 0
        options = []

        for o in question.options.prefetch_related("voters").all():
            o.nb_votes = o.voters.count()
            max_votes = max(max_votes, o.nb_votes)
            options.append(o)

        for o in options:
            o.winner = o.nb_votes == max_votes

        question.max_votes = max_votes
        question.save()

        Option.objects.bulk_update(options, ["nb_votes", "winner"])

    def tally_schultze(question):
        """On dépouille un vote par classement et on crée la matrice des duels"""
        from .models import Duel, Option, Rank

        options = list(question.options.all())
        nb_options = len(options)
        ranks = Rank.objects.select_related("vote").filter(vote__option__in=options)
        ranks_by_user = {}

        for r in ranks:
            user = r.vote.user
            if user in ranks_by_user:
                ranks_by_user[user].append(r)
            else:
                ranks_by_user[user] = [r]

        ballots = []

        # Pour chaque votant·e, on regarde son classement
        for user in ranks_by_user:
            votes = ranks_by_user[user]
            ballot = np.zeros((nb_options, nb_options))

            for i in range(nb_options):
                for j in range(i):
                    ballot[i, j] = int(votes[i].rank < votes[j].rank)
                    ballot[j, i] = int(votes[j].rank < votes[i].rank)

            ballots.append(ballot)

        if ballots != []:
            # On additionne les classements de tout le monde pour avoir la matrice
            # des duels
            duels = sum(ballots)

            # Configuration du graphe
            graph = nx.DiGraph()

            graph.add_nodes_from(options)

            cells = []

            # On enregistre la matrice
            for i in range(nb_options):
                for j in range(nb_options):
                    if duels[i, j] > duels[j, i]:
                        graph.add_edge(
                            options[i], options[j], weight=(duels[i, j] - duels[j, i])
                        )

                    if duels[i, j] > 0:
                        cells.append(
                            Duel(
                                question=question,
                                winner=options[i],
                                loser=options[j],
                                amount=duels[i, j],
                            )
                        )
            Duel.objects.bulk_create(cells)

            # On utilise la méthode de Schwartz pour trouver les options gagnantes
            while graph.edges():
                losers = set()  # Les options qui se font battre strictement
                for n in graph.nodes():
                    losers |= set(descendants(graph, n)) - set(ancestors(graph, n))

                if losers:
                    # On supprime les options perdantes
                    for n in losers:
                        graph.remove_node(n)

                else:
                    # On n'a pas d'options perdantes, on supprime les arêtes de poids
                    # le plus faible
                    min_weight = min(nx.get_edge_attributes(graph, "weight").values())
                    min_edges = []
                    for (u, v) in graph.edges():
                        if graph[u][v]["weight"] == min_weight:
                            min_edges.append((u, v))

                    for (u, v) in min_edges:
                        graph.remove_edge(u, v)

            # Les options gagnantes sont celles encore présentes dans le graphe
            winners = graph.nodes()
            for o in options:
                o.winner = o in winners

            Option.objects.bulk_update(options, ["winner"])


class ValidateFunctions:
    """Classe pour valider les formsets selon le type de question"""

    def always_true(vote_form):
        """Retourne True pour les votes sans validation particulière"""
        return True

    def unique_selected(vote_form):
        """Vérifie qu'une seule option est choisie"""
        nb_selected = 0
        for v in vote_form:
            nb_selected += v.cleaned_data["selected"]

        if nb_selected == 0:
            vote_form._non_form_errors.append(
                ValidationError(_("Vous devez sélectionnner une option."))
            )
            return False
        elif nb_selected > 1:
            vote_form._non_form_errors.append(
                ValidationError(_("Vous ne pouvez pas sélectionner plus d'une option."))
            )
            return False
        return True

    def limit_ranks(vote_form):
        """Limite le classement au nombre d'options"""
        nb_options = len(vote_form)
        valid = True

        for v in vote_form:
            rank = v.cleaned_data["rank"]
            if rank is None:
                rank = nb_options

            if rank > nb_options:
                v.add_error(
                    "rank", _("Le classement maximal est {}.").format(nb_options)
                )
                valid = False
            elif rank < 1:
                v.add_error("rank", _("Le classement minimal est 1."))
                valid = False
            v.cleaned_data["rank"] = rank  # On ajoute le défaut

        return valid


class ResultsData:
    """Classe pour afficher des informations supplémentaires après la fin d'une élection"""

    def select(question):
        """On renvoie l'explication des couleurs"""
        return render_to_string("elections/results/select.html")

    def rank(question):
        """On récupère la matrice des résultats et on l'affiche"""
        duels = question.duels.all()
        options = list(question.options.all())
        n = len(options)

        matrix = np.zeros((n, n), dtype=int)

        for d in duels:
            i, j = options.index(d.loser), options.index(d.winner)
            matrix[i, j] = d.amount

        return render_to_string(
            "elections/results/rank.html",
            {"q": question, "matrix": matrix, "range": range(1, n + 1)},
        )


# #############################################################################
#  Fonctions pour importer une liste de votant·e·s
# #############################################################################


def create_users(election, csv_file):
    """Crée les votant·e·s pour l'élection donnée, en remplissant les champs
    `username`, `election` et `full_name`.
    """
    dialect = csv.Sniffer().sniff(csv_file.readline().decode("utf-8"))
    csv_file.seek(0)
    reader = csv.reader(io.StringIO(csv_file.read().decode("utf-8")), dialect)
    for (username, full_name, email) in reader:
        election.registered_voters.create(
            username=f"{election.id}__{username}", email=email, full_name=full_name
        )


def check_csv(csv_file):
    """Vérifie que le fichier donnant la liste de votant·e·s est bien formé"""
    try:
        dialect = csv.Sniffer().sniff(csv_file.readline().decode("utf-8"))
    except csv.Error:
        return [
            _(
                "Format invalide. Vérifiez que le fichier est bien formé (i.e. "
                "chaque ligne de la forme 'login,nom,email')."
            )
        ]
    csv_file.seek(0)
    reader = csv.reader(io.StringIO(csv_file.read().decode("utf-8")), dialect)

    errors = []
    users = {}
    line_nb = 0
    for line in reader:
        line_nb += 1
        if len(line) != 3:
            errors.append(
                _("La ligne {} n'a pas le bon nombre d'éléments.").format(line_nb)
            )
        else:
            if line[0] == "":
                errors.append(
                    _("Valeur manquante dans la ligne {} : 'login'.").format(line_nb)
                )
            else:
                if line[0] in users:
                    errors.append(
                        _("Doublon dans les logins : lignes {} et {}.").format(
                            line_nb, users[line[0]]
                        )
                    )
                else:
                    users[line[0]] = line_nb
            if line[1] == "":
                errors.append(
                    _("Valeur manquante dans la ligne {} : 'nom'.").format(line_nb)
                )
            try:
                validate_email(line[2])
            except ValidationError:
                errors.append(
                    _("Adresse mail invalide à la ligne {} : '{}'.").format(
                        line_nb, line[2]
                    )
                )

    return errors


def generate_password():
    random.seed()
    alphabet = "abcdefghjkmnopqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ23456789"
    password = ""
    for i in range(15):
        password += random.choice(alphabet)

    return password


def send_mail(election, mail_form):
    """Envoie le mail d'annonce de l'élection avec identifiants et mot de passe
    aux votant·e·s, le mdp est généré en même temps que le mail est envoyé.
    """
    from .models import User

    voters = list(election.registered_voters.all())
    e_url = reverse("election.view", args=[election.id])
    url = f"https://vote.eleves.ens.fr/{e_url}"
    messages = []
    for v in voters:
        password = generate_password()
        v.password = make_password(password)
        messages.append(
            EmailMessage(
                subject=mail_form.cleaned_data["objet"],
                body=mail_form.cleaned_data["message"].format(
                    full_name=v.full_name,
                    election_url=url,
                    username=v.get_username(),
                    password=password,
                ),
                to=[v.email],
            )
        )
    get_connection(fail_silently=False).send_messages(messages)
    User.objects.bulk_update(voters, ["password"])