import csv import uuid from datetime import timedelta from smtplib import SMTPRecipientsRefused from urllib.parse import parse_qs, urlencode, urlparse, urlunparse from django.contrib import messages from django.contrib.auth.decorators import login_required from django.contrib.auth.mixins import LoginRequiredMixin from django.contrib.auth.models import User from django.contrib.auth.views import ( LoginView as DjangoLoginView, LogoutView as DjangoLogoutView, redirect_to_login, ) from django.contrib.sites.models import Site from django.core.mail import send_mail from django.http import Http404, HttpResponse, HttpResponseForbidden from django.shortcuts import get_object_or_404, redirect, render from django.template import loader from django.urls import reverse_lazy from django.utils import timezone from django.utils.translation import ugettext_lazy as _ from django.views.generic import FormView, TemplateView from django_cas_ng.views import LogoutView as CasLogoutView from icalendar import Calendar, Event as Vevent from bda.models import Spectacle, Tirage from gestioncof.autocomplete import cof_autocomplete from gestioncof.decorators import BuroRequiredMixin, buro_required, cof_required from gestioncof.forms import ( CalendarForm, ClubsForm, EventForm, EventFormset, EventStatusFilterForm, ExteAuthenticationForm, GestioncofConfigForm, PhoneForm, ProfileForm, RegistrationPassUserForm, RegistrationProfileForm, RegistrationUserForm, SurveyForm, SurveyStatusFilterForm, UserForm, ) from gestioncof.models import ( CalendarSubscription, Club, CofProfile, Event, EventCommentField, EventCommentValue, EventOption, EventOptionChoice, EventRegistration, Survey, SurveyAnswer, SurveyQuestion, SurveyQuestionAnswer, ) from shared.views import AutocompleteView, Select2QuerySetView class HomeView(LoginRequiredMixin, TemplateView): template_name = "gestioncof/home.html" def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) context["surveys"] = Survey.objects.filter(old=False) context["events"] = Event.objects.filter(old=False) context["open_surveys"] = Survey.objects.filter(survey_open=True, old=False) context["active_tirages"] = Tirage.objects.filter(active=True) context["open_tirages"] = Tirage.objects.filter( active=True, ouverture__lte=timezone.now() ) context["now"] = timezone.now() return context def login(request): if request.user.is_authenticated: return redirect("home") context = {} if request.method == "GET" and "next" in request.GET: context["next"] = request.GET["next"] return render(request, "login_switch.html", context) class LoginExtView(DjangoLoginView): template_name = "login.html" form_class = ExteAuthenticationForm def form_invalid(self, form): for e in form.non_field_errors().as_data(): if e.code in ["has_clipper", "no_password"]: return render(self.request, "login_error.html", {"error_code": e.code}) return super().form_invalid(form) class CustomCasLogoutView(CasLogoutView): """ Actuellement, le CAS de l'ENS est pété et n'a pas le bon paramètre GET pour rediriger après déconnexion. On change la redirection à la main dans la vue de logout. """ def get(self, request): # CasLogoutView.get() retourne un HttpResponseRedirect response = super().get(request) parse_result = urlparse(response.url) qd = parse_qs(parse_result.query) if "url" in qd.keys(): # Le 2e pop est nécessaire car CAS n'aime pas # les paramètres sous forme de liste qd["service"] = qd.pop("url").pop() # La méthode _replace est documentée ! new_url = parse_result._replace(query=urlencode(qd)) return redirect(urlunparse(new_url)) @login_required def logout(request, next_page=None): if next_page is None: next_page = request.GET.get("next", None) profile = getattr(request.user, "profile", None) if profile and profile.login_clipper: if next_page is None: # On ne voit pas les messages quand on se déconnecte de CAS msg = None else: msg = _("Déconnexion de GestioCOF et CAS réussie. À bientôt {}.") logout_view = CustomCasLogoutView.as_view() else: msg = _("Déconnexion de GestioCOF réussie. À bientôt {}.") logout_view = DjangoLogoutView.as_view( next_page=next_page, template_name="logout.html" ) if msg is not None: messages.success(request, msg.format(request.user.get_short_name())) return logout_view(request) @login_required def survey(request, survey_id): survey = get_object_or_404( Survey.objects.prefetch_related("questions", "questions__answers"), id=survey_id ) if not survey.survey_open or survey.old: raise Http404 success = False deleted = False if request.method == "POST": form = SurveyForm(request.POST, survey=survey) if request.POST.get("delete"): try: current_answer = SurveyAnswer.objects.get( user=request.user, survey=survey ) current_answer.delete() current_answer = None except SurveyAnswer.DoesNotExist: current_answer = None form = SurveyForm(survey=survey) success = True deleted = True else: if form.is_valid(): all_answers = [] for question_id, answers_ids in form.answers(): question = get_object_or_404( SurveyQuestion, id=question_id, survey=survey ) if type(answers_ids) != list: answers_ids = [answers_ids] if not question.multi_answers and len(answers_ids) > 1: raise Http404 for answer_id in answers_ids: if not answer_id: continue answer_id = int(answer_id) answer = SurveyQuestionAnswer.objects.get( id=answer_id, survey_question=question ) all_answers.append(answer) try: current_answer = SurveyAnswer.objects.get( user=request.user, survey=survey ) except SurveyAnswer.DoesNotExist: current_answer = SurveyAnswer(user=request.user, survey=survey) current_answer.save() current_answer.answers.set(all_answers) current_answer.save() success = True else: try: current_answer = SurveyAnswer.objects.get(user=request.user, survey=survey) form = SurveyForm(survey=survey, current_answers=current_answer.answers) except SurveyAnswer.DoesNotExist: current_answer = None form = SurveyForm(survey=survey) # Messages if success: if deleted: messages.success(request, "Votre réponse a bien été supprimée") else: messages.success( request, "Votre réponse a bien été enregistrée ! Vous " "pouvez cependant la modifier jusqu'à la fin " "du sondage.", ) return render( request, "gestioncof/survey.html", {"survey": survey, "form": form, "current_answer": current_answer}, ) def get_event_form_choices(event, form): all_choices = [] for option_id, choices_ids in form.choices(): option = get_object_or_404(EventOption, id=option_id, event=event) if type(choices_ids) != list: choices_ids = [choices_ids] if not option.multi_choices and len(choices_ids) > 1: raise Http404 for choice_id in choices_ids: if not choice_id: continue choice_id = int(choice_id) choice = EventOptionChoice.objects.get(id=choice_id, event_option=option) all_choices.append(choice) return all_choices def update_event_form_comments(event, form, registration): for commentfield_id, value in form.comments(): field = get_object_or_404(EventCommentField, id=commentfield_id, event=event) if value == field.default: continue (storage, _) = EventCommentValue.objects.get_or_create( commentfield=field, registration=registration ) storage.content = value storage.save() @login_required def event(request, event_id): event = get_object_or_404(Event, id=event_id) if (not event.registration_open) or event.old: raise Http404 success = False if request.method == "POST": form = EventForm(request.POST, event=event) if form.is_valid(): all_choices = get_event_form_choices(event, form) (current_registration, _) = EventRegistration.objects.get_or_create( user=request.user, event=event ) current_registration.options.set(all_choices) current_registration.save() success = True else: try: current_registration = EventRegistration.objects.get( user=request.user, event=event ) form = EventForm(event=event, current_choices=current_registration.options) except EventRegistration.DoesNotExist: form = EventForm(event=event) # Messages if success: messages.success( request, "Votre inscription a bien été enregistrée ! " "Vous pouvez cependant la modifier jusqu'à " "la fin des inscriptions.", ) return render(request, "gestioncof/event.html", {"event": event, "form": form}) def clean_post_for_status(initial): d = initial.copy() for k, v in d.items(): if k.startswith("id_"): del d[k] d[k[3:]] = v return d @buro_required def event_status(request, event_id): event = get_object_or_404(Event, id=event_id) registrations_query = EventRegistration.objects.filter(event=event) post_data = clean_post_for_status(request.POST) form = EventStatusFilterForm(post_data or None, event=event) if form.is_valid(): for option_id, choice_id, value in form.filters(): if option_id == "has_paid": if value == "yes": registrations_query = registrations_query.filter(paid=True) elif value == "no": registrations_query = registrations_query.filter(paid=False) continue choice = get_object_or_404( EventOptionChoice, id=choice_id, event_option__id=option_id ) if value == "none": continue if value == "yes": registrations_query = registrations_query.filter( options__id__exact=choice.id ) elif value == "no": registrations_query = registrations_query.exclude( options__id__exact=choice.id ) user_choices = registrations_query.prefetch_related("user").all() options = EventOption.objects.filter(event=event).all() choices_count = {} for option in options: for choice in option.choices.all(): choices_count[choice.id] = 0 for user_choice in user_choices: for choice in user_choice.options.all(): choices_count[choice.id] += 1 return render( request, "event_status.html", { "event": event, "user_choices": user_choices, "options": options, "choices_count": choices_count, "form": form, }, ) @buro_required def survey_status(request, survey_id): survey = get_object_or_404(Survey, id=survey_id) answers_query = SurveyAnswer.objects.filter(survey=survey) post_data = clean_post_for_status(request.POST) form = SurveyStatusFilterForm(post_data or None, survey=survey) if form.is_valid(): for question_id, answer_id, value in form.filters(): answer = get_object_or_404( SurveyQuestionAnswer, id=answer_id, survey_question__id=question_id ) if value == "none": continue if value == "yes": answers_query = answers_query.filter(answers__id__exact=answer.id) elif value == "no": answers_query = answers_query.exclude(answers__id__exact=answer.id) user_answers = answers_query.prefetch_related("user").all() questions = SurveyQuestion.objects.filter(survey=survey).all() answers_count = {} for question in questions: for answer in question.answers.all(): answers_count[answer.id] = 0 for user_answer in user_answers: for answer in user_answer.answers.all(): answers_count[answer.id] += 1 return render( request, "survey_status.html", { "survey": survey, "user_answers": user_answers, "questions": questions, "answers_count": answers_count, "form": form, }, ) @login_required def profile(request): user = request.user data = request.POST if request.method == "POST" else None user_form = UserForm(data=data, instance=user, prefix="u") profile_form_klass = ProfileForm if user.profile.is_cof else PhoneForm profile_form = profile_form_klass(data=data, instance=user.profile, prefix="p") if request.method == "POST": if user_form.is_valid() and profile_form.is_valid(): user_form.save() profile_form.save() messages.success(request, _("Votre profil a été mis à jour avec succès !")) context = {"user_form": user_form, "profile_form": profile_form} return render(request, "gestioncof/profile.html", context) def registration_set_ro_fields(user_form, profile_form): user_form.fields["username"].widget.attrs["readonly"] = True profile_form.fields["login_clipper"].widget.attrs["readonly"] = True @buro_required def registration_form2(request, login_clipper=None, username=None, fullname=None): events = Event.objects.filter(old=False).all() member = None if login_clipper: try: # check if the given user is already registered member = User.objects.get(username=login_clipper) username = member.username login_clipper = None except User.DoesNotExist: # new user, but prefill # user user_form = RegistrationUserForm( initial={ "username": login_clipper, "email": "%s@clipper.ens.fr" % login_clipper, } ) if fullname: bits = fullname.split(" ") user_form.fields["first_name"].initial = bits[0] if len(bits) > 1: user_form.fields["last_name"].initial = " ".join(bits[1:]) # profile profile_form = RegistrationProfileForm( initial={"login_clipper": login_clipper} ) registration_set_ro_fields(user_form, profile_form) # events & clubs event_formset = EventFormset(events=events, prefix="events") clubs_form = ClubsForm() if username: member = get_object_or_404(User, username=username) (profile, _) = CofProfile.objects.get_or_create(user=member) # already existing, prefill user_form = RegistrationUserForm(instance=member) profile_form = RegistrationProfileForm(instance=profile) registration_set_ro_fields(user_form, profile_form) # events current_registrations = [] for event in events: try: current_registrations.append( EventRegistration.objects.get(user=member, event=event) ) except EventRegistration.DoesNotExist: current_registrations.append(None) event_formset = EventFormset( events=events, prefix="events", current_registrations=current_registrations ) # Clubs clubs_form = ClubsForm(initial={"clubs": member.clubs.all()}) elif not login_clipper: # new user user_form = RegistrationPassUserForm() profile_form = RegistrationProfileForm() event_formset = EventFormset(events=events, prefix="events") clubs_form = ClubsForm() return render( request, "gestioncof/registration_form.html", { "member": member, "login_clipper": login_clipper, "user_form": user_form, "profile_form": profile_form, "event_formset": event_formset, "clubs_form": clubs_form, }, ) def notify_new_member(request, member: User): if not member.email: messages.warning( request, "GestioCOF n'a pas d'adresse mail pour {}, ".format(member) + "aucun email de bienvenue n'a été envoyé", ) return # Try to send a welcome email and report SMTP errors try: send_mail( "Bienvenue au COF", loader.render_to_string( "gestioncof/mails/welcome.txt", context={"member": member} ), "cof@ens.fr", [member.email], ) except SMTPRecipientsRefused: messages.error( request, "Error lors de l'envoi de l'email de bienvenue à {} ({})".format( member, member.email ), ) @buro_required def registration(request): if request.POST: request_dict = request.POST.copy() member = None login_clipper = None # ----- # Remplissage des formulaires # ----- if "password1" in request_dict or "password2" in request_dict: user_form = RegistrationPassUserForm(request_dict) else: user_form = RegistrationUserForm(request_dict) profile_form = RegistrationProfileForm(request_dict) clubs_form = ClubsForm(request_dict) events = Event.objects.filter(old=False).all() event_formset = EventFormset(events=events, data=request_dict, prefix="events") if "user_exists" in request_dict and request_dict["user_exists"]: username = request_dict["username"] try: member = User.objects.get(username=username) user_form = RegistrationUserForm(request_dict, instance=member) if member.profile.login_clipper: login_clipper = member.profile.login_clipper except User.DoesNotExist: pass else: pass # ----- # Validation des formulaires # ----- if user_form.is_valid(): member = user_form.save() profile, _ = CofProfile.objects.get_or_create(user=member) was_cof = profile.is_cof # Maintenant on remplit le formulaire de profil profile_form = RegistrationProfileForm(request_dict, instance=profile) if ( profile_form.is_valid() and event_formset.is_valid() and clubs_form.is_valid() ): # Enregistrement du profil profile = profile_form.save() if profile.is_cof and not was_cof: notify_new_member(request, member) # Enregistrement des inscriptions aux événements for form in event_formset: if "status" not in form.cleaned_data: form.cleaned_data["status"] = "no" if form.cleaned_data["status"] == "no": try: current_registration = EventRegistration.objects.get( user=member, event=form.event ) current_registration.delete() except EventRegistration.DoesNotExist: pass continue all_choices = get_event_form_choices(form.event, form) ( current_registration, created_reg, ) = EventRegistration.objects.get_or_create( user=member, event=form.event ) update_event_form_comments(form.event, form, current_registration) current_registration.options.set(all_choices) current_registration.paid = form.cleaned_data["status"] == "paid" current_registration.save() # if form.event.title == "Mega 15" and created_reg: # field = EventCommentField.objects.get( # event=form.event, name="Commentaires") # try: # comments = EventCommentValue.objects.get( # commentfield=field, # registration=current_registration).content # except EventCommentValue.DoesNotExist: # comments = field.default # FIXME : il faut faire quelque chose de propre ici, # par exemple écrire un mail générique pour # l'inscription aux événements et/ou donner la # possibilité d'associer un mail aux événements # send_custom_mail(...) # Enregistrement des inscriptions aux clubs member.clubs.clear() for club in clubs_form.cleaned_data["clubs"]: club.membres.add(member) club.save() # --- # Success # --- msg = ( "L'inscription de {:s} ({:s}) a été " "enregistrée avec succès.".format( member.get_full_name(), member.email ) ) if profile.is_cof: msg += "\nIl est désormais membre du COF n°{:d} !".format( member.profile.id ) messages.success(request, msg, extra_tags="safe") return render( request, "gestioncof/registration_post.html", { "user_form": user_form, "profile_form": profile_form, "member": member, "login_clipper": login_clipper, "event_formset": event_formset, "clubs_form": clubs_form, }, ) else: return render(request, "registration.html") # ----- # Clubs # ----- @login_required def membres_club(request, name): # Vérification des permissions : l'utilisateur doit être membre du burô # ou respo du club. user = request.user club = get_object_or_404(Club, name=name) if not request.user.profile.is_buro and club not in user.clubs_geres.all(): return HttpResponseForbidden("

Permission denied

") members_no_respo = club.membres.exclude(clubs_geres=club).all() return render( request, "membres_clubs.html", {"club": club, "members_no_respo": members_no_respo}, ) @buro_required def change_respo(request, club_name, user_id): club = get_object_or_404(Club, name=club_name) user = get_object_or_404(User, id=user_id) if user in club.respos.all(): club.respos.remove(user) elif user in club.membres.all(): club.respos.add(user) else: raise Http404 return redirect("membres-club", name=club_name) @cof_required def liste_clubs(request): clubs = Club.objects if request.user.profile.is_buro: data = {"owned_clubs": clubs.all()} else: data = { "owned_clubs": request.user.clubs_geres.all(), "other_clubs": clubs.exclude(respos=request.user), } return render(request, "liste_clubs.html", data) @buro_required def export_members(request): response = HttpResponse(content_type="text/csv") response["Content-Disposition"] = "attachment; filename=membres_cof.csv" writer = csv.writer(response) for profile in CofProfile.objects.filter(is_cof=True).all(): user = profile.user bits = [ user.id, user.username, user.first_name, user.last_name, user.email, profile.phone, profile.occupation, profile.departement, profile.type_cotiz, ] writer.writerow([str(bit) for bit in bits]) return response # ---------------------------------------- # Début des exports Mega machins hardcodés # ---------------------------------------- MEGA_YEAR = 2018 MEGA_EVENT_NAME = "MEGA 2018" MEGA_COMMENTFIELD_NAME = "Commentaires" MEGA_CONSCRITORGAFIELD_NAME = "Orga ? Conscrit ?" MEGA_CONSCRIT = "Conscrit" MEGA_ORGA = "Orga" def csv_export_mega(filename, qs): response = HttpResponse(content_type="text/csv") response["Content-Disposition"] = "attachment; filename=" + filename writer = csv.writer(response) for reg in qs.all(): user = reg.user profile = user.profile comments = "---".join([comment.content for comment in reg.comments.all()]) bits = [ user.username, user.first_name, user.last_name, user.email, profile.phone, user.id, profile.comments if profile.comments else "", comments, ] writer.writerow([str(bit) for bit in bits]) return response @buro_required def export_mega_remarksonly(request): filename = "remarques_mega_{}.csv".format(MEGA_YEAR) response = HttpResponse(content_type="text/csv") response["Content-Disposition"] = "attachment; filename=" + filename writer = csv.writer(response) event = Event.objects.get(title=MEGA_EVENT_NAME) commentfield = event.commentfields.get(name=MEGA_COMMENTFIELD_NAME) for val in commentfield.values.all(): reg = val.registration user = reg.user profile = user.profile bits = [ user.username, user.first_name, user.last_name, user.email, profile.phone, profile.id, profile.comments, val.content, ] writer.writerow([str(bit) for bit in bits]) return response # @buro_required # def export_mega_bytype(request, type): # types = {"orga-actif": "Orga élève", # "orga-branleur": "Orga étudiant", # "conscrit-eleve": "Conscrit élève", # "conscrit-etudiant": "Conscrit étudiant"} # # if type not in types: # raise Http404 # # event = Event.objects.get(title="MEGA 2017") # type_option = event.options.get(name="Type") # participant_type = type_option.choices.get(value=types[type]).id # qs = EventRegistration.objects.filter(event=event).filter( # options__id__exact=participant_type) # return csv_export_mega(type + '_mega_2017.csv', qs) @buro_required def export_mega_orgas(request): event = Event.objects.get(title=MEGA_EVENT_NAME) type_option = event.options.get(name=MEGA_CONSCRITORGAFIELD_NAME) participant_type = type_option.choices.get(value=MEGA_ORGA).id qs = EventRegistration.objects.filter(event=event).filter( options__id=participant_type ) return csv_export_mega("orgas_mega_{}.csv".format(MEGA_YEAR), qs) @buro_required def export_mega_participants(request): event = Event.objects.get(title=MEGA_EVENT_NAME) type_option = event.options.get(name=MEGA_CONSCRITORGAFIELD_NAME) participant_type = type_option.choices.get(value=MEGA_CONSCRIT).id qs = EventRegistration.objects.filter(event=event).filter( options__id=participant_type ) return csv_export_mega("conscrits_mega_{}.csv".format(MEGA_YEAR), qs) @buro_required def export_mega(request): event = Event.objects.get(title=MEGA_EVENT_NAME) qs = EventRegistration.objects.filter(event=event).order_by("user__username") return csv_export_mega("all_mega_{}.csv".format(MEGA_YEAR), qs) # ------------------------------ # Fin des exports Mega hardcodés # ------------------------------ @buro_required def utile_cof(request): return render(request, "gestioncof/utile_cof.html", {}) @buro_required def utile_bda(request): tirages = Tirage.objects.all() return render(request, "utile_bda.html", {"tirages": tirages}) @buro_required def liste_bdadiff(request): titre = "BdA diffusion" personnes = CofProfile.objects.filter(mailing_bda=True, is_cof=True).all() return render(request, "liste_mails.html", {"titre": titre, "personnes": personnes}) @buro_required def liste_bdarevente(request): titre = "BdA revente" personnes = CofProfile.objects.filter(mailing_bda_revente=True, is_cof=True).all() return render(request, "liste_mails.html", {"titre": titre, "personnes": personnes}) @buro_required def liste_diffcof(request): titre = "Diffusion COF" personnes = CofProfile.objects.filter(mailing_cof=True, is_cof=True).all() return render(request, "liste_mails.html", {"titre": titre, "personnes": personnes}) @cof_required def calendar(request): try: instance = CalendarSubscription.objects.get(user=request.user) except CalendarSubscription.DoesNotExist: instance = None if request.method == "POST": form = CalendarForm(request.POST, instance=instance) if form.is_valid(): subscription = form.save(commit=False) if instance is None: subscription.user = request.user subscription.token = uuid.uuid4() subscription.save() form.save_m2m() messages.success(request, "Calendrier mis à jour avec succès.") return render( request, "gestioncof/calendar_subscription.html", {"form": form, "token": str(subscription.token)}, ) else: messages.error(request, "Formulaire incorrect.") return render( request, "gestioncof/calendar_subscription.html", {"form": form} ) else: return render( request, "gestioncof/calendar_subscription.html", { "form": CalendarForm(instance=instance), "token": instance.token if instance else None, }, ) def calendar_ics(request, token): subscription = get_object_or_404(CalendarSubscription, token=token) shows = subscription.other_shows.all() if subscription.subscribe_to_my_shows: shows |= Spectacle.objects.filter( attribues__participant__user=subscription.user, tirage__active=True ) shows = shows.distinct() vcal = Calendar() site = Site.objects.get_current() for show in shows: vevent = Vevent() vevent.add("dtstart", show.date) vevent.add("dtend", show.date + timedelta(seconds=7200)) vevent.add("summary", show.title) vevent.add("location", show.location.name) vevent.add( "uid", "show-{:d}-{:d}@{:s}".format(show.pk, show.tirage_id, site.domain) ) vcal.add_component(vevent) if subscription.subscribe_to_events: for event in Event.objects.filter(old=False).all(): vevent = Vevent() vevent.add("dtstart", event.start_date) vevent.add("dtend", event.end_date) vevent.add("summary", event.title) vevent.add("location", event.location) vevent.add("description", event.description) vevent.add("uid", "event-{:d}@{:s}".format(event.pk, site.domain)) vcal.add_component(vevent) response = HttpResponse(content=vcal.to_ical()) response["Content-Type"] = "text/calendar" return response class ConfigUpdate(FormView): form_class = GestioncofConfigForm template_name = "gestioncof/banner_update.html" success_url = reverse_lazy("home") def dispatch(self, request, *args, **kwargs): if request.user is None or not request.user.is_superuser: return redirect_to_login(request.get_full_path()) return super().dispatch(request, *args, **kwargs) def form_valid(self, form): form.save() return super().form_valid(form) ## # Autocomplete views # # https://django-autocomplete-light.readthedocs.io/en/master/tutorial.html#create-an-autocomplete-view ## class UserAutocompleteView(BuroRequiredMixin, Select2QuerySetView): model = User search_fields = ("username", "first_name", "last_name") class RegistrationAutocompleteView(BuroRequiredMixin, AutocompleteView): template_name = "gestioncof/search_results.html" search_composer = cof_autocomplete