invoicing: split views (#78125)

This commit is contained in:
Lauréline Guérin 2023-06-01 11:27:40 +02:00
parent 6e46d0aef0
commit dc8b89112a
No known key found for this signature in database
GPG Key ID: 1FAB9B9B4F93D473
9 changed files with 1041 additions and 959 deletions

View File

@ -16,126 +16,129 @@
from django.urls import path
from . import views
from .views import campaign as campaign_views
from .views import home as home_views
from .views import pool as pool_views
from .views import regie as regie_views
urlpatterns = [
path('', views.home, name='lingo-manager-invoicing-home'),
path('regies/', views.regies_list, name='lingo-manager-invoicing-regie-list'),
path('', home_views.home, name='lingo-manager-invoicing-home'),
path('regies/', regie_views.regies_list, name='lingo-manager-invoicing-regie-list'),
path(
'regie/add/',
views.regie_add,
regie_views.regie_add,
name='lingo-manager-invoicing-regie-add',
),
path(
'regie/<int:pk>/',
views.regie_detail,
regie_views.regie_detail,
name='lingo-manager-invoicing-regie-detail',
),
path(
'regie/<int:pk>/edit/',
views.regie_edit,
regie_views.regie_edit,
name='lingo-manager-invoicing-regie-edit',
),
path(
'regie/<int:pk>/delete/',
views.regie_delete,
regie_views.regie_delete,
name='lingo-manager-invoicing-regie-delete',
),
path('regies/import/', views.regies_import, name='lingo-manager-invoicing-regie-import'),
path('regies/export/', views.regies_export, name='lingo-manager-invoicing-regie-export'),
path('regies/import/', regie_views.regies_import, name='lingo-manager-invoicing-regie-import'),
path('regies/export/', regie_views.regies_export, name='lingo-manager-invoicing-regie-export'),
path(
'regie/<int:regie_pk>/campaign/add/',
views.campaign_add,
campaign_views.campaign_add,
name='lingo-manager-invoicing-campaign-add',
),
path(
'regie/<int:regie_pk>/campaign/<int:pk>/',
views.campaign_detail,
campaign_views.campaign_detail,
name='lingo-manager-invoicing-campaign-detail',
),
path(
'regie/<int:regie_pk>/campaign/<int:pk>/edit/',
views.campaign_edit,
campaign_views.campaign_edit,
name='lingo-manager-invoicing-campaign-edit',
),
path(
'regie/<int:regie_pk>/campaign/<int:pk>/dates/edit/',
views.campaign_dates_edit,
campaign_views.campaign_dates_edit,
name='lingo-manager-invoicing-campaign-dates-edit',
),
path(
'regie/<int:regie_pk>/campaign/<int:pk>/delete/',
views.campaign_delete,
campaign_views.campaign_delete,
name='lingo-manager-invoicing-campaign-delete',
),
path(
'regie/<int:regie_pk>/campaign/<int:pk>/unlock-check/',
views.campaign_unlock_check,
campaign_views.campaign_unlock_check,
name='lingo-manager-invoicing-campaign-unlock-check',
),
path(
'regie/<int:regie_pk>/campaign/<int:pk>/finalize/',
views.campaign_finalize,
campaign_views.campaign_finalize,
name='lingo-manager-invoicing-campaign-finalize',
),
path(
'regie/<int:regie_pk>/campaign/<int:pk>/pool/add/',
views.pool_add,
pool_views.pool_add,
name='lingo-manager-invoicing-pool-add',
),
path(
'regie/<int:regie_pk>/campaign/<int:pk>/pool/<int:pool_pk>/',
views.pool_detail,
pool_views.pool_detail,
name='lingo-manager-invoicing-pool-detail',
),
path(
'regie/<int:regie_pk>/campaign/<int:pk>/pool/<int:pool_pk>/journal/',
views.pool_journal,
pool_views.pool_journal,
name='lingo-manager-invoicing-pool-journal',
),
path(
'regie/<int:regie_pk>/campaign/<int:pk>/pool/<int:pool_pk>/promote/',
views.pool_promote,
pool_views.pool_promote,
name='lingo-manager-invoicing-pool-promote',
),
path(
'regie/<int:regie_pk>/campaign/<int:pk>/pool/<int:pool_pk>/delete/',
views.pool_delete,
pool_views.pool_delete,
name='lingo-manager-invoicing-pool-delete',
),
path(
'regie/<int:regie_pk>/campaign/<int:pk>/pool/<int:pool_pk>/invoice/<int:invoice_pk>/pdf/',
views.invoice_pdf,
pool_views.invoice_pdf,
name='lingo-manager-invoicing-invoice-pdf',
),
path(
'ajax/regie/<int:regie_pk>/campaign/<int:pk>/pool/<int:pool_pk>/invoice/<int:invoice_pk>/lines/',
views.invoice_line_list,
pool_views.invoice_line_list,
name='lingo-manager-invoicing-invoice-line-list',
),
path(
'ajax/regie/<int:regie_pk>/campaign/<int:pk>/pool/<int:pool_pk>/line/<int:line_pk>/<slug:status>/',
views.line_set_error_status,
pool_views.line_set_error_status,
name='lingo-manager-invoicing-line-set-error-status',
),
path(
'regie/<int:regie_pk>/non-invoiced-lines/',
views.non_invoiced_line_list,
regie_views.non_invoiced_line_list,
name='lingo-manager-invoicing-non-invoiced-line-list',
),
path(
'regie/<int:regie_pk>/invoices/',
views.regie_invoice_list,
regie_views.regie_invoice_list,
name='lingo-manager-invoicing-regie-invoice-list',
),
path(
'regie/<int:regie_pk>/invoice/<int:invoice_pk>/pdf/',
views.regie_invoice_pdf,
regie_views.regie_invoice_pdf,
name='lingo-manager-invoicing-regie-invoice-pdf',
),
path(
'ajax/regie/<int:regie_pk>/invoice/<int:invoice_pk>/lines/',
views.regie_invoice_line_list,
regie_views.regie_invoice_line_list,
name='lingo-manager-invoicing-regie-invoice-line-list',
),
]

View File

@ -1,929 +0,0 @@
# lingo - payment and billing system
# Copyright (C) 2022 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/>.
import collections
import datetime
import json
from django.contrib import messages
from django.db import transaction
from django.db.models import CharField, Count, IntegerField, JSONField, OuterRef, Subquery, Value
from django.db.models.functions import Coalesce
from django.http import Http404, HttpResponse
from django.shortcuts import get_object_or_404, redirect
from django.urls import reverse, reverse_lazy
from django.utils.translation import gettext_lazy as _
from django.utils.translation import ngettext
from django.views.generic import (
CreateView,
DeleteView,
DetailView,
FormView,
ListView,
TemplateView,
UpdateView,
)
from weasyprint import HTML
from lingo.agendas.chrono import ChronoError, mark_events_invoiced, unlock_events_check
from lingo.agendas.models import Agenda
from lingo.invoicing.forms import (
CampaignDatesForm,
CampaignForm,
DraftInvoiceFilterSet,
DraftInvoiceLineFilterSet,
InvoiceFilterSet,
InvoiceLineFilterSet,
RegieInvoiceFilterSet,
)
from lingo.invoicing.models import (
Campaign,
Counter,
DraftInvoice,
DraftInvoiceLine,
InjectedLine,
Invoice,
InvoiceLine,
InvoicePayment,
Pool,
Regie,
RegieImportError,
)
from lingo.pricing.forms import ImportForm
def is_ajax(request):
return request.headers.get('x-requested-with') == 'XMLHttpRequest'
def import_regies(data):
results = collections.defaultdict(list)
with transaction.atomic():
regies = data.get('regies', [])
for regie in regies:
created, regie_obj = Regie.import_json(regie)
if created:
results['created'].append(regie_obj)
else:
results['updated'].append(regie_obj)
return results
class HomeView(TemplateView):
template_name = 'lingo/invoicing/manager_home.html'
home = HomeView.as_view()
class RegiesListView(ListView):
template_name = 'lingo/invoicing/manager_regie_list.html'
model = Regie
regies_list = RegiesListView.as_view()
class RegieAddView(CreateView):
template_name = 'lingo/invoicing/manager_regie_form.html'
model = Regie
fields = ['label', 'description', 'cashier_role']
def get_success_url(self):
return reverse('lingo-manager-invoicing-regie-detail', args=[self.object.pk])
regie_add = RegieAddView.as_view()
class RegieDetailView(DetailView):
template_name = 'lingo/invoicing/manager_regie_detail.html'
model = Regie
def get_context_data(self, **kwargs):
kwargs['regie'] = self.object
kwargs['agendas'] = Agenda.objects.filter(regie=self.object)
kwargs['campaigns'] = self.object.campaign_set.all().order_by('-date_start')
has_related_objects = False
if kwargs['campaigns']:
has_related_objects = True
elif self.object.injectedline_set.exists():
has_related_objects = True
kwargs['has_related_objects'] = has_related_objects
return super().get_context_data(**kwargs)
regie_detail = RegieDetailView.as_view()
class RegieEditView(UpdateView):
template_name = 'lingo/invoicing/manager_regie_form.html'
model = Regie
fields = ['label', 'description', 'cashier_role', 'counter_name', 'number_format']
def get_success_url(self):
return reverse('lingo-manager-invoicing-regie-detail', args=[self.object.pk])
regie_edit = RegieEditView.as_view()
class RegieDeleteView(DeleteView):
template_name = 'lingo/manager_confirm_delete.html'
model = Regie
def get_queryset(self):
return super().get_queryset().filter(campaign__isnull=True, injectedline__isnull=True)
def delete(self, request, *args, **kwargs):
self.object = self.get_object()
Counter.objects.filter(regie=self.object).delete()
return super().delete(request, *args, **kwargs)
def get_success_url(self):
return reverse('lingo-manager-invoicing-regie-list')
regie_delete = RegieDeleteView.as_view()
class RegiesExportView(ListView):
model = Regie
def get(self, request, *args, **kwargs):
response = HttpResponse(content_type='application/json')
today = datetime.date.today()
attachment = 'attachment; filename="export_regies_{}.json"'.format(today.strftime('%Y%m%d'))
response['Content-Disposition'] = attachment
json.dump({'regies': [regie.export_json() for regie in self.get_queryset()]}, response, indent=2)
return response
regies_export = RegiesExportView.as_view()
class RegiesImportView(FormView):
form_class = ImportForm
template_name = 'lingo/invoicing/manager_import.html'
success_url = reverse_lazy('lingo-manager-invoicing-regie-list')
def form_valid(self, form):
try:
config_json = json.loads(self.request.FILES['config_json'].read())
except ValueError:
form.add_error('config_json', _('File is not in the expected JSON format.'))
return self.form_invalid(form)
try:
results = import_regies(config_json)
except RegieImportError as exc:
form.add_error('config_json', '%s' % exc)
return self.form_invalid(form)
import_messages = {
'create': lambda x: ngettext(
'A regie was created.',
'%(count)d regies were created.',
x,
),
'update': lambda x: ngettext(
'A regie was updated.',
'%(count)d regie were updated.',
x,
),
}
create_message = _('No regie created.')
update_message = _('No regie updated.')
created = len(results.get('created', []))
updated = len(results.get('updated', []))
if created:
create_message = import_messages.get('create')(created) % {'count': created}
if updated:
update_message = import_messages.get('update')(updated) % {'count': updated}
message = "%s %s" % (create_message, update_message)
messages.info(self.request, message)
return super().form_valid(form)
regies_import = RegiesImportView.as_view()
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()
class PoolDetailView(ListView):
template_name = 'lingo/invoicing/manager_pool_detail.html'
paginate_by = 100
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
self.campaign = get_object_or_404(Campaign, pk=kwargs['pk'], regie=self.regie)
self.object = get_object_or_404(Pool, pk=kwargs['pool_pk'], campaign=self.campaign)
return super().dispatch(request, *args, **kwargs)
def get_queryset(self):
invoice_model = Invoice
filter_model = InvoiceFilterSet
if self.object.draft:
invoice_model = DraftInvoice
filter_model = DraftInvoiceFilterSet
data = self.request.GET or None
self.filterset = filter_model(
data=data,
queryset=invoice_model.objects.filter(pool=self.object).order_by('created_at'),
pool=self.object,
)
return self.filterset.qs
def get_context_data(self, **kwargs):
kwargs['regie'] = self.regie
kwargs['object'] = self.campaign
kwargs['pool'] = self.object
kwargs['filterset'] = self.filterset
line_model = InvoiceLine
line_values = ['status', 'error_status']
if self.object.draft:
line_model = DraftInvoiceLine
line_values = ['status']
all_lines = line_model.objects.filter(pool=self.object).values(*line_values)
self.object.error_count = len(
[line for line in all_lines if line['status'] == 'error' and not line.get('error_status')]
)
self.object.warning_count = len([line for line in all_lines if line['status'] == 'warning'])
self.object.success_count = len([line for line in all_lines if line['status'] == 'success'])
kwargs['has_running_pool'] = any(
p.status in ['registered', 'running'] for p in self.campaign.pool_set.all()
)
return super().get_context_data(**kwargs)
pool_detail = PoolDetailView.as_view()
class PoolJournalView(ListView):
template_name = 'lingo/invoicing/manager_pool_journal.html'
paginate_by = 100
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
self.campaign = get_object_or_404(Campaign, pk=kwargs['pk'], regie=self.regie)
self.object = get_object_or_404(Pool, pk=kwargs['pool_pk'], campaign=self.campaign)
return super().dispatch(request, *args, **kwargs)
def get_queryset(self):
line_model = InvoiceLine
filter_model = InvoiceLineFilterSet
if self.object.draft:
line_model = DraftInvoiceLine
filter_model = DraftInvoiceLineFilterSet
all_lines = line_model.objects.filter(pool=self.object).order_by('pk').select_related('invoice')
self.object.error_count = len(
[line for line in all_lines if line.status == 'error' and not getattr(line, 'error_status', '')]
)
self.object.warning_count = len([line for line in all_lines if line.status == 'warning'])
self.object.success_count = len([line for line in all_lines if line.status == 'success'])
data = self.request.GET or None
self.filterset = filter_model(data=data, queryset=all_lines, pool=self.object)
return self.filterset.qs if data and [v for v in data.values() if v] else all_lines
def get_context_data(self, **kwargs):
kwargs['regie'] = self.regie
kwargs['object'] = self.campaign
kwargs['pool'] = self.object
kwargs['filterset'] = self.filterset
return super().get_context_data(**kwargs)
pool_journal = PoolJournalView.as_view()
class PoolAddView(FormView):
template_name = 'lingo/invoicing/manager_pool_add.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, 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_valid()
self.object.generate()
return redirect(
'%s#open:pools'
% reverse('lingo-manager-invoicing-campaign-detail', args=[self.regie.pk, self.object.pk])
)
pool_add = PoolAddView.as_view()
class PoolPromoteView(FormView):
template_name = 'lingo/invoicing/manager_pool_promote.html'
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
self.object = get_object_or_404(
Pool,
campaign__id=kwargs['pk'],
campaign__regie=self.regie,
campaign__invalid=False,
campaign__finalized=False,
pk=kwargs['pool_pk'],
draft=True,
status='completed',
)
if not self.object.is_last:
raise Http404
return super().dispatch(request, *args, **kwargs)
def get_context_data(self, **kwargs):
kwargs['form'] = None
kwargs['regie'] = self.regie
kwargs['object'] = self.object.campaign
kwargs['pool'] = self.object
return super().get_context_data(**kwargs)
def post(self, request, *args, **kwargs):
self.object.promote()
return redirect(
'%s#open:pools'
% reverse(
'lingo-manager-invoicing-campaign-detail', args=[self.regie.pk, self.object.campaign.pk]
)
)
pool_promote = PoolPromoteView.as_view()
class PoolDeleteView(DeleteView):
template_name = 'lingo/manager_confirm_delete.html'
model = Pool
pk_url_kwarg = 'pool_pk'
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
self.campaign = get_object_or_404(
Campaign.objects.filter(regie=self.regie, finalized=False).exclude(
pool__status__in=['registered', 'running']
),
pk=kwargs['pk'],
)
return super().dispatch(request, *args, **kwargs)
def get_queryset(self):
return self.campaign.pool_set.all()
def delete(self, request, *args, **kwargs):
self.object = self.get_object()
if self.object.is_last and self.object.draft:
self.campaign.mark_as_invalid()
invoice_model = Invoice
line_model = InvoiceLine
if self.object.draft:
invoice_model = DraftInvoice
line_model = DraftInvoiceLine
line_model.objects.filter(pool=self.object).delete()
invoice_model.objects.filter(pool=self.object).delete()
return super().delete(request, *args, **kwargs)
def get_success_url(self):
return '%s#open:pools' % reverse(
'lingo-manager-invoicing-campaign-detail', args=[self.regie.pk, self.campaign.pk]
)
pool_delete = PoolDeleteView.as_view()
class PDFMixin:
def get(self, request, *args, **kwargs):
self.object = self.get_object()
result = self.object.html()
if 'html' in request.GET:
return HttpResponse(result)
html = HTML(string=result)
pdf = html.write_pdf()
response = HttpResponse(pdf, content_type='application/pdf')
response['Content-Disposition'] = 'attachment; filename="%s.pdf"' % self.object.formatted_number
return response
class InvoicePDFView(PDFMixin, DetailView):
pk_url_kwarg = 'invoice_pk'
model = Invoice
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
self.pool = get_object_or_404(
Pool, pk=kwargs['pool_pk'], campaign=kwargs['pk'], campaign__regie=self.regie
)
return super().dispatch(request, *args, **kwargs)
def get_queryset(self):
invoice_model = Invoice
if self.pool.draft:
invoice_model = DraftInvoice
return invoice_model.objects.filter(pool=self.pool)
invoice_pdf = InvoicePDFView.as_view()
class InvoiceLineListView(ListView):
template_name = 'lingo/invoicing/manager_invoice_lines.html'
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
self.pool = get_object_or_404(
Pool, pk=kwargs['pool_pk'], campaign_id=kwargs['pk'], campaign__regie=self.regie
)
invoice_model = Invoice
if self.pool.draft:
invoice_model = DraftInvoice
self.invoice = get_object_or_404(invoice_model, pk=kwargs['invoice_pk'], pool=self.pool)
return super().dispatch(request, *args, **kwargs)
def get_queryset(self):
return self.invoice.lines.all().order_by('user_external_id', 'event_date', 'pk')
def get_context_data(self, **kwargs):
kwargs['regie'] = self.pool.campaign.regie
kwargs['object'] = self.pool.campaign
kwargs['pool'] = self.pool
if not self.pool.draft:
kwargs['invoice'] = self.invoice
kwargs['invoice_payments'] = (
InvoicePayment.objects.filter(invoice=self.invoice)
.select_related('payment')
.order_by('created_at')
)
return super().get_context_data(**kwargs)
invoice_line_list = InvoiceLineListView.as_view()
class LineSetErrorStatusView(DetailView):
model = InvoiceLine
pk_url_kwarg = 'line_pk'
template_name = 'lingo/invoicing/manager_line_detail_fragment.html'
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(
status='error',
pool=self.kwargs['pool_pk'],
pool__campaign=self.kwargs['pk'],
pool__campaign__regie=self.regie,
)
)
def get_context_data(self, **kwargs):
kwargs['regie'] = self.regie
kwargs['object'] = self.object.pool.campaign
kwargs['pool'] = self.object.pool
kwargs['line'] = self.object
return super().get_context_data(**kwargs)
def get(self, request, *args, **kwargs):
self.object = self.get_object()
error_status = kwargs['status']
if error_status == 'reset':
self.object.error_status = ''
elif error_status == 'ignore':
self.object.error_status = 'ignored'
elif error_status == 'fix':
self.object.error_status = 'fixed'
else:
raise Http404
self.object.save()
if is_ajax(self.request):
context = self.get_context_data(object=self.object)
return self.render_to_response(context)
return redirect(
reverse(
'lingo-manager-invoicing-pool-journal', args=[self.regie.pk, kwargs['pk'], kwargs['pool_pk']]
)
)
line_set_error_status = LineSetErrorStatusView.as_view()
class NonInvoicedLineListView(ListView):
template_name = 'lingo/invoicing/manager_non_invoiced_line_list.html'
paginate_by = 100
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):
fields = [
'pk',
'event_date',
'slug',
'label',
'quantity',
'unit_amount',
'total_amount',
'user_external_id',
'payer_external_id',
'payer_first_name',
'payer_last_name',
'user_first_name',
'user_last_name',
'event',
'pricing_data',
'status',
'pool_id',
]
qs1 = InvoiceLine.objects.filter(
status='error', error_status='', pool__campaign__regie=self.regie
).values(*fields)
qs2 = (
InjectedLine.objects.filter(invoiceline__isnull=True, regie=self.regie)
.annotate(
user_first_name=Value('', output_field=CharField()),
user_last_name=Value('', output_field=CharField()),
event=Value({}, output_field=JSONField()),
pricing_data=Value({}, output_field=JSONField()),
status=Value('injected', output_field=CharField()),
pool_id=Value(0, output_field=IntegerField()),
)
.values(*fields)
)
qs = qs1.union(qs2).order_by('event_date', 'user_external_id', 'label', 'pk')
return qs
def get_context_data(self, **kwargs):
kwargs['regie'] = self.regie
context = super().get_context_data(**kwargs)
pools = Pool.objects.filter(draft=False).in_bulk()
for line in context['object_list']:
if line['status'] == 'error':
line['user_name'] = InvoiceLine(
user_first_name=line['user_first_name'], user_last_name=line['user_last_name']
).user_name
line['payer_name'] = InvoiceLine(
payer_first_name=line['payer_first_name'], payer_last_name=line['payer_last_name']
).payer_name
line['error_display'] = InvoiceLine(
status=line['status'], pricing_data=line['pricing_data']
).get_error_display()
line['campaign_id'] = pools[line['pool_id']].campaign_id
line['chrono_event_url'] = InvoiceLine(event=line['event']).get_chrono_event_url()
return context
non_invoiced_line_list = NonInvoicedLineListView.as_view()
class RegieInvoiceListView(ListView):
template_name = 'lingo/invoicing/manager_invoice_list.html'
paginate_by = 100
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):
self.filterset = RegieInvoiceFilterSet(
data=self.request.GET or None,
queryset=Invoice.objects.filter(regie=self.regie).prefetch_related('pool').order_by('created_at'),
)
return self.filterset.qs
def get_context_data(self, **kwargs):
kwargs['regie'] = self.regie
kwargs['filterset'] = self.filterset
return super().get_context_data(**kwargs)
regie_invoice_list = RegieInvoiceListView.as_view()
class RegieInvoicePDFView(PDFMixin, DetailView):
pk_url_kwarg = 'invoice_pk'
model = Invoice
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)
regie_invoice_pdf = RegieInvoicePDFView.as_view()
class RegieInvoiceLineListView(ListView):
template_name = 'lingo/invoicing/manager_invoice_lines.html'
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
self.invoice = get_object_or_404(Invoice, pk=kwargs['invoice_pk'], regie=self.regie)
return super().dispatch(request, *args, **kwargs)
def get_queryset(self):
return self.invoice.lines.all().order_by('user_external_id', 'event_date', 'pk')
def get_context_data(self, **kwargs):
kwargs['regie'] = self.regie
kwargs['invoice'] = self.invoice
kwargs['invoice_payments'] = (
InvoicePayment.objects.filter(invoice=self.invoice)
.select_related('payment')
.order_by('created_at')
)
return super().get_context_data(**kwargs)
regie_invoice_line_list = RegieInvoiceLineListView.as_view()

View File

View File

@ -0,0 +1,281 @@
# 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()

View File

@ -0,0 +1,24 @@
# 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.views.generic import TemplateView
class HomeView(TemplateView):
template_name = 'lingo/invoicing/manager_home.html'
home = HomeView.as_view()

View File

@ -0,0 +1,349 @@
# 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.http import Http404
from django.shortcuts import get_object_or_404, redirect
from django.urls import reverse
from django.views.generic import DeleteView, DetailView, FormView, ListView
from lingo.invoicing.forms import (
DraftInvoiceFilterSet,
DraftInvoiceLineFilterSet,
InvoiceFilterSet,
InvoiceLineFilterSet,
)
from lingo.invoicing.models import (
Campaign,
DraftInvoice,
DraftInvoiceLine,
Invoice,
InvoiceLine,
InvoicePayment,
Pool,
Regie,
)
from lingo.invoicing.views.utils import PDFMixin
def is_ajax(request):
return request.headers.get('x-requested-with') == 'XMLHttpRequest'
class PoolDetailView(ListView):
template_name = 'lingo/invoicing/manager_pool_detail.html'
paginate_by = 100
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
self.campaign = get_object_or_404(Campaign, pk=kwargs['pk'], regie=self.regie)
self.object = get_object_or_404(Pool, pk=kwargs['pool_pk'], campaign=self.campaign)
return super().dispatch(request, *args, **kwargs)
def get_queryset(self):
invoice_model = Invoice
filter_model = InvoiceFilterSet
if self.object.draft:
invoice_model = DraftInvoice
filter_model = DraftInvoiceFilterSet
data = self.request.GET or None
self.filterset = filter_model(
data=data,
queryset=invoice_model.objects.filter(pool=self.object).order_by('created_at'),
pool=self.object,
)
return self.filterset.qs
def get_context_data(self, **kwargs):
kwargs['regie'] = self.regie
kwargs['object'] = self.campaign
kwargs['pool'] = self.object
kwargs['filterset'] = self.filterset
line_model = InvoiceLine
line_values = ['status', 'error_status']
if self.object.draft:
line_model = DraftInvoiceLine
line_values = ['status']
all_lines = line_model.objects.filter(pool=self.object).values(*line_values)
self.object.error_count = len(
[line for line in all_lines if line['status'] == 'error' and not line.get('error_status')]
)
self.object.warning_count = len([line for line in all_lines if line['status'] == 'warning'])
self.object.success_count = len([line for line in all_lines if line['status'] == 'success'])
kwargs['has_running_pool'] = any(
p.status in ['registered', 'running'] for p in self.campaign.pool_set.all()
)
return super().get_context_data(**kwargs)
pool_detail = PoolDetailView.as_view()
class PoolJournalView(ListView):
template_name = 'lingo/invoicing/manager_pool_journal.html'
paginate_by = 100
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
self.campaign = get_object_or_404(Campaign, pk=kwargs['pk'], regie=self.regie)
self.object = get_object_or_404(Pool, pk=kwargs['pool_pk'], campaign=self.campaign)
return super().dispatch(request, *args, **kwargs)
def get_queryset(self):
line_model = InvoiceLine
filter_model = InvoiceLineFilterSet
if self.object.draft:
line_model = DraftInvoiceLine
filter_model = DraftInvoiceLineFilterSet
all_lines = line_model.objects.filter(pool=self.object).order_by('pk').select_related('invoice')
self.object.error_count = len(
[line for line in all_lines if line.status == 'error' and not getattr(line, 'error_status', '')]
)
self.object.warning_count = len([line for line in all_lines if line.status == 'warning'])
self.object.success_count = len([line for line in all_lines if line.status == 'success'])
data = self.request.GET or None
self.filterset = filter_model(data=data, queryset=all_lines, pool=self.object)
return self.filterset.qs if data and [v for v in data.values() if v] else all_lines
def get_context_data(self, **kwargs):
kwargs['regie'] = self.regie
kwargs['object'] = self.campaign
kwargs['pool'] = self.object
kwargs['filterset'] = self.filterset
return super().get_context_data(**kwargs)
pool_journal = PoolJournalView.as_view()
class PoolAddView(FormView):
template_name = 'lingo/invoicing/manager_pool_add.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, 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_valid()
self.object.generate()
return redirect(
'%s#open:pools'
% reverse('lingo-manager-invoicing-campaign-detail', args=[self.regie.pk, self.object.pk])
)
pool_add = PoolAddView.as_view()
class PoolPromoteView(FormView):
template_name = 'lingo/invoicing/manager_pool_promote.html'
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
self.object = get_object_or_404(
Pool,
campaign__id=kwargs['pk'],
campaign__regie=self.regie,
campaign__invalid=False,
campaign__finalized=False,
pk=kwargs['pool_pk'],
draft=True,
status='completed',
)
if not self.object.is_last:
raise Http404
return super().dispatch(request, *args, **kwargs)
def get_context_data(self, **kwargs):
kwargs['form'] = None
kwargs['regie'] = self.regie
kwargs['object'] = self.object.campaign
kwargs['pool'] = self.object
return super().get_context_data(**kwargs)
def post(self, request, *args, **kwargs):
self.object.promote()
return redirect(
'%s#open:pools'
% reverse(
'lingo-manager-invoicing-campaign-detail', args=[self.regie.pk, self.object.campaign.pk]
)
)
pool_promote = PoolPromoteView.as_view()
class PoolDeleteView(DeleteView):
template_name = 'lingo/manager_confirm_delete.html'
model = Pool
pk_url_kwarg = 'pool_pk'
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
self.campaign = get_object_or_404(
Campaign.objects.filter(regie=self.regie, finalized=False).exclude(
pool__status__in=['registered', 'running']
),
pk=kwargs['pk'],
)
return super().dispatch(request, *args, **kwargs)
def get_queryset(self):
return self.campaign.pool_set.all()
def delete(self, request, *args, **kwargs):
self.object = self.get_object()
if self.object.is_last and self.object.draft:
self.campaign.mark_as_invalid()
invoice_model = Invoice
line_model = InvoiceLine
if self.object.draft:
invoice_model = DraftInvoice
line_model = DraftInvoiceLine
line_model.objects.filter(pool=self.object).delete()
invoice_model.objects.filter(pool=self.object).delete()
return super().delete(request, *args, **kwargs)
def get_success_url(self):
return '%s#open:pools' % reverse(
'lingo-manager-invoicing-campaign-detail', args=[self.regie.pk, self.campaign.pk]
)
pool_delete = PoolDeleteView.as_view()
class InvoicePDFView(PDFMixin, DetailView):
pk_url_kwarg = 'invoice_pk'
model = Invoice
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
self.pool = get_object_or_404(
Pool, pk=kwargs['pool_pk'], campaign=kwargs['pk'], campaign__regie=self.regie
)
return super().dispatch(request, *args, **kwargs)
def get_queryset(self):
invoice_model = Invoice
if self.pool.draft:
invoice_model = DraftInvoice
return invoice_model.objects.filter(pool=self.pool)
invoice_pdf = InvoicePDFView.as_view()
class InvoiceLineListView(ListView):
template_name = 'lingo/invoicing/manager_invoice_lines.html'
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
self.pool = get_object_or_404(
Pool, pk=kwargs['pool_pk'], campaign_id=kwargs['pk'], campaign__regie=self.regie
)
invoice_model = Invoice
if self.pool.draft:
invoice_model = DraftInvoice
self.invoice = get_object_or_404(invoice_model, pk=kwargs['invoice_pk'], pool=self.pool)
return super().dispatch(request, *args, **kwargs)
def get_queryset(self):
return self.invoice.lines.all().order_by('user_external_id', 'event_date', 'pk')
def get_context_data(self, **kwargs):
kwargs['regie'] = self.pool.campaign.regie
kwargs['object'] = self.pool.campaign
kwargs['pool'] = self.pool
if not self.pool.draft:
kwargs['invoice'] = self.invoice
kwargs['invoice_payments'] = (
InvoicePayment.objects.filter(invoice=self.invoice)
.select_related('payment')
.order_by('created_at')
)
return super().get_context_data(**kwargs)
invoice_line_list = InvoiceLineListView.as_view()
class LineSetErrorStatusView(DetailView):
model = InvoiceLine
pk_url_kwarg = 'line_pk'
template_name = 'lingo/invoicing/manager_line_detail_fragment.html'
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(
status='error',
pool=self.kwargs['pool_pk'],
pool__campaign=self.kwargs['pk'],
pool__campaign__regie=self.regie,
)
)
def get_context_data(self, **kwargs):
kwargs['regie'] = self.regie
kwargs['object'] = self.object.pool.campaign
kwargs['pool'] = self.object.pool
kwargs['line'] = self.object
return super().get_context_data(**kwargs)
def get(self, request, *args, **kwargs):
self.object = self.get_object()
error_status = kwargs['status']
if error_status == 'reset':
self.object.error_status = ''
elif error_status == 'ignore':
self.object.error_status = 'ignored'
elif error_status == 'fix':
self.object.error_status = 'fixed'
else:
raise Http404
self.object.save()
if is_ajax(self.request):
context = self.get_context_data(object=self.object)
return self.render_to_response(context)
return redirect(
reverse(
'lingo-manager-invoicing-pool-journal', args=[self.regie.pk, kwargs['pk'], kwargs['pool_pk']]
)
)
line_set_error_status = LineSetErrorStatusView.as_view()

View File

@ -0,0 +1,323 @@
# lingo - payment and billing system
# Copyright (C) 2022 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/>.
import collections
import datetime
import json
from django.contrib import messages
from django.db import transaction
from django.db.models import CharField, IntegerField, JSONField, Value
from django.http import HttpResponse
from django.shortcuts import get_object_or_404
from django.urls import reverse, reverse_lazy
from django.utils.translation import gettext_lazy as _
from django.utils.translation import ngettext
from django.views.generic import CreateView, DeleteView, DetailView, FormView, ListView, UpdateView
from lingo.agendas.models import Agenda
from lingo.invoicing.forms import RegieInvoiceFilterSet
from lingo.invoicing.models import (
Counter,
InjectedLine,
Invoice,
InvoiceLine,
InvoicePayment,
Pool,
Regie,
RegieImportError,
)
from lingo.invoicing.views.utils import PDFMixin
from lingo.pricing.forms import ImportForm
def import_regies(data):
results = collections.defaultdict(list)
with transaction.atomic():
regies = data.get('regies', [])
for regie in regies:
created, regie_obj = Regie.import_json(regie)
if created:
results['created'].append(regie_obj)
else:
results['updated'].append(regie_obj)
return results
class RegiesListView(ListView):
template_name = 'lingo/invoicing/manager_regie_list.html'
model = Regie
regies_list = RegiesListView.as_view()
class RegieAddView(CreateView):
template_name = 'lingo/invoicing/manager_regie_form.html'
model = Regie
fields = ['label', 'description', 'cashier_role']
def get_success_url(self):
return reverse('lingo-manager-invoicing-regie-detail', args=[self.object.pk])
regie_add = RegieAddView.as_view()
class RegieDetailView(DetailView):
template_name = 'lingo/invoicing/manager_regie_detail.html'
model = Regie
def get_context_data(self, **kwargs):
kwargs['regie'] = self.object
kwargs['agendas'] = Agenda.objects.filter(regie=self.object)
kwargs['campaigns'] = self.object.campaign_set.all().order_by('-date_start')
has_related_objects = False
if kwargs['campaigns']:
has_related_objects = True
elif self.object.injectedline_set.exists():
has_related_objects = True
kwargs['has_related_objects'] = has_related_objects
return super().get_context_data(**kwargs)
regie_detail = RegieDetailView.as_view()
class RegieEditView(UpdateView):
template_name = 'lingo/invoicing/manager_regie_form.html'
model = Regie
fields = ['label', 'description', 'cashier_role', 'counter_name', 'number_format']
def get_success_url(self):
return reverse('lingo-manager-invoicing-regie-detail', args=[self.object.pk])
regie_edit = RegieEditView.as_view()
class RegieDeleteView(DeleteView):
template_name = 'lingo/manager_confirm_delete.html'
model = Regie
def get_queryset(self):
return super().get_queryset().filter(campaign__isnull=True, injectedline__isnull=True)
def delete(self, request, *args, **kwargs):
self.object = self.get_object()
Counter.objects.filter(regie=self.object).delete()
return super().delete(request, *args, **kwargs)
def get_success_url(self):
return reverse('lingo-manager-invoicing-regie-list')
regie_delete = RegieDeleteView.as_view()
class RegiesExportView(ListView):
model = Regie
def get(self, request, *args, **kwargs):
response = HttpResponse(content_type='application/json')
today = datetime.date.today()
attachment = 'attachment; filename="export_regies_{}.json"'.format(today.strftime('%Y%m%d'))
response['Content-Disposition'] = attachment
json.dump({'regies': [regie.export_json() for regie in self.get_queryset()]}, response, indent=2)
return response
regies_export = RegiesExportView.as_view()
class RegiesImportView(FormView):
form_class = ImportForm
template_name = 'lingo/invoicing/manager_import.html'
success_url = reverse_lazy('lingo-manager-invoicing-regie-list')
def form_valid(self, form):
try:
config_json = json.loads(self.request.FILES['config_json'].read())
except ValueError:
form.add_error('config_json', _('File is not in the expected JSON format.'))
return self.form_invalid(form)
try:
results = import_regies(config_json)
except RegieImportError as exc:
form.add_error('config_json', '%s' % exc)
return self.form_invalid(form)
import_messages = {
'create': lambda x: ngettext(
'A regie was created.',
'%(count)d regies were created.',
x,
),
'update': lambda x: ngettext(
'A regie was updated.',
'%(count)d regie were updated.',
x,
),
}
create_message = _('No regie created.')
update_message = _('No regie updated.')
created = len(results.get('created', []))
updated = len(results.get('updated', []))
if created:
create_message = import_messages.get('create')(created) % {'count': created}
if updated:
update_message = import_messages.get('update')(updated) % {'count': updated}
message = "%s %s" % (create_message, update_message)
messages.info(self.request, message)
return super().form_valid(form)
regies_import = RegiesImportView.as_view()
class NonInvoicedLineListView(ListView):
template_name = 'lingo/invoicing/manager_non_invoiced_line_list.html'
paginate_by = 100
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):
fields = [
'pk',
'event_date',
'slug',
'label',
'quantity',
'unit_amount',
'total_amount',
'user_external_id',
'payer_external_id',
'payer_first_name',
'payer_last_name',
'user_first_name',
'user_last_name',
'event',
'pricing_data',
'status',
'pool_id',
]
qs1 = InvoiceLine.objects.filter(
status='error', error_status='', pool__campaign__regie=self.regie
).values(*fields)
qs2 = (
InjectedLine.objects.filter(invoiceline__isnull=True, regie=self.regie)
.annotate(
user_first_name=Value('', output_field=CharField()),
user_last_name=Value('', output_field=CharField()),
event=Value({}, output_field=JSONField()),
pricing_data=Value({}, output_field=JSONField()),
status=Value('injected', output_field=CharField()),
pool_id=Value(0, output_field=IntegerField()),
)
.values(*fields)
)
qs = qs1.union(qs2).order_by('event_date', 'user_external_id', 'label', 'pk')
return qs
def get_context_data(self, **kwargs):
kwargs['regie'] = self.regie
context = super().get_context_data(**kwargs)
pools = Pool.objects.filter(draft=False).in_bulk()
for line in context['object_list']:
if line['status'] == 'error':
line['user_name'] = InvoiceLine(
user_first_name=line['user_first_name'], user_last_name=line['user_last_name']
).user_name
line['payer_name'] = InvoiceLine(
payer_first_name=line['payer_first_name'], payer_last_name=line['payer_last_name']
).payer_name
line['error_display'] = InvoiceLine(
status=line['status'], pricing_data=line['pricing_data']
).get_error_display()
line['campaign_id'] = pools[line['pool_id']].campaign_id
line['chrono_event_url'] = InvoiceLine(event=line['event']).get_chrono_event_url()
return context
non_invoiced_line_list = NonInvoicedLineListView.as_view()
class RegieInvoiceListView(ListView):
template_name = 'lingo/invoicing/manager_invoice_list.html'
paginate_by = 100
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):
self.filterset = RegieInvoiceFilterSet(
data=self.request.GET or None,
queryset=Invoice.objects.filter(regie=self.regie).prefetch_related('pool').order_by('created_at'),
)
return self.filterset.qs
def get_context_data(self, **kwargs):
kwargs['regie'] = self.regie
kwargs['filterset'] = self.filterset
return super().get_context_data(**kwargs)
regie_invoice_list = RegieInvoiceListView.as_view()
class RegieInvoicePDFView(PDFMixin, DetailView):
pk_url_kwarg = 'invoice_pk'
model = Invoice
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)
regie_invoice_pdf = RegieInvoicePDFView.as_view()
class RegieInvoiceLineListView(ListView):
template_name = 'lingo/invoicing/manager_invoice_lines.html'
def dispatch(self, request, *args, **kwargs):
self.regie = get_object_or_404(Regie, pk=kwargs['regie_pk'])
self.invoice = get_object_or_404(Invoice, pk=kwargs['invoice_pk'], regie=self.regie)
return super().dispatch(request, *args, **kwargs)
def get_queryset(self):
return self.invoice.lines.all().order_by('user_external_id', 'event_date', 'pk')
def get_context_data(self, **kwargs):
kwargs['regie'] = self.regie
kwargs['invoice'] = self.invoice
kwargs['invoice_payments'] = (
InvoicePayment.objects.filter(invoice=self.invoice)
.select_related('payment')
.order_by('created_at')
)
return super().get_context_data(**kwargs)
regie_invoice_line_list = RegieInvoiceLineListView.as_view()

View File

@ -0,0 +1,31 @@
# 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.http import HttpResponse
from weasyprint import HTML
class PDFMixin:
def get(self, request, *args, **kwargs):
self.object = self.get_object()
result = self.object.html()
if 'html' in request.GET:
return HttpResponse(result)
html = HTML(string=result)
pdf = html.write_pdf()
response = HttpResponse(pdf, content_type='application/pdf')
response['Content-Disposition'] = 'attachment; filename="%s.pdf"' % self.object.formatted_number
return response

View File

@ -824,7 +824,7 @@ def test_add_pool(app, admin_user):
app.get('/manage/invoicing/regie/%s/campaign/%s/pool/add/' % (regie.pk, campaign.pk), status=404)
@mock.patch('lingo.invoicing.views.unlock_events_check')
@mock.patch('lingo.invoicing.views.campaign.unlock_events_check')
def test_unlock_check(mock_unlock, app, admin_user):
regie = Regie.objects.create(label='Foo')
agenda = Agenda.objects.create(label='Foo bar', regie=regie)
@ -919,7 +919,7 @@ def test_unlock_check(mock_unlock, app, admin_user):
app.get('/manage/invoicing/regie/%s/campaign/%s/unlock-check/' % (regie.pk, campaign.pk), status=404)
@mock.patch('lingo.invoicing.views.mark_events_invoiced')
@mock.patch('lingo.invoicing.views.campaign.mark_events_invoiced')
def test_finalize(mock_invoiced, app, admin_user):
regie = Regie.objects.create(label='Foo')
agenda = Agenda.objects.create(label='Foo bar', regie=regie)