lingo/lingo/invoicing/views/campaign.py

282 lines
11 KiB
Python

# lingo - payment and billing system
# Copyright (C) 2023 Entr'ouvert
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from django.contrib import messages
from django.db.models import Count, IntegerField, OuterRef, Subquery, Value
from django.db.models.functions import Coalesce
from django.shortcuts import get_object_or_404, redirect
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from django.views.generic import CreateView, DeleteView, DetailView, FormView, UpdateView
from lingo.agendas.chrono import ChronoError, mark_events_invoiced, unlock_events_check
from lingo.invoicing.forms import CampaignDatesForm, CampaignForm
from lingo.invoicing.models import Campaign, DraftInvoice, DraftInvoiceLine, InvoiceLine, Pool, Regie
class CampaignAddView(CreateView):
template_name = 'lingo/invoicing/manager_campaign_form.html'
model = Campaign
form_class = CampaignForm
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
return super().dispatch(request, *args, **kwargs)
def get_context_data(self, **kwargs):
kwargs['regie'] = self.regie
return super().get_context_data(**kwargs)
def get_form_kwargs(self):
kwargs = super().get_form_kwargs()
kwargs['instance'] = Campaign(regie=self.regie)
return kwargs
def get_success_url(self):
return reverse('lingo-manager-invoicing-campaign-detail', args=[self.regie.pk, self.object.pk])
campaign_add = CampaignAddView.as_view()
class CampaignDetailView(DetailView):
template_name = 'lingo/invoicing/manager_campaign_detail.html'
model = Campaign
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
return super().dispatch(request, *args, **kwargs)
def get_queryset(self):
return super().get_queryset().filter(regie=self.regie)
def get_context_data(self, **kwargs):
draft_lines = DraftInvoiceLine.objects.filter(pool=OuterRef('pk')).order_by().values('pool')
count_draft_error = draft_lines.filter(status='error').annotate(count=Count('pool')).values('count')
count_draft_warning = (
draft_lines.filter(status='warning').annotate(count=Count('pool')).values('count')
)
count_draft_success = (
draft_lines.filter(status='success').annotate(count=Count('pool')).values('count')
)
lines = InvoiceLine.objects.filter(pool=OuterRef('pk')).order_by().values('pool')
count_error = (
lines.filter(status='error', error_status='').annotate(count=Count('pool')).values('count')
)
count_warning = lines.filter(status='warning').annotate(count=Count('pool')).values('count')
count_success = lines.filter(status='success').annotate(count=Count('pool')).values('count')
kwargs['regie'] = self.regie
kwargs['pools'] = self.object.pool_set.annotate(
draft_error_count=Coalesce(Subquery(count_draft_error, output_field=IntegerField()), Value(0)),
draft_warning_count=Coalesce(
Subquery(count_draft_warning, output_field=IntegerField()), Value(0)
),
draft_success_count=Coalesce(
Subquery(count_draft_success, output_field=IntegerField()), Value(0)
),
error_count=Coalesce(Subquery(count_error, output_field=IntegerField()), Value(0)),
warning_count=Coalesce(Subquery(count_warning, output_field=IntegerField()), Value(0)),
success_count=Coalesce(Subquery(count_success, output_field=IntegerField()), Value(0)),
).order_by('-created_at')
kwargs['has_running_pool'] = any(p.status in ['registered', 'running'] for p in kwargs['pools'])
kwargs['has_real_pool'] = any(not p.draft for p in kwargs['pools'])
kwargs['has_real_completed_pool'] = any(
not p.draft and p.status == 'completed' for p in kwargs['pools']
)
if self.object.invalid:
messages.warning(self.request, _('The last pool is invalid, please start a new pool.'))
return super().get_context_data(**kwargs)
campaign_detail = CampaignDetailView.as_view()
class CampaignEditView(UpdateView):
template_name = 'lingo/invoicing/manager_campaign_form.html'
model = Campaign
form_class = CampaignForm
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
return super().dispatch(request, *args, **kwargs)
def get_queryset(self):
return (
super()
.get_queryset()
.filter(regie=self.regie, finalized=False)
.exclude(pool__draft=False)
.exclude(pool__status__in=['registered', 'running'])
)
def get_context_data(self, **kwargs):
kwargs['regie'] = self.regie
return super().get_context_data(**kwargs)
def get_success_url(self):
return reverse('lingo-manager-invoicing-campaign-detail', args=[self.regie.pk, self.object.pk])
campaign_edit = CampaignEditView.as_view()
class CampaignDatesEditView(UpdateView):
template_name = 'lingo/invoicing/manager_campaign_dates_form.html'
model = Campaign
form_class = CampaignDatesForm
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
return super().dispatch(request, *args, **kwargs)
def get_queryset(self):
return (
super()
.get_queryset()
.filter(regie=self.regie, finalized=False)
.exclude(pool__status__in=['registered', 'running'])
)
def get_context_data(self, **kwargs):
kwargs['regie'] = self.regie
return super().get_context_data(**kwargs)
def get_success_url(self):
return '%s#open:dates' % reverse(
'lingo-manager-invoicing-campaign-detail', args=[self.regie.pk, self.object.pk]
)
campaign_dates_edit = CampaignDatesEditView.as_view()
class CampaignDeleteView(DeleteView):
template_name = 'lingo/manager_confirm_delete.html'
model = Campaign
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
return super().dispatch(request, *args, **kwargs)
def get_queryset(self):
return (
super()
.get_queryset()
.filter(regie=self.regie, finalized=False)
.exclude(pool__draft=False)
.exclude(pool__status__in=['registered', 'running'])
)
def delete(self, request, *args, **kwargs):
self.object = self.get_object()
DraftInvoiceLine.objects.filter(pool__campaign=self.object).delete()
DraftInvoice.objects.filter(pool__campaign=self.object).delete()
Pool.objects.filter(campaign=self.object).delete()
return super().delete(request, *args, **kwargs)
def get_success_url(self):
return '%s#open:campaigns' % reverse('lingo-manager-invoicing-regie-detail', args=[self.regie.pk])
campaign_delete = CampaignDeleteView.as_view()
class CampaignUnlockCheckView(FormView):
template_name = 'lingo/invoicing/manager_campaign_unlock_check.html'
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
self.object = get_object_or_404(
Campaign.objects.filter(regie=self.regie, invalid=False, finalized=False)
.exclude(pool__draft=False)
.exclude(pool__status__in=['registered', 'running']),
pk=kwargs['pk'],
)
return super().dispatch(request, *args, **kwargs)
def get_context_data(self, **kwargs):
kwargs['form'] = None
kwargs['regie'] = self.regie
kwargs['object'] = self.object
return super().get_context_data(**kwargs)
def post(self, request, *args, **kwargs):
self.object.mark_as_invalid()
agendas = [a.slug for a in self.object.agendas.all()]
if agendas:
try:
unlock_events_check(
agenda_slugs=agendas,
date_start=self.object.date_start,
date_end=self.object.date_end,
)
except ChronoError as e:
messages.error(self.request, _('Fail to unlock events check: %s') % e)
return redirect(
'%s#open:pools'
% reverse('lingo-manager-invoicing-campaign-detail', args=[self.regie.pk, self.object.pk])
)
campaign_unlock_check = CampaignUnlockCheckView.as_view()
class CampaignFinalizeView(FormView):
template_name = 'lingo/invoicing/manager_campaign_finalize.html'
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
self.object = get_object_or_404(
Campaign.objects.filter(regie=self.regie, invalid=False, finalized=False).filter(
pk__in=Pool.objects.filter(draft=False, status='completed').values('campaign')
),
pk=kwargs['pk'],
)
return super().dispatch(request, *args, **kwargs)
def get_context_data(self, **kwargs):
kwargs['form'] = None
kwargs['regie'] = self.regie
kwargs['object'] = self.object
return super().get_context_data(**kwargs)
def post(self, request, *args, **kwargs):
try:
agendas = [a.slug for a in self.object.agendas.all()]
if agendas:
try:
mark_events_invoiced(
agenda_slugs=agendas,
date_start=self.object.date_start,
date_end=self.object.date_end,
)
except ChronoError as e:
messages.error(self.request, _('Fail to mark events as invoiced: %s') % e)
raise
except ChronoError:
pass
else:
self.object.mark_as_finalized()
return redirect(
'%s#open:pools'
% reverse('lingo-manager-invoicing-campaign-detail', args=[self.regie.pk, self.object.pk])
)
campaign_finalize = CampaignFinalizeView.as_view()