1271 lines
48 KiB
Python
1271 lines
48 KiB
Python
from django.shortcuts import render, get_object_or_404
|
|
from django.views.generic import ListView, DetailView, FormView
|
|
from django.views.generic.edit import CreateView, UpdateView, DeleteView
|
|
from django.contrib.auth.mixins import LoginRequiredMixin, UserPassesTestMixin, PermissionRequiredMixin
|
|
from django.http import QueryDict
|
|
from django import forms
|
|
from django.contrib.postgres.search import SearchQuery, SearchHeadline
|
|
from django.core.exceptions import PermissionDenied
|
|
|
|
from django.http import HttpResponseRedirect, HttpResponse, HttpResponseNotFound
|
|
from django.urls import reverse
|
|
import urllib
|
|
from collections import Counter
|
|
|
|
from .forms import EventSubmissionForm, EventForm, BatchImportationForm, FixDuplicates, SelectEventInList, MergeDuplicates, RecurrentImportForm, CategorisationRuleImportForm, ModerationQuestionForm, ModerationAnswerForm, ModerateForm, CategorisationForm, EventAddPlaceForm, PlaceForm
|
|
|
|
from .models import Event, Category, StaticContent, ContactMessage, BatchImportation, DuplicatedEvents, RecurrentImport, CategorisationRule, remove_accents, ModerationQuestion, ModerationAnswer, Place
|
|
from django.utils import timezone
|
|
from enum import StrEnum
|
|
from datetime import date, timedelta
|
|
from django.db.models import Q, F
|
|
|
|
from django.urls import reverse_lazy
|
|
from django.utils.translation import gettext_lazy as _
|
|
from django.utils.translation import activate, get_language_info
|
|
import django_filters
|
|
from django.core.paginator import Paginator, EmptyPage, PageNotAnInteger
|
|
from django.contrib.auth.decorators import login_required, permission_required
|
|
|
|
from django.contrib import messages
|
|
from django.contrib.messages.views import SuccessMessageMixin
|
|
|
|
from .calendar import CalendarMonth, CalendarWeek, CalendarDay
|
|
|
|
from .import_tasks.importer import URL2Events
|
|
from .import_tasks.extractor import Extractor
|
|
from .import_tasks.downloader import ChromiumHeadlessDownloader
|
|
|
|
from .celery import app as celery_app, import_events_from_json, run_recurrent_import, run_all_recurrent_imports
|
|
|
|
import unicodedata
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
#
|
|
#
|
|
# Useful for translation
|
|
to_be_translated = [
|
|
_("Recurrent import name"),
|
|
_("Add another"),
|
|
_("Browse..."),
|
|
_("Naviguer..."),
|
|
_("No file selected.")
|
|
]
|
|
|
|
|
|
|
|
|
|
def get_event_qs(request):
|
|
if request.user.is_authenticated:
|
|
return Event.objects.filter()
|
|
else:
|
|
return Event.objects.filter(status=Event.STATUS.PUBLISHED)
|
|
|
|
|
|
def page_not_found(request, exception=None):
|
|
return render(request, 'page-erreur.html', status=404, context={"error": 404})
|
|
|
|
|
|
def internal_server_error(request):
|
|
return render(request, 'page-erreur.html', status=500, context={"error": 500})
|
|
|
|
|
|
class CategoryCheckboxSelectMultiple(forms.CheckboxSelectMultiple):
|
|
template_name = 'agenda_culturel/forms/category-checkbox.html'
|
|
option_template_name = 'agenda_culturel/forms/checkbox-option.html'
|
|
|
|
class TagCheckboxSelectMultiple(forms.CheckboxSelectMultiple):
|
|
template_name = 'agenda_culturel/forms/tag-checkbox.html'
|
|
option_template_name = 'agenda_culturel/forms/checkbox-option.html'
|
|
|
|
|
|
class EventFilter(django_filters.FilterSet):
|
|
RECURRENT_CHOICES = [("remove_recurrent", "Masquer les événements récurrents"), ("only_recurrent", "Montrer uniquement les événements récurrents")]
|
|
|
|
exclude_tags = django_filters.MultipleChoiceFilter(label="Exclure les étiquettes",
|
|
choices=[(t, t) for t in Event.get_all_tags()],
|
|
lookup_expr='icontains',
|
|
field_name="tags",
|
|
exclude=True,
|
|
widget=TagCheckboxSelectMultiple)
|
|
|
|
tags = django_filters.MultipleChoiceFilter(label="Filtrer par étiquettes",
|
|
choices=[(t, t) for t in Event.get_all_tags()],
|
|
lookup_expr='icontains',
|
|
field_name="tags",
|
|
widget=TagCheckboxSelectMultiple)
|
|
|
|
recurrences = django_filters.ChoiceFilter(label="Filtrer par récurrence",
|
|
choices=RECURRENT_CHOICES,
|
|
method="filter_recurrences")
|
|
|
|
category = django_filters.ModelMultipleChoiceFilter(label="Filtrer par catégories",
|
|
field_name="category__id",
|
|
to_field_name='id',
|
|
queryset=Category.objects.all(),
|
|
widget=CategoryCheckboxSelectMultiple)
|
|
|
|
city = django_filters.MultipleChoiceFilter(label="Filtrer par ville",
|
|
field_name='exact_location__city',
|
|
choices=[(c, c) for c in Place.get_all_cities()],
|
|
widget=forms.CheckboxSelectMultiple)
|
|
|
|
status = django_filters.MultipleChoiceFilter(label="Filtrer par status",
|
|
choices=Event.STATUS.choices,
|
|
field_name="status",
|
|
widget=forms.CheckboxSelectMultiple)
|
|
|
|
|
|
class Meta:
|
|
model = Event
|
|
fields = ["category", "city", "tags", "exclude_tags", "status", "recurrences"]
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
if not kwargs["request"].user.is_authenticated:
|
|
self.form.fields.pop("status")
|
|
|
|
def filter_recurrences(self, queryset, name, value):
|
|
# construct the full lookup expression
|
|
lookup = '__'.join([name, 'isnull'])
|
|
return queryset.filter(**{lookup: value == "remove_recurrent"})
|
|
|
|
def get_url(self):
|
|
if isinstance(self.form.data, QueryDict):
|
|
return self.form.data.urlencode()
|
|
else:
|
|
print(self.form.data)
|
|
return ""
|
|
|
|
def get_url_without_filters(self):
|
|
return self.request.get_full_path().split("?")[0]
|
|
|
|
def get_categories(self):
|
|
return self.form.cleaned_data["category"]
|
|
|
|
def get_tags(self):
|
|
return self.form.cleaned_data["tags"]
|
|
|
|
def get_exclude_tags(self):
|
|
return self.form.cleaned_data["exclude_tags"]
|
|
|
|
def get_status(self):
|
|
return self.form.cleaned_data["status"]
|
|
|
|
def get_cities(self):
|
|
return self.form.cleaned_data["city"]
|
|
|
|
def get_status_names(self):
|
|
if "status" in self.form.cleaned_data:
|
|
return [dict(Event.STATUS.choices)[s] for s in self.form.cleaned_data["status"]]
|
|
else:
|
|
return []
|
|
def get_recurrence_filtering(self):
|
|
if "recurrences" in self.form.cleaned_data:
|
|
d = dict(self.RECURRENT_CHOICES)
|
|
v = self.form.cleaned_data["recurrences"]
|
|
if v in d:
|
|
return d[v]
|
|
else:
|
|
return ""
|
|
else:
|
|
return ""
|
|
|
|
def is_active(self, only_categories=False):
|
|
if only_categories:
|
|
return len(self.form.cleaned_data["category"]) != 0
|
|
else:
|
|
if "status" in self.form.cleaned_data and len(self.form.cleaned_data["status"]) != 0:
|
|
return True
|
|
return len(self.form.cleaned_data["category"]) != 0 or len(self.form.cleaned_data["tags"]) != 0 or len(self.form.cleaned_data["exclude_tags"]) != 0 or len(self.form.cleaned_data["recurrences"]) != 0 or len(self.form.cleaned_data["city"]) != 0
|
|
|
|
def is_selected(self, cat):
|
|
return cat in self.form.cleaned_data["category"]
|
|
|
|
|
|
def mentions_legales(request):
|
|
context = { "title": "Mentions légales", "static_content": "mentions_legales", "url_path": reverse_lazy("mentions_legales") }
|
|
return render(request, 'agenda_culturel/page-single.html', context)
|
|
|
|
def about(request):
|
|
rimports = RecurrentImport.objects.order_by("name__unaccent").all()
|
|
context = { "title": "À propos", "static_content": "about", "url_path": reverse_lazy("about"), "rimports": rimports }
|
|
return render(request, 'agenda_culturel/page-rimports-list.html', context)
|
|
|
|
|
|
def home(request):
|
|
return week_view(request, home=True)
|
|
|
|
def month_view(request, year = None, month = None):
|
|
now = date.today()
|
|
if year is None:
|
|
year = now.year
|
|
if month is None:
|
|
month = now.month
|
|
|
|
filter = EventFilter(request.GET, queryset=get_event_qs(request), request=request)
|
|
cmonth = CalendarMonth(year, month, filter)
|
|
|
|
|
|
context = {"year": year, "month": cmonth.get_month_name(), "calendar": cmonth, "filter": filter }
|
|
return render(request, 'agenda_culturel/page-month.html', context)
|
|
|
|
|
|
def week_view(request, year=None, week=None, home=False):
|
|
now = date.today()
|
|
if year is None:
|
|
year = now.year
|
|
if week is None:
|
|
week = now.isocalendar()[1]
|
|
|
|
filter = EventFilter(request.GET, queryset=get_event_qs(request), request=request)
|
|
cweek = CalendarWeek(year, week, filter)
|
|
|
|
context = {"year": year, "week": week, "calendar": cweek, "filter": filter }
|
|
if home:
|
|
context["home"] = 1
|
|
return render(request, 'agenda_culturel/page-week.html', context)
|
|
|
|
|
|
def day_view(request, year = None, month = None, day = None):
|
|
now = date.today()
|
|
if year is None:
|
|
year = now.year
|
|
if month is None:
|
|
month = now.month
|
|
if day is None:
|
|
day = now.day
|
|
|
|
day = date(year, month, day)
|
|
|
|
filter = EventFilter(request.GET, get_event_qs(request), request=request)
|
|
cday = CalendarDay(day, filter)
|
|
|
|
categories = Counter([e.category if e.category is not None else Category.get_default_category() for e in cday.get_events()])
|
|
categories = [(k, v) for k, v in categories.items()]
|
|
categories.sort(key=lambda k: -k[1])
|
|
|
|
context = {"day": day, "events": cday.get_events(), "filter": filter, "categories": categories}
|
|
return render(request, 'agenda_culturel/page-day.html', context)
|
|
|
|
|
|
def view_tag(request, t):
|
|
events = Event.objects.filter(tags__contains=[t]).order_by("start_day", "start_time")
|
|
context = {"tag": t, "events": events}
|
|
return render(request, 'agenda_culturel/tag.html', context)
|
|
|
|
|
|
def tag_list(request):
|
|
|
|
tags = Event.get_all_tags()
|
|
context = {"tags": sorted(tags, key=lambda x: remove_accents(x).lower())}
|
|
return render(request, 'agenda_culturel/tags.html', context)
|
|
|
|
|
|
class StaticContentCreateView(LoginRequiredMixin, CreateView):
|
|
model = StaticContent
|
|
fields = ['text']
|
|
permission_required = ("agenda_culturel.add_staticcontent")
|
|
|
|
def form_valid(self, form):
|
|
form.instance.name = self.request.GET["name"]
|
|
form.instance.url_path = self.request.GET["url_path"]
|
|
return super().form_valid(form)
|
|
|
|
|
|
class StaticContentUpdateView(SuccessMessageMixin, PermissionRequiredMixin, LoginRequiredMixin, UpdateView):
|
|
model = StaticContent
|
|
permission_required = ("agenda_culturel.change_staticcontent")
|
|
fields = ['text']
|
|
success_message = _('The static content has been successfully updated.')
|
|
|
|
|
|
class EventUpdateView(SuccessMessageMixin, PermissionRequiredMixin, LoginRequiredMixin, UpdateView):
|
|
model = Event
|
|
permission_required = ("agenda_culturel.change_event")
|
|
form_class = EventForm
|
|
success_message = _('The event has been successfully modified.')
|
|
|
|
def get_form_kwargs(self):
|
|
kwargs = super().get_form_kwargs()
|
|
kwargs['is_authenticated'] = self.request.user.is_authenticated
|
|
return kwargs
|
|
|
|
|
|
class EventDeleteView(SuccessMessageMixin, PermissionRequiredMixin, LoginRequiredMixin, DeleteView):
|
|
model = Event
|
|
permission_required = ("agenda_culturel.delete_event")
|
|
success_url = reverse_lazy('moderation')
|
|
success_message = _('The event has been successfully deleted.')
|
|
|
|
|
|
class EventDetailView(UserPassesTestMixin, DetailView):
|
|
model = Event
|
|
template_name = "agenda_culturel/page-event.html"
|
|
|
|
def test_func(self):
|
|
return self.request.user.is_authenticated or self.get_object().status == Event.STATUS.PUBLISHED
|
|
|
|
def get_object(self):
|
|
o = super().get_object()
|
|
y = self.kwargs["year"]
|
|
m = self.kwargs["month"]
|
|
d = self.kwargs["day"]
|
|
obj = o.get_recurrence_at_date(y, m, d)
|
|
obj.set_current_date(date(y, m, d))
|
|
return obj
|
|
|
|
|
|
class EventModerateView(SuccessMessageMixin, PermissionRequiredMixin, LoginRequiredMixin, UpdateView):
|
|
model = Event
|
|
permission_required = ("agenda_culturel.change_event", "agenda_culturel.use_moderation_question")
|
|
success_message = _('The event has been successfully modified.')
|
|
form_class = ModerateForm
|
|
template_name = 'agenda_culturel/event_moderation_form.html'
|
|
|
|
def form_valid(self, form):
|
|
|
|
mas = ModerationAnswer.objects.all()
|
|
logger.warning("ON valide la forme")
|
|
|
|
for f in form.cleaned_data:
|
|
ModerationAnswer.objects.get(pk=form.cleaned_data[f]).apply_answer(form.instance)
|
|
|
|
form.instance.moderated_date = timezone.now()
|
|
|
|
return super().form_valid(form)
|
|
|
|
|
|
|
|
@login_required(login_url="/accounts/login/")
|
|
@permission_required('agenda_culturel.change_event')
|
|
def change_status_event(request, pk, status):
|
|
event = get_object_or_404(Event, pk=pk)
|
|
|
|
if request.method == 'POST':
|
|
event.status = Event.STATUS(status)
|
|
event.save(update_fields=["status"])
|
|
messages.success(request, _("The status has been successfully modified."))
|
|
|
|
if request.user.is_authenticated:
|
|
return HttpResponseRedirect(event.get_absolute_url())
|
|
else:
|
|
return HttpResponseRedirect(reverse_lazy("home"))
|
|
else:
|
|
cancel_url = request.META.get('HTTP_REFERER', '')
|
|
if cancel_url == "":
|
|
cancel_url = reverse_lazy("home")
|
|
return render(request, 'agenda_culturel/event_confirm_change_status.html', {"status": status, "event": event, "cancel_url": cancel_url})
|
|
|
|
def import_from_url(request):
|
|
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
if request.method == 'POST' and "title" in request.POST:
|
|
form = EventForm(request.POST, is_authenticated=request.user.is_authenticated)
|
|
if form.is_valid():
|
|
new_event = form.save()
|
|
if request.user.is_authenticated:
|
|
messages.success(request, _("The event is saved."))
|
|
return HttpResponseRedirect(new_event.get_absolute_url())
|
|
else:
|
|
messages.success(request, _("The event has been submitted and will be published as soon as it has been validated by the moderation team."))
|
|
return HttpResponseRedirect(reverse("home"))
|
|
else:
|
|
return render(request, 'agenda_culturel/event_form.html', context={'form': form })
|
|
|
|
else:
|
|
form = EventSubmissionForm()
|
|
|
|
initial = {"start_day": date.today() + timedelta(days=1),
|
|
"start_time": "20:00",
|
|
"end_time": "22:00"}
|
|
|
|
form_event = EventForm(initial=initial)
|
|
|
|
if request.method == 'POST':
|
|
|
|
form = EventSubmissionForm(request.POST)
|
|
|
|
if form.is_valid():
|
|
cd = form.cleaned_data
|
|
url = cd.get('url')
|
|
|
|
url = Extractor.clean_url(url)
|
|
|
|
existing = Event.objects.filter(uuids__contains=[url])
|
|
|
|
if len(existing) == 0:
|
|
event = None
|
|
|
|
u2e = URL2Events(ChromiumHeadlessDownloader(), single_event=True)
|
|
events_structure = u2e.process(url, published=request.user.is_authenticated)
|
|
if events_structure is not None and "events" in events_structure and len(events_structure["events"]) > 0:
|
|
event = Event.from_structure(events_structure["events"][0], events_structure["header"]["url"])
|
|
# TODO: use celery to import the other events
|
|
|
|
if event != None:
|
|
form = EventForm(instance=event, is_authenticated=request.user.is_authenticated)
|
|
messages.success(request, _("The event has been successfully extracted, and you can now submit it after modifying it if necessary."))
|
|
return render(request, 'agenda_culturel/event_form.html', context={'form': form })
|
|
else:
|
|
form = EventForm(initial={'reference_urls': [url]}, is_authenticated=request.user.is_authenticated)
|
|
messages.error(request, _("Unable to extract an event from the proposed URL. Please use the form below to submit the event."))
|
|
return render(request, 'agenda_culturel/import.html', context={'form': form, 'form_event': form_event})
|
|
else:
|
|
published = [e for e in existing if e.status == Event.STATUS.PUBLISHED]
|
|
drafts = [e for e in existing if e.status == Event.STATUS.DRAFT]
|
|
trash = [e for e in existing if e.status == Event.STATUS.TRASH]
|
|
|
|
if request.user.is_authenticated or len(published) > 1:
|
|
event = published[0] if len(published) > 1 else existing[0]
|
|
messages.info(request, _("This URL has already been submitted, and you can find the event below."))
|
|
return HttpResponseRedirect(event.get_absolute_url())
|
|
else:
|
|
if len(drafts) > 0:
|
|
messages.info(request, _("This URL has already been submitted and is awaiting moderation."))
|
|
elif len(trash) > 0:
|
|
messages.info(request, _("This URL has already been submitted, but has not been selected for publication by the moderation team."))
|
|
|
|
|
|
return render(request, 'agenda_culturel/import.html', context={'form': form, 'form_event': form_event})
|
|
|
|
|
|
class EventFilterAdmin(django_filters.FilterSet):
|
|
status = django_filters.MultipleChoiceFilter(choices=Event.STATUS.choices, widget=forms.CheckboxSelectMultiple)
|
|
|
|
class Meta:
|
|
model = Event
|
|
fields = ['status']
|
|
|
|
|
|
|
|
|
|
class ContactMessageCreateView(SuccessMessageMixin, CreateView):
|
|
model = ContactMessage
|
|
template_name = "agenda_culturel/contactmessage_create_form.html"
|
|
fields = ['subject', 'name', 'email', 'message']
|
|
|
|
success_url = reverse_lazy('home')
|
|
success_message = _('Your message has been sent successfully.')
|
|
|
|
def get_form(self, form_class=None):
|
|
if form_class is None:
|
|
form_class = self.get_form_class()
|
|
return form_class(**self.get_form_kwargs())
|
|
|
|
|
|
|
|
class ContactMessageUpdateView(SuccessMessageMixin, PermissionRequiredMixin, LoginRequiredMixin, UpdateView):
|
|
model = ContactMessage
|
|
permission_required = ("agenda_culturel.change_contactmessage")
|
|
template_name = "agenda_culturel/contactmessage_moderation_form.html"
|
|
fields = ('closed', 'comments')
|
|
|
|
success_message = _('The contact message properties has been successfully modified.')
|
|
|
|
success_url = reverse_lazy('contactmessages')
|
|
|
|
|
|
def get_form_kwargs(self):
|
|
"""Return the keyword arguments for instantiating the form."""
|
|
kwargs = super().get_form_kwargs()
|
|
if hasattr(self, 'object'):
|
|
kwargs.update({'instance': self.object})
|
|
return kwargs
|
|
|
|
|
|
|
|
class ContactMessagesFilterAdmin(django_filters.FilterSet):
|
|
closed = django_filters.MultipleChoiceFilter(label="Status", choices=((True, _("Closed")), (False, _("Open"))), widget=forms.CheckboxSelectMultiple)
|
|
class Meta:
|
|
model = ContactMessage
|
|
fields = ['closed']
|
|
|
|
|
|
|
|
@login_required(login_url="/accounts/login/")
|
|
@permission_required('agenda_culturel.view_event')
|
|
def moderation(request):
|
|
filter = EventFilterAdmin(request.GET, queryset=Event.objects.all().order_by("-created_date"))
|
|
paginator = Paginator(filter.qs, 10)
|
|
page = request.GET.get('page')
|
|
|
|
try:
|
|
response = paginator.page(page)
|
|
except PageNotAnInteger:
|
|
response = paginator.page(1)
|
|
except EmptyPage:
|
|
response = paginator.page(paginator.num_pages)
|
|
|
|
return render(request, 'agenda_culturel/moderation.html', {'filter': filter, 'paginator_filter': response} )
|
|
|
|
@login_required(login_url="/accounts/login/")
|
|
@permission_required('agenda_culturel.view_contactmessage')
|
|
def contactmessages(request):
|
|
filter = ContactMessagesFilterAdmin(request.GET, queryset=ContactMessage.objects.all().order_by("-date"))
|
|
paginator = Paginator(filter.qs, 10)
|
|
page = request.GET.get('page')
|
|
|
|
try:
|
|
response = paginator.page(page)
|
|
except PageNotAnInteger:
|
|
response = paginator.page(1)
|
|
except EmptyPage:
|
|
response = paginator.page(paginator.num_pages)
|
|
|
|
return render(request, 'agenda_culturel/contactmessages.html', {'filter': filter, 'paginator_filter': response} )
|
|
|
|
|
|
|
|
class SimpleSearchEventFilter(django_filters.FilterSet):
|
|
q = django_filters.CharFilter(method='custom_filter', label=_("Search"))
|
|
|
|
status = django_filters.MultipleChoiceFilter(label="Filtrer par status",
|
|
choices=Event.STATUS.choices,
|
|
field_name="status",
|
|
widget=forms.CheckboxSelectMultiple)
|
|
|
|
def custom_filter(self, queryset, name, value):
|
|
search_query = SearchQuery(value, config='french')
|
|
qs = queryset.filter(
|
|
Q(title__icontains=value) | Q(location__icontains=value) | Q(description__icontains=value))
|
|
for f in ["title", "location", "description"]:
|
|
params = { f + "_hl": SearchHeadline(f,
|
|
search_query,
|
|
start_sel="<span class=\"highlight\">",
|
|
stop_sel="</span>",
|
|
config='french')}
|
|
qs = qs.annotate(**params)
|
|
return qs
|
|
|
|
class Meta:
|
|
model = Event
|
|
fields = ['q']
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
if not kwargs["request"].user.is_authenticated:
|
|
self.form.fields.pop("status")
|
|
|
|
|
|
class SearchEventFilter(django_filters.FilterSet):
|
|
tags = django_filters.CharFilter(lookup_expr='icontains')
|
|
title = django_filters.CharFilter(method="hl_filter_contains")
|
|
location = django_filters.CharFilter(method="hl_filter_contains")
|
|
description = django_filters.CharFilter(method="hl_filter_contains")
|
|
start_day = django_filters.DateFromToRangeFilter(widget=django_filters.widgets.RangeWidget(attrs={'type': 'date'}))
|
|
status = django_filters.MultipleChoiceFilter(label="Filtrer par status",
|
|
choices=Event.STATUS.choices,
|
|
field_name="status",
|
|
widget=forms.CheckboxSelectMultiple)
|
|
|
|
o = django_filters.OrderingFilter(
|
|
# tuple-mapping retains order
|
|
fields=(
|
|
('title', 'title'),
|
|
('description', 'description'),
|
|
('start_day', 'start_day'),
|
|
),
|
|
)
|
|
|
|
def hl_filter_contains(self, queryset, name, value):
|
|
|
|
# first check if it contains
|
|
filter_contains = { name + "__contains": value }
|
|
queryset = queryset.filter(**filter_contains)
|
|
|
|
# then hightlight the result
|
|
search_query = SearchQuery(value, config='french')
|
|
params = { name + "_hl": SearchHeadline(name,
|
|
search_query,
|
|
start_sel="<span class=\"highlight\">",
|
|
stop_sel="</span>",
|
|
config='french')}
|
|
return queryset.annotate(**params)
|
|
|
|
|
|
class Meta:
|
|
model = Event
|
|
fields = ['title', 'location', 'description', 'category', 'tags', 'start_day']
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
if not kwargs["request"].user.is_authenticated:
|
|
self.form.fields.pop("status")
|
|
|
|
|
|
def event_search(request, full=False):
|
|
|
|
if full:
|
|
filter = SearchEventFilter(request.GET, queryset=get_event_qs(request).order_by("-start_day"), request=request)
|
|
else:
|
|
filter = SimpleSearchEventFilter(request.GET, queryset=get_event_qs(request).order_by("-start_day"), request=request)
|
|
|
|
|
|
paginator = Paginator(filter.qs, 10)
|
|
page = request.GET.get('page')
|
|
|
|
try:
|
|
response = paginator.page(page)
|
|
except PageNotAnInteger:
|
|
response = paginator.page(1)
|
|
except EmptyPage:
|
|
response = paginator.page(paginator.num_pages)
|
|
|
|
return render(request, 'agenda_culturel/search.html', {'filter': filter,
|
|
'has_results': len(request.GET) != 0 or (len(request.GET) > 1 and "page" in request.GET),
|
|
'paginator_filter': response,
|
|
'full': full})
|
|
|
|
def event_search_full(request):
|
|
return event_search(request, True)
|
|
|
|
|
|
#########################
|
|
## batch importations
|
|
#########################
|
|
|
|
@login_required(login_url="/accounts/login/")
|
|
@permission_required('agenda_culturel.view_batchimportation')
|
|
def imports(request):
|
|
paginator = Paginator(BatchImportation.objects.all().order_by("-created_date"), 10)
|
|
page = request.GET.get('page')
|
|
|
|
try:
|
|
response = paginator.page(page)
|
|
except PageNotAnInteger:
|
|
response = paginator.page(1)
|
|
except EmptyPage:
|
|
response = paginator.page(paginator.num_pages)
|
|
|
|
return render(request, 'agenda_culturel/imports.html', {'paginator_filter': response} )
|
|
|
|
|
|
@login_required(login_url="/accounts/login/")
|
|
@permission_required(['agenda_culturel.add_batchimportation', 'agenda_culturel.run_batchimportation'])
|
|
def add_import(request):
|
|
form = BatchImportationForm()
|
|
|
|
if request.method == 'POST':
|
|
form = BatchImportationForm(request.POST)
|
|
|
|
if form.is_valid():
|
|
|
|
result = import_events_from_json.delay(form.data["json"])
|
|
|
|
messages.success(request, _("The import has been run successfully."))
|
|
return HttpResponseRedirect(reverse_lazy("imports"))
|
|
|
|
return render(request, 'agenda_culturel/batchimportation_form.html', {"form": form})
|
|
|
|
|
|
@login_required(login_url="/accounts/login/")
|
|
@permission_required(['agenda_culturel.view_batchimportation', 'agenda_culturel.run_batchimportation'])
|
|
def cancel_import(request, pk):
|
|
import_process = get_object_or_404(BatchImportation, pk=pk)
|
|
|
|
if request.method == 'POST':
|
|
celery_app.control.revoke(import_process.celery_id)
|
|
|
|
import_process.status = BatchImportation.STATUS.CANCELED
|
|
import_process.save(update_fields=["status"])
|
|
|
|
messages.success(request, _("The import has been canceled."))
|
|
return HttpResponseRedirect(reverse_lazy("imports"))
|
|
else:
|
|
cancel_url = reverse_lazy("imports")
|
|
return render(request, 'agenda_culturel/cancel_import_confirm.html', {"object": import_process, "cancel_url": cancel_url})
|
|
|
|
#########################
|
|
## recurrent importations
|
|
#########################
|
|
|
|
@login_required(login_url="/accounts/login/")
|
|
@permission_required('agenda_culturel.view_recurrentimport')
|
|
def recurrent_imports(request):
|
|
paginator = Paginator(RecurrentImport.objects.all().order_by("-pk"), 10)
|
|
page = request.GET.get('page')
|
|
|
|
try:
|
|
response = paginator.page(page)
|
|
except PageNotAnInteger:
|
|
response = paginator.page(1)
|
|
except EmptyPage:
|
|
response = paginator.page(paginator.num_pages)
|
|
|
|
return render(request, 'agenda_culturel/rimports.html', {'paginator_filter': response} )
|
|
|
|
|
|
class RecurrentImportCreateView(LoginRequiredMixin, PermissionRequiredMixin, CreateView):
|
|
model = RecurrentImport
|
|
permission_required = ("agenda_culturel.add_recurrentimport")
|
|
success_url = reverse_lazy('recurrent_imports')
|
|
form_class = RecurrentImportForm
|
|
|
|
|
|
class RecurrentImportUpdateView(SuccessMessageMixin, PermissionRequiredMixin, LoginRequiredMixin, UpdateView):
|
|
model = RecurrentImport
|
|
permission_required = ("agenda_culturel.change_recurrentimport")
|
|
form_class = RecurrentImportForm
|
|
success_message = _('The recurrent import has been successfully modified.')
|
|
|
|
|
|
class RecurrentImportDeleteView(SuccessMessageMixin, PermissionRequiredMixin, LoginRequiredMixin, DeleteView):
|
|
model = RecurrentImport
|
|
permission_required = ("agenda_culturel.delete_recurrentimport")
|
|
success_url = reverse_lazy('recurrent_imports')
|
|
success_message = _('The recurrent import has been successfully deleted.')
|
|
|
|
|
|
@login_required(login_url="/accounts/login/")
|
|
@permission_required(['agenda_culturel.view_recurrentimport', 'agenda_culturel.view_batchimportation'])
|
|
def view_rimport(request, pk):
|
|
obj = get_object_or_404(RecurrentImport, pk=pk)
|
|
paginator = Paginator(BatchImportation.objects.filter(recurrentImport=pk).order_by("-created_date"), 10)
|
|
page = request.GET.get('page')
|
|
|
|
try:
|
|
response = paginator.page(page)
|
|
except PageNotAnInteger:
|
|
response = paginator.page(1)
|
|
except EmptyPage:
|
|
response = paginator.page(paginator.num_pages)
|
|
|
|
return render(request, 'agenda_culturel/page-rimport.html', {'paginator_filter': response, 'object': obj} )
|
|
|
|
|
|
|
|
@login_required(login_url="/accounts/login/")
|
|
@permission_required(['agenda_culturel.view_recurrentimport', 'agenda_culturel.run_recurrentimport'])
|
|
def run_rimport(request, pk):
|
|
rimport = get_object_or_404(RecurrentImport, pk=pk)
|
|
|
|
if request.method == 'POST':
|
|
|
|
# run recurrent import
|
|
result = run_recurrent_import.delay(pk)
|
|
|
|
messages.success(request, _("The import has been launched."))
|
|
return HttpResponseRedirect(reverse_lazy("view_rimport", args=[pk]))
|
|
else:
|
|
return render(request, 'agenda_culturel/run_rimport_confirm.html', {"object": rimport })
|
|
|
|
|
|
@login_required(login_url="/accounts/login/")
|
|
@permission_required(['agenda_culturel.view_recurrentimport', 'agenda_culturel.run_recurrentimport'])
|
|
def run_all_rimports(request):
|
|
if request.method == 'POST':
|
|
|
|
# run recurrent import
|
|
result = run_all_recurrent_imports.delay()
|
|
|
|
messages.success(request, _("Imports has been launched."))
|
|
return HttpResponseRedirect(reverse_lazy("recurrent_imports"))
|
|
else:
|
|
return render(request, 'agenda_culturel/run_all_rimports_confirm.html')
|
|
|
|
|
|
#########################
|
|
## duplicated events
|
|
#########################
|
|
|
|
|
|
class DuplicatedEventsDetailView(LoginRequiredMixin, DetailView):
|
|
model = DuplicatedEvents
|
|
template_name = "agenda_culturel/fix_duplicate.html"
|
|
|
|
|
|
@login_required(login_url="/accounts/login/")
|
|
@permission_required(['agenda_culturel.change_event', 'agenda_culturel.change_duplicatedevents'])
|
|
def merge_duplicate(request, pk):
|
|
edup = get_object_or_404(DuplicatedEvents, pk=pk)
|
|
form = MergeDuplicates(duplicates=edup)
|
|
|
|
if request.method == 'POST':
|
|
form = MergeDuplicates(request.POST, duplicates=edup)
|
|
if form.is_valid():
|
|
events = edup.get_duplicated()
|
|
|
|
# build fields for the new event
|
|
new_event_data = {}
|
|
for f in edup.get_items_comparison():
|
|
if f["similar"]:
|
|
new_event_data[f["key"]] = getattr(events[0], f["key"])
|
|
else:
|
|
selected = form.get_selected_events_id(f["key"])
|
|
if selected is None:
|
|
new_event_data[f["key"]] = None
|
|
elif isinstance(selected, list):
|
|
values = [x for x in [getattr(events[s], f["key"]) for s in selected] if x is not None]
|
|
if len(values) == 0:
|
|
new_event_data[f["key"]] = None
|
|
else:
|
|
if isinstance(values[0], str):
|
|
new_event_data[f["key"]] = "\n".join(values)
|
|
else:
|
|
new_event_data[f["key"]] = sum(values, [])
|
|
else:
|
|
new_event_data[f["key"]] = getattr(events[selected], f["key"])
|
|
|
|
for specific_tag in ["uuids", "import_sources"]:
|
|
new_event_data[specific_tag] = sum([x for x in [getattr(e, specific_tag) for e in events] if x is not None], [])
|
|
|
|
# create a new event that merge the selected events
|
|
new_event = Event(**new_event_data)
|
|
new_event.set_skip_duplicate_check()
|
|
new_event.save()
|
|
|
|
# move the old ones in trash
|
|
for e in events:
|
|
e.status = Event.STATUS.TRASH
|
|
Event.objects.bulk_update(events, fields=["status"])
|
|
|
|
messages.info(request, _("The merge has been successfully completed."))
|
|
return HttpResponseRedirect(new_event.get_absolute_url())
|
|
|
|
return render(request, 'agenda_culturel/merge_duplicate.html', context={'form': form, 'object': edup})
|
|
|
|
|
|
@login_required(login_url="/accounts/login/")
|
|
@permission_required(['agenda_culturel.change_event', 'agenda_culturel.change_duplicatedevents'])
|
|
def fix_duplicate(request, pk):
|
|
|
|
edup = get_object_or_404(DuplicatedEvents, pk=pk)
|
|
form = FixDuplicates(nb_events=edup.nb_duplicated())
|
|
|
|
if request.method == 'POST':
|
|
form = FixDuplicates(request.POST, nb_events=edup.nb_duplicated())
|
|
|
|
|
|
if form.is_valid():
|
|
if form.is_action_no_duplicates():
|
|
events = edup.get_duplicated()
|
|
if len(events) == 0:
|
|
date = None
|
|
else:
|
|
s_events = [e for e in events if not e.has_recurrences()]
|
|
if len(s_events) != 0:
|
|
s_event = s_events[0]
|
|
else:
|
|
s_event = events[0]
|
|
date = s_event.start_day
|
|
|
|
messages.success(request, _("Events have been marked as unduplicated."))
|
|
edup.delete()
|
|
if date is None:
|
|
return HttpResponseRedirect(reverse_lazy("home"))
|
|
else:
|
|
return HttpResponseRedirect(reverse_lazy("day_view", args=[date.year, date.month, date.day]))
|
|
|
|
elif form.is_action_select():
|
|
selected = form.get_selected_event(edup)
|
|
not_selected = [e for e in edup.get_duplicated() if e != selected]
|
|
nb = len(not_selected)
|
|
for e in not_selected:
|
|
e.status = Event.STATUS.TRASH
|
|
Event.objects.bulk_update(not_selected, fields=["status"])
|
|
url = selected.get_absolute_url()
|
|
edup.delete()
|
|
if nb == 1:
|
|
messages.success(request, _("The selected event has been retained, while the other has been moved to the recycle bin."))
|
|
else:
|
|
messages.success(request, _("The selected event has been retained, while the others have been moved to the recycle bin."))
|
|
return HttpResponseRedirect(url)
|
|
elif form.is_action_remove():
|
|
event = form.get_selected_event(edup)
|
|
event.possibly_duplicated = None
|
|
event.save()
|
|
messages.success(request, _("The event has been withdrawn from the group and made independent."))
|
|
if edup.nb_duplicated() == 1:
|
|
return HttpResponseRedirect(event.get_absolute_url())
|
|
else:
|
|
form = FixDuplicates(nb_events=edup.nb_duplicated())
|
|
else:
|
|
return HttpResponseRedirect(reverse_lazy("merge_duplicate", args=[edup.pk]))
|
|
|
|
return render(request, 'agenda_culturel/fix_duplicate.html', context={'form': form, 'object': edup})
|
|
|
|
|
|
|
|
class DuplicatedEventsUpdateView(LoginRequiredMixin, UpdateView):
|
|
model = DuplicatedEvents
|
|
fields = ()
|
|
template_name = "agenda_culturel/fix_duplicate.html"
|
|
|
|
|
|
@login_required(login_url="/accounts/login/")
|
|
@permission_required('agenda_culturel.view_duplicatedevents')
|
|
def duplicates(request):
|
|
nb_removed = DuplicatedEvents.remove_singletons()
|
|
nb_similar = DuplicatedEvents.remove_similar_entries()
|
|
if nb_removed > 0 or nb_similar > 0:
|
|
messages.success(request, _("Cleaning up duplicates: {} item(s) removed.").format(nb_removed + nb_similar))
|
|
|
|
paginator = Paginator(DuplicatedEvents.objects.all(), 10)
|
|
page = request.GET.get('page')
|
|
|
|
try:
|
|
response = paginator.page(page)
|
|
except PageNotAnInteger:
|
|
response = paginator.page(1)
|
|
except EmptyPage:
|
|
response = paginator.page(paginator.num_pages)
|
|
|
|
return render(request, 'agenda_culturel/duplicates.html', {'filter': filter, 'paginator_filter': response} )
|
|
|
|
def set_duplicate(request, year, month, day, pk):
|
|
event = get_object_or_404(Event, pk=pk)
|
|
cday = CalendarDay(date(year, month, day))
|
|
others = [e for e in cday.get_events() if e != event and (event.possibly_duplicated is None or event.possibly_duplicated != e.possibly_duplicated)]
|
|
|
|
form = SelectEventInList(events=others)
|
|
|
|
if request.method == 'POST':
|
|
form = SelectEventInList(request.POST, events=others)
|
|
if form.is_valid():
|
|
selected = [o for o in others if o.pk == int(form.cleaned_data["event"])]
|
|
event.set_possibly_duplicated(selected)
|
|
event.save()
|
|
if request.user.is_authenticated:
|
|
messages.success(request, _("The event was successfully duplicated."))
|
|
return HttpResponseRedirect(reverse_lazy("view_duplicate", args=[event.possibly_duplicated.pk]))
|
|
else:
|
|
messages.info(request, _("The event has been successfully flagged as a duplicate. The moderation team will deal with your suggestion shortly."))
|
|
return HttpResponseRedirect(event.get_absolute_url())
|
|
|
|
|
|
return render(request, 'agenda_culturel/set_duplicate.html', context={'form': form, 'event': event})
|
|
|
|
|
|
|
|
#########################
|
|
## categorisation rules
|
|
#########################
|
|
|
|
@login_required(login_url="/accounts/login/")
|
|
@permission_required('agenda_culturel.view_categorisationrule')
|
|
def categorisation_rules(request):
|
|
paginator = Paginator(CategorisationRule.objects.all().order_by("weight", "pk"), 10)
|
|
page = request.GET.get('page')
|
|
|
|
try:
|
|
response = paginator.page(page)
|
|
except PageNotAnInteger:
|
|
response = paginator.page(1)
|
|
except EmptyPage:
|
|
response = paginator.page(paginator.num_pages)
|
|
|
|
return render(request, 'agenda_culturel/categorisation_rules.html', {'paginator_filter': response} )
|
|
|
|
class CategorisationRuleCreateView(LoginRequiredMixin, PermissionRequiredMixin, CreateView):
|
|
model = CategorisationRule
|
|
permission_required = ("agenda_culturel.add_categorisationrule")
|
|
success_url = reverse_lazy('categorisation_rules')
|
|
form_class = CategorisationRuleImportForm
|
|
|
|
|
|
class CategorisationRuleUpdateView(SuccessMessageMixin, PermissionRequiredMixin, LoginRequiredMixin, UpdateView):
|
|
model = CategorisationRule
|
|
permission_required = ("agenda_culturel.change_categorisationrule")
|
|
form_class = CategorisationRuleImportForm
|
|
success_url = reverse_lazy('categorisation_rules')
|
|
success_message = _('The categorisation rule has been successfully modified.')
|
|
|
|
|
|
class CategorisationRuleDeleteView(SuccessMessageMixin, PermissionRequiredMixin, LoginRequiredMixin, DeleteView):
|
|
model = CategorisationRule
|
|
permission_required = ("agenda_culturel.delete_categorisationrule")
|
|
success_url = reverse_lazy('categorisation_rules')
|
|
success_message = _('The categorisation rule has been successfully deleted.')
|
|
|
|
@login_required(login_url="/accounts/login/")
|
|
@permission_required('agenda_culturel.apply_categorisationrules')
|
|
def apply_categorisation_rules(request):
|
|
|
|
if request.method == 'POST':
|
|
form = CategorisationForm(request.POST)
|
|
if form.is_valid():
|
|
nb = 0
|
|
for epk, c in form.get_validated():
|
|
e = Event.objects.get(pk=epk)
|
|
logger.warning("cat " + c)
|
|
cat = Category.objects.get(name=c)
|
|
e.category = cat
|
|
e.save()
|
|
nb += 1
|
|
|
|
if nb != 0:
|
|
if nb == 1:
|
|
messages.success(request, _("The rules were successfully applied and 1 event was categorised."))
|
|
else:
|
|
messages.success(request, _("The rules were successfully applied and {} events were categorised.").format(nb))
|
|
else:
|
|
messages.info(request, _("The rules were successfully applied and no events were categorised."))
|
|
|
|
return HttpResponseRedirect(reverse_lazy("categorisation_rules"))
|
|
else:
|
|
return render(request, 'agenda_culturel/categorise_events_form.html', context={'form': form})
|
|
else:
|
|
# first we check if events are not correctly categorised
|
|
to_categorise = []
|
|
for e in Event.objects.exclude(category=Category.get_default_category_id()):
|
|
c = CategorisationRule.match_rules(e)
|
|
if c and c != e.category:
|
|
to_categorise.append((e, c))
|
|
|
|
|
|
# then we apply rules on events without category
|
|
nb = 0
|
|
for e in Event.objects.filter(category=Category.get_default_category_id()):
|
|
success = CategorisationRule.apply_rules(e)
|
|
if success:
|
|
nb += 1
|
|
e.save()
|
|
|
|
# set messages
|
|
if nb != 0:
|
|
if nb == 1:
|
|
messages.success(request, _("The rules were successfully applied and 1 event was categorised."))
|
|
else:
|
|
messages.success(request, _("The rules were successfully applied and {} events were categorised.").format(nb))
|
|
else:
|
|
messages.info(request, _("The rules were successfully applied and no events were categorised."))
|
|
|
|
if len(to_categorise) != 0:
|
|
form = CategorisationForm(events=to_categorise)
|
|
return render(request, 'agenda_culturel/categorise_events_form.html', context={'form': form,
|
|
'events': dict((e.pk, e) for e, c in to_categorise),
|
|
'categories': dict((e.pk, c) for e, c in to_categorise)})
|
|
else:
|
|
return HttpResponseRedirect(reverse_lazy("categorisation_rules"))
|
|
|
|
|
|
#########################
|
|
## Moderation Q&A
|
|
#########################
|
|
|
|
|
|
class ModerationQuestionListView(PermissionRequiredMixin, ListView):
|
|
model = ModerationQuestion
|
|
paginate_by = 10
|
|
permission_required = ("agenda_culturel.view_moderationquestion")
|
|
|
|
class ModerationQuestionCreateView(SuccessMessageMixin, PermissionRequiredMixin, CreateView):
|
|
model = ModerationQuestion
|
|
permission_required = ("agenda_culturel.add_moderationquestion")
|
|
|
|
def get_success_url(self):
|
|
return reverse_lazy('view_mquestion', kwargs={'pk': self.object.pk})
|
|
form_class = ModerationQuestionForm
|
|
success_message = _('The moderation question has been created with success.')
|
|
|
|
|
|
class ModerationQuestionDetailView(PermissionRequiredMixin, DetailView):
|
|
model = ModerationQuestion
|
|
permission_required = ("agenda_culturel.view_moderationquestion", "agenda_culturel.view_moderationanswer")
|
|
|
|
|
|
class ModerationQuestionUpdateView(PermissionRequiredMixin, UpdateView):
|
|
model = ModerationQuestion
|
|
fields = ['question']
|
|
permission_required = ("agenda_culturel.change_moderationquestion")
|
|
|
|
class ModerationQuestionDeleteView(PermissionRequiredMixin, DeleteView):
|
|
model = ModerationQuestion
|
|
permission_required = ("agenda_culturel.delete_moderationquestion")
|
|
success_url = reverse_lazy('view_mquestions')
|
|
|
|
|
|
class ModerationAnswerCreateView(PermissionRequiredMixin, SuccessMessageMixin, CreateView):
|
|
model = ModerationAnswer
|
|
permission_required = ("agenda_culturel.add_answerquestion")
|
|
form_class = ModerationAnswerForm
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
context['question'] = get_object_or_404(ModerationQuestion, pk=self.kwargs['qpk'])
|
|
return context
|
|
|
|
def form_valid(self, form):
|
|
form.instance.question = ModerationQuestion.objects.get(pk=self.kwargs['qpk'])
|
|
return super().form_valid(form)
|
|
|
|
def get_success_url(self):
|
|
return reverse_lazy('view_mquestion', kwargs={'pk': self.kwargs['qpk']})
|
|
|
|
|
|
class ModerationAnswerUpdateView(PermissionRequiredMixin, UpdateView):
|
|
model = ModerationAnswer
|
|
fields = ['answer', 'adds_tags', 'removes_tags']
|
|
permission_required = ("agenda_culturel.change_answerquestion")
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
context['question'] = get_object_or_404(ModerationQuestion, pk=self.kwargs['qpk'])
|
|
return context
|
|
|
|
def get_success_url(self):
|
|
return reverse_lazy('view_mquestion', kwargs={'pk': self.kwargs['qpk']})
|
|
|
|
class ModerationAnswerDeleteView(PermissionRequiredMixin, DeleteView):
|
|
model = ModerationAnswer
|
|
permission_required = ("agenda_culturel.delete_answerquestion")
|
|
|
|
|
|
#########################
|
|
## Places
|
|
#########################
|
|
|
|
class PlaceListView(ListView):
|
|
model = Place
|
|
paginate_by = 10
|
|
ordering = ['name__unaccent']
|
|
|
|
|
|
class PlaceDetailView(ListView):
|
|
model = Place
|
|
template_name = "agenda_culturel/place_detail.html"
|
|
paginate_by = 10
|
|
|
|
def get_queryset(self):
|
|
self.place = get_object_or_404(Place, pk=self.kwargs['pk'])
|
|
return Event.objects.filter(exact_location=self.place).order_by("-start_day")
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
context['object'] = self.place
|
|
return context
|
|
|
|
|
|
class UpdatePlaces:
|
|
def form_valid(self, form):
|
|
result = super().form_valid(form)
|
|
p = form.instance
|
|
|
|
if not hasattr(self, "nb_applied"):
|
|
self.nb_applied = 0
|
|
|
|
# if required, find all matching events
|
|
if form.apply():
|
|
self.nb_applied += p.associate_matching_events()
|
|
|
|
if self.nb_applied > 1:
|
|
messages.success(self.request, _("{} events have been updated.").format(self.nb_applied))
|
|
elif self.nb_applied == 1:
|
|
messages.success(self.request, _("1 event has been updated."))
|
|
else:
|
|
messages.info(self.request, _("No events have been modified."))
|
|
return result
|
|
|
|
|
|
|
|
class PlaceUpdateView(UpdatePlaces, PermissionRequiredMixin, SuccessMessageMixin, UpdateView):
|
|
model = Place
|
|
permission_required = ("agenda_culturel.change_place")
|
|
success_message = _('The place has been successfully updated.')
|
|
form_class = PlaceForm
|
|
|
|
|
|
class PlaceCreateView(UpdatePlaces, PermissionRequiredMixin, SuccessMessageMixin, CreateView):
|
|
model = Place
|
|
permission_required = ("agenda_culturel.add_place")
|
|
success_message = _('The place has been successfully created.')
|
|
form_class = PlaceForm
|
|
|
|
|
|
class PlaceDeleteView(PermissionRequiredMixin, DeleteView):
|
|
model = Place
|
|
permission_required = ("agenda_culturel.delete_place")
|
|
success_url = reverse_lazy('view_places')
|
|
|
|
class UnknownPlacesListView(PermissionRequiredMixin, ListView):
|
|
model = Event
|
|
permission_required = ("agenda_culturel.add_place")
|
|
paginate_by = 10
|
|
ordering = ['-pk']
|
|
template_name = 'agenda_culturel/place_unknown_list.html'
|
|
queryset = Event.objects.filter(exact_location__isnull=True)
|
|
|
|
def fix_unknown_places(request):
|
|
# get all places
|
|
places = Place.objects.all()
|
|
# get all events without exact location
|
|
u_events = Event.objects.filter(exact_location__isnull=True)
|
|
|
|
to_be_updated = []
|
|
# try to find matches
|
|
for ue in u_events:
|
|
for p in places:
|
|
if p.match(ue):
|
|
ue.exact_location = p
|
|
to_be_updated.append(ue)
|
|
continue
|
|
# update events with a location
|
|
Event.objects.bulk_update(to_be_updated, fields=["exact_location"])
|
|
|
|
# create a success message
|
|
nb = len(to_be_updated)
|
|
if nb > 1:
|
|
messages.success(request, _("{} events have been updated.").format(nb))
|
|
elif nb == 1:
|
|
messages.success(request, _("1 event has been updated."))
|
|
else:
|
|
messages.info(request, _("No events have been modified."))
|
|
|
|
# come back to the list of places
|
|
return HttpResponseRedirect(reverse_lazy("view_unknown_places"))
|
|
|
|
|
|
class UnknownPlaceAddView(PermissionRequiredMixin, SuccessMessageMixin, UpdateView):
|
|
model = Event
|
|
permission_required = ("agenda_culturel.change_place", "agenda_culturel.change_event")
|
|
form_class = EventAddPlaceForm
|
|
template_name = 'agenda_culturel/place_unknown_form.html'
|
|
|
|
def form_valid(self, form):
|
|
|
|
self.modified_event = form.cleaned_data.get('place')
|
|
result = super().form_valid(form)
|
|
|
|
if form.cleaned_data.get('place'):
|
|
messages.success(self.request, _("The selected place has been assigned to the event."))
|
|
if form.cleaned_data.get("add_alias"):
|
|
messages.success(self.request, _("A new alias has been added to the selected place."))
|
|
|
|
nb_applied = form.cleaned_data.get('place').associate_matching_events()
|
|
|
|
if nb_applied > 1:
|
|
messages.success(self.request, _("{} events have been updated.").format(nb_applied))
|
|
elif nb_applied == 1:
|
|
messages.success(self.request, _("1 event has been updated."))
|
|
else:
|
|
messages.info(self.request, _("No events have been modified."))
|
|
|
|
return result
|
|
|
|
def get_success_url(self):
|
|
if self.modified_event:
|
|
return reverse_lazy('view_unknown_places')
|
|
else:
|
|
return reverse_lazy('add_place_from_event', args=[self.object.pk])
|
|
|
|
|
|
class PlaceFromEventCreateView(PlaceCreateView):
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
context["event"] = self.event
|
|
return context
|
|
|
|
def get_initial(self, *args, **kwargs):
|
|
initial = super().get_initial(**kwargs)
|
|
self.event = get_object_or_404(Event, pk=self.kwargs['pk'])
|
|
if self.event.location:
|
|
initial['aliases'] = [self.event.location]
|
|
return initial
|
|
|