entrouvert
/
lingo
15
0
Fork 0
Code Issues Pull Requests 7 Releases Wiki Activity

invoicing: draft invoice lines generation (#71528)

This commit is contained in:
Lauréline Guérin 2022-11-28 16:41:29 +01:00
parent 2b506e6269
commit b995170048
No known key found for this signature in database
GPG Key ID: 1FAB9B9B4F93D473
10 changed files with 1672 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
MANIFEST.in
View File

@ -10,6 +10,9 @@ recursive-include lingo/templates *.html *.txt
recursive-include lingo/manager/templates *.html *.txt
recursive-include lingo/pricing/templates *.html *.txt
# sql (migrations)
recursive-include lingo/invoicing/sql *.sql
include COPYING README
include MANIFEST.in
include VERSION

0
lingo/invoicing/management/__init__.py Normal file
View File

0
lingo/invoicing/management/commands/__init__.py Normal file
View File

45
lingo/invoicing/management/commands/generate_invoices.py Normal file
View File

@ -0,0 +1,45 @@
# 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 datetime
from django.core.management.base import BaseCommand, CommandError
from lingo.invoicing.utils import generate_invoices
class Command(BaseCommand):
help = 'Generate invoicing for a period'
def add_arguments(self, parser):
parser.add_argument('date_start')
parser.add_argument('date_end')
def handle(self, *args, **options):
try:
date_start = datetime.datetime.fromisoformat(options['date_start']).date()
except ValueError:
raise CommandError('Bad value "%s" for date_start' % options['date_start'])
try:
date_end = datetime.datetime.fromisoformat(options['date_end']).date()
except ValueError:
raise CommandError('Bad value "%s" for date_end' % options['date_end'])
generate_invoices(date_start=date_start, date_end=date_end)
self.stdout.write(
self.style.SUCCESS('Invoicing generation OK (start: %s, end: %s)' % (date_start, date_end))
)

27
lingo/invoicing/migrations/0003_invoice.py Normal file
View File

@ -0,0 +1,27 @@
import os
from django.db import migrations
with open(
os.path.join(
os.path.dirname(os.path.realpath(__file__)),
'..',
'sql',
'invoice_triggers_for_amount.sql',
)
) as sql_file:
sql_triggers = sql_file.read()
sql_forwards = sql_triggers
class Migration(migrations.Migration):
dependencies = [
('invoicing', '0002_invoice'),
]
operations = [
migrations.RunSQL(sql=sql_forwards, reverse_sql=migrations.RunSQL.noop),
]

4
lingo/invoicing/models.py
View File

@ -89,9 +89,7 @@ class Regie(models.Model):
class AbstractInvoice(models.Model):
label = models.CharField(_('Label'), max_length=300)
total_amount = models.DecimalField(
max_digits=9, decimal_places=2, default=0
) # XXX PG trigger to maintain this ?
total_amount = models.DecimalField(max_digits=9, decimal_places=2, default=0)
date_issue = models.DateField(_('Issue date'))
regie = models.ForeignKey(Regie, on_delete=models.PROTECT)
payer = models.CharField(_('Payer'), max_length=300)

44
lingo/invoicing/sql/invoice_triggers_for_amount.sql Normal file
View File

@ -0,0 +1,44 @@
CREATE OR REPLACE FUNCTION set_invoice_total_amount() RETURNS TRIGGER AS $$
DECLARE
invoice_ids integer[];
BEGIN
IF TG_OP = 'INSERT' THEN
invoice_ids := ARRAY[NEW.invoice_id];
ELSIF TG_OP = 'DELETE' THEN
invoice_ids := ARRAY[OLD.invoice_id];
ELSIF TG_OP = 'UPDATE' THEN
invoice_ids := ARRAY[NEW.invoice_id, OLD.invoice_id];
END IF;
EXECUTE 'UPDATE ' || substring(TG_TABLE_NAME for length(TG_TABLE_NAME) - 4) || ' i
SET total_amount = COALESCE(
(
SELECT SUM(l.total_amount)
FROM ' || TG_TABLE_NAME || ' l
WHERE l.invoice_id = i.id
AND l.status = ''success''
), 0
)
WHERE id = ANY($1);' USING invoice_ids;
IF TG_OP = 'DELETE' THEN
RETURN OLD;
ELSE
RETURN NEW;
END IF;
END;
$$ LANGUAGE plpgsql;
DROP TRIGGER IF EXISTS set_draftinvoice_total_amount_trg ON invoicing_draftinvoiceline;
CREATE TRIGGER set_draftinvoice_total_amount_trg
AFTER INSERT OR UPDATE OF total_amount, invoice_id, status OR DELETE ON invoicing_draftinvoiceline
FOR EACH ROW
EXECUTE PROCEDURE set_invoice_total_amount();
DROP TRIGGER IF EXISTS set_invoice_total_amount_trg ON invoicing_invoiceline;
CREATE TRIGGER set_invoice_total_amount_trg
AFTER INSERT OR UPDATE OF total_amount, invoice_id, status OR DELETE ON invoicing_invoiceline
FOR EACH ROW
EXECUTE PROCEDURE set_invoice_total_amount();

221
lingo/invoicing/utils.py Normal file
View File

@ -0,0 +1,221 @@
# 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 datetime
from django.test.client import RequestFactory
from django.utils.translation import ugettext_lazy as _
from lingo.agendas.chrono import get_check_status, get_subscriptions
from lingo.agendas.models import Agenda
from lingo.invoicing.models import DraftInvoice, DraftInvoiceLine
from lingo.pricing.models import AgendaPricing, AgendaPricingNotFound, PricingError
def get_agendas(date_start, date_end):
agendas_pricings = AgendaPricing.objects.filter(flat_fee_schedule=False).extra(
where=["(date_start, date_end) OVERLAPS (%s, %s)"], params=[date_start, date_end]
)
return Agenda.objects.filter(pk__in=agendas_pricings.values('agendas')).order_by('pk')
def get_all_subscriptions(agendas, date_start, date_end):
# result:
# {
# 'user_id': {
# 'agenda_slug': [sub1, sub2, ...],
# 'agenda_slug2': [sub1, sub2, ...],
# },
# ...
# }
all_subscriptions = {}
for agenda in agendas:
subscriptions = get_subscriptions(agenda_slug=agenda.slug, date_start=date_start, date_end=date_end)
for subscription in subscriptions:
if subscription['user_external_id'] not in all_subscriptions:
all_subscriptions[subscription['user_external_id']] = {}
user_subs = all_subscriptions[subscription['user_external_id']]
if agenda.slug not in user_subs:
user_subs[agenda.slug] = []
user_subs[agenda.slug].append(subscription)
return all_subscriptions
def get_invoice_lines_for_user(agendas, user_external_id, subscriptions, date_start, date_end):
def get_subscription(subscriptions, event_date):
# get subscription matching event_date
for subscription in subscriptions:
sub_start_date = datetime.date.fromisoformat(subscription['date_start'])
sub_end_date = datetime.date.fromisoformat(subscription['date_end'])
if sub_start_date > event_date:
continue
if sub_end_date <= event_date:
continue
return subscription
if not agendas or not subscriptions:
return []
# get check status for user_external_id, on agendas, for the period
check_status_list = get_check_status(
agenda_slugs=list(subscriptions.keys()),
user_external_id=user_external_id,
date_start=date_start,
date_end=date_end,
)
request = RequestFactory().get('/') # XXX
agendas_by_slug = {a.slug: a for a in agendas}
# build lines from check status
lines = []
for check_status in check_status_list:
serialized_event = check_status['event']
event_date = datetime.datetime.fromisoformat(serialized_event['start_datetime']).date()
event_slug = '%s@%s' % (serialized_event['agenda'], serialized_event['slug'])
if serialized_event['agenda'] not in subscriptions:
# should not happen, check-status endpoint is based on real subscriptions
continue
serialized_subscription = get_subscription(
subscriptions=subscriptions[serialized_event['agenda']],
event_date=event_date,
)
if not serialized_subscription:
# should not happen, check-status endpoint is based on real subscriptions
continue
agenda = agendas_by_slug[serialized_event['agenda']]
try:
agenda_pricing = AgendaPricing.get_agenda_pricing(
agenda=agenda, start_date=event_date, flat_fee_schedule=False
) # XXX optimize SQL request
pricing_data = agenda_pricing.get_pricing_data_for_event(
request=request,
agenda=agenda,
event=serialized_event,
subscription=serialized_subscription,
check_status=check_status['check_status'],
booking=check_status['booking'],
user_external_id=user_external_id,
adult_external_id=user_external_id, # XXX
)
except AgendaPricingNotFound:
# no pricing, no invoice
# can happen if pricing model defined only on a part of the requested period
# XXX error, warning, or ignore ?
continue
except PricingError as e:
# XXX explicit each error
# XXX and log context ?
pricing_error = {'error': e.details}
lines.append(
DraftInvoiceLine(
slug=event_slug,
label=serialized_event['label'],
quantity=0,
unit_amount=0,
total_amount=0,
user_external_id=user_external_id,
payer_external_id=user_external_id, # XXX
event=serialized_event,
pricing_data=pricing_error,
status='error',
)
)
else:
# XXX log all context !
lines.append(
DraftInvoiceLine(
slug=event_slug,
label=serialized_event['label'],
quantity=1,
unit_amount=pricing_data['pricing'],
total_amount=pricing_data['pricing'],
user_external_id=user_external_id,
payer_external_id=user_external_id, # XXX
event=serialized_event,
pricing_data=pricing_data,
status='success',
)
)
DraftInvoiceLine.objects.bulk_create(lines)
return lines
def get_all_invoice_lines(agendas, subscriptions, date_start, date_end):
lines = []
for user_external_id, user_subs in subscriptions.items():
# generate lines for each user
lines += get_invoice_lines_for_user(
agendas=agendas,
user_external_id=user_external_id,
subscriptions=user_subs,
date_start=date_start,
date_end=date_end,
)
return lines
def generate_invoices_from_lines(agendas, all_lines, date_start, date_end):
agendas_by_slug = {a.slug: a for a in agendas}
# regroup lines by regie, and by payer_external_id (payer)
lines_by_regie = {}
for line in all_lines:
if line.status != 'success':
# ignore lines in error
continue
agenda_slug = line.event['agenda']
if agenda_slug not in agendas_by_slug:
# should not happen
continue
regie = agendas_by_slug[agenda_slug].regie
if not regie:
# XXX what should we do if regie is not configured ?
continue
if regie.pk not in lines_by_regie:
lines_by_regie[regie.pk] = {}
regie_subs = lines_by_regie[regie.pk]
if line.payer_external_id not in regie_subs:
regie_subs[line.payer_external_id] = []
regie_subs[line.payer_external_id].append(line)
# generate invoices by regie and by payer_external_id (payer)
invoices = []
for regie_id, regie_subs in lines_by_regie.items():
for payer_external_id, adult_lines in regie_subs.items():
invoice = DraftInvoice.objects.create(
label=_('Invoice from %s to %s') % (date_start, date_end - datetime.timedelta(days=1)),
date_issue=date_end, # XXX
regie_id=regie_id,
payer=payer_external_id,
)
DraftInvoiceLine.objects.filter(pk__in=[line.pk for line in adult_lines]).update(invoice=invoice)
invoices.append(invoice)
return invoices
def generate_invoices(date_start, date_end):
# get agendas with pricing corresponding to the period
agendas = get_agendas(date_start=date_start, date_end=date_end)
# get subscriptions for each agenda, for the period
subscriptions = get_all_subscriptions(agendas=agendas, date_start=date_start, date_end=date_end)
# get invoice lines for all subscribed users, for each agenda in the corresponding period
lines = get_all_invoice_lines(
agendas=agendas, subscriptions=subscriptions, date_start=date_start, date_end=date_end
)
# and generate invoices
generate_invoices_from_lines(agendas=agendas, all_lines=lines, date_start=date_start, date_end=date_end)

1195
tests/invoicing/test_invoice_generation.py Normal file
View File

File diff suppressed because it is too large Load Diff

136
tests/invoicing/test_models.py Normal file
View File

@ -0,0 +1,136 @@
import datetime
import pytest
from lingo.invoicing.models import DraftInvoice, DraftInvoiceLine, Invoice, InvoiceLine, Regie
pytestmark = pytest.mark.django_db
@pytest.mark.parametrize('draft', [True, False])
def test_invoice_total_amount(draft):
regie = Regie.objects.create()
invoice_model = DraftInvoice if draft else Invoice
line_model = DraftInvoiceLine if draft else InvoiceLine
invoice = invoice_model.objects.create(date_issue=datetime.date.today(), regie=regie)
assert invoice.total_amount == 0
invoice2 = invoice_model.objects.create(date_issue=datetime.date.today(), regie=regie)
assert invoice2.total_amount == 0
# line with error status, ignored
line = line_model.objects.create(
invoice=invoice, # with invoice
quantity=0,
unit_amount=0,
total_amount=0,
status='error',
)
invoice.refresh_from_db()
assert invoice.total_amount == 0
invoice2.refresh_from_db()
assert invoice2.total_amount == 0
# update line
line.unit_amount = 10
line.quantity = 1
line.total_amount = 10
line.save()
invoice.refresh_from_db()
# still in error
assert invoice.total_amount == 0
invoice2.refresh_from_db()
assert invoice2.total_amount == 0
# update line status
line.status = 'success'
line.save()
invoice.refresh_from_db()
assert invoice.total_amount == 10
invoice2.refresh_from_db()
assert invoice2.total_amount == 0
# update other field
line.unit_amount = 12 # total amount is wrong
line.save()
invoice.refresh_from_db()
assert invoice.total_amount == 10
invoice2.refresh_from_db()
assert invoice2.total_amount == 0
# update total_amount
line.total_amount = 12
line.save()
invoice.refresh_from_db()
assert invoice.total_amount == 12
invoice2.refresh_from_db()
assert invoice2.total_amount == 0
# create line with invoice, status success
line2 = line_model.objects.create(
invoice=invoice,
quantity=1,
unit_amount=20,
total_amount=20,
status='success',
)
invoice.refresh_from_db()
assert invoice.total_amount == 32
invoice2.refresh_from_db()
assert invoice2.total_amount == 0
# change invoice
line2.invoice = invoice2
line2.save()
invoice.refresh_from_db()
assert invoice.total_amount == 12
invoice2.refresh_from_db()
assert invoice2.total_amount == 20
# delete line
line2.delete()
invoice.refresh_from_db()
assert invoice.total_amount == 12
invoice2.refresh_from_db()
assert invoice2.total_amount == 0
# create line without invoice, status success
line3 = line_model.objects.create(
quantity=1,
unit_amount=20,
total_amount=20,
status='success',
)
invoice.refresh_from_db()
assert invoice.total_amount == 12
invoice2.refresh_from_db()
assert invoice2.total_amount == 0
# set invoice
line3.invoice = invoice
line3.save()
invoice.refresh_from_db()
assert invoice.total_amount == 32
invoice2.refresh_from_db()
assert invoice2.total_amount == 0
# reset invoice
line3.invoice = None
line3.save()
invoice.refresh_from_db()
assert invoice.total_amount == 12
invoice2.refresh_from_db()
assert invoice2.total_amount == 0
# no changes
line3.save()
invoice.refresh_from_db()
assert invoice.total_amount == 12
invoice2.refresh_from_db()
assert invoice2.total_amount == 0
# delete line
line3.delete()
invoice.refresh_from_db()
assert invoice.total_amount == 12
invoice2.refresh_from_db()
assert invoice2.total_amount == 0