Enuxia-Public/enuxia_einvoice
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: build EN16931 invoice payloads

Browse Source
This commit is contained in:
Julien Denizot
2026-06-19 10:54:47 +00:00
parent 73df50d558
commit 7fb5c88eeb
7 changed files with 1242 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
enuxia_einvoice/builders/__init__.py
View File
enuxia_einvoice/builders/en16931.py
+486
View File
@@ -0,0 +1,486 @@
# Copyright (c) 2026, Enuxia and contributors
# For license information, please see license.txt
from __future__ import annotations
import html
import json
import re
from collections import defaultdict
from decimal import Decimal, ROUND_HALF_UP
from typing import Any
class EN16931BuildError(ValueError):
"""La facture ERPNext ne peut pas être convertie proprement."""
UOM_CODES = {
"nos": "C62",
"unit": "C62",
"unité": "C62",
"piece": "C62",
"pièce": "C62",
"litre": "LTR",
"liter": "LTR",
"l": "LTR",
"kg": "KGM",
"kilogram": "KGM",
"kilogramme": "KGM",
"gram": "GRM",
"gramme": "GRM",
"hour": "HUR",
"heure": "HUR",
"day": "DAY",
"jour": "DAY",
"meter": "MTR",
"metre": "MTR",
"mètre": "MTR",
}
def _get(
document: Any,
fieldname: str,
default: Any = None,
) -> Any:
if document is None:
return default
if hasattr(document, "get"):
value = document.get(fieldname)
elif isinstance(document, dict):
value = document.get(fieldname)
else:
value = getattr(document, fieldname, default)
return default if value is None else value
def _decimal(value: Any) -> Decimal:
return Decimal(str(value or 0))
def _money(value: Any) -> str:
amount = _decimal(value).quantize(
Decimal("0.01"),
rounding=ROUND_HALF_UP,
)
return format(amount, ".2f")
def _quantity(value: Any) -> str:
quantity = _decimal(value)
text = format(quantity.normalize(), "f")
return text if "." in text else f"{text}.0"
def _plain_text(value: Any) -> str:
text = str(value or "")
text = re.sub(
r"<[^>]+>",
" ",
text,
)
text = html.unescape(text)
return " ".join(text.split())
def _uom_code(value: Any) -> str:
uom = str(value or "").strip()
code = UOM_CODES.get(
uom.casefold()
)
if not code:
raise EN16931BuildError(
f"L’unité ERPNext « {uom} » "
"n’a pas encore de code UN/ECE configuré."
)
return code
def _load_item_tax_rates(
raw_value: Any,
) -> list[Decimal]:
if not raw_value:
return []
if isinstance(raw_value, str):
try:
raw_value = json.loads(raw_value)
except json.JSONDecodeError as exc:
raise EN16931BuildError(
"Le détail de TVA d’une ligne "
"contient un JSON invalide."
) from exc
if not isinstance(raw_value, dict):
return []
rates: set[Decimal] = set()
for raw_rate in raw_value.values():
rate = _decimal(raw_rate)
if rate >= 0:
rates.add(rate)
return sorted(rates)
def _get_item_tax_rate(
item: Any,
*,
default_rate: Decimal | None,
) -> Decimal:
rates = _load_item_tax_rates(
_get(item, "item_tax_rate")
)
if len(rates) > 1:
raise EN16931BuildError(
"Une ligne utilise plusieurs taux de TVA. "
"Ce cas n’est pas encore pris en charge."
)
if rates:
return rates[0]
if default_rate is not None:
return default_rate
raise EN16931BuildError(
"Impossible de déterminer le taux de TVA "
"d’une ligne de facture."
)
def build_en16931_invoice(
invoice: Any,
*,
seller: dict[str, Any],
buyer: dict[str, Any],
) -> dict[str, Any]:
"""Construire une facture EN16931 pour les cas simples V1."""
if int(_get(invoice, "docstatus", 0)) != 1:
raise EN16931BuildError(
"La facture doit être soumise."
)
if _get(invoice, "is_return", 0):
raise EN16931BuildError(
"Les avoirs ne sont pas encore pris en charge."
)
if _decimal(
_get(invoice, "discount_amount")
) != 0:
raise EN16931BuildError(
"Les remises globales ne sont pas encore prises en charge."
)
raw_taxes = _get(invoice, "taxes", []) or []
taxes_by_rate: dict[Decimal, Decimal] = defaultdict(
lambda: Decimal("0")
)
for tax in raw_taxes:
tax_amount = _decimal(
_get(tax, "tax_amount")
)
if tax_amount == 0:
continue
if _get(tax, "charge_type") != "On Net Total":
raise EN16931BuildError(
"Seules les taxes de type "
"« On Net Total » sont prises en charge."
)
if int(
_get(tax, "included_in_print_rate", 0)
):
raise EN16931BuildError(
"La TVA incluse dans le prix n’est "
"pas encore prise en charge."
)
rate = _decimal(
_get(tax, "rate")
)
if rate < 0:
raise EN16931BuildError(
"Un taux de taxe négatif a été détecté."
)
taxes_by_rate[rate] += tax_amount
default_tax_rate = None
if len(taxes_by_rate) == 1:
default_tax_rate = next(
iter(taxes_by_rate)
)
raw_items = _get(invoice, "items", []) or []
if not raw_items:
raise EN16931BuildError(
"La facture ne contient aucune ligne."
)
lines: list[dict[str, Any]] = []
taxable_by_rate: dict[Decimal, Decimal] = defaultdict(
lambda: Decimal("0")
)
invoice_lines_total = Decimal("0")
for position, item in enumerate(
raw_items,
start=1,
):
quantity = _decimal(
_get(item, "qty")
)
net_amount = _decimal(
_get(item, "net_amount")
)
net_rate = _decimal(
_get(item, "net_rate")
)
if quantity <= 0:
raise EN16931BuildError(
f"La quantité de la ligne {position} "
"doit être supérieure à zéro."
)
if net_amount < 0:
raise EN16931BuildError(
f"Le montant de la ligne {position} "
"ne peut pas être négatif."
)
tax_rate = _get_item_tax_rate(
item,
default_rate=default_tax_rate,
)
item_name = (
_plain_text(
_get(item, "description")
)
or _plain_text(
_get(item, "item_name")
)
or str(
_get(item, "item_code") or ""
)
)
if not item_name:
raise EN16931BuildError(
f"La ligne {position} n’a pas de désignation."
)
uom = (
_get(item, "uom")
or _get(item, "stock_uom")
)
lines.append(
{
"identifier": f"{position:03d}",
"invoiced_quantity": _quantity(
quantity
),
"invoiced_quantity_code": _uom_code(
uom
),
"item_information": {
"name": item_name,
},
"net_amount": _money(
net_amount
),
"price_details": {
"item_net_price": _money(
net_rate
),
},
"vat_information": {
"invoiced_item_vat_category_code": "S",
"invoiced_item_vat_rate": _money(
tax_rate
),
},
}
)
invoice_lines_total += net_amount
taxable_by_rate[tax_rate] += net_amount
net_total = _decimal(
_get(invoice, "net_total")
)
if abs(invoice_lines_total - net_total) > Decimal("0.01"):
raise EN16931BuildError(
"La somme des lignes ne correspond pas "
"au total HT ERPNext."
)
vat_break_down: list[dict[str, str]] = []
calculated_vat_total = Decimal("0")
for tax_rate in sorted(taxable_by_rate):
taxable_amount = taxable_by_rate[tax_rate]
tax_amount = taxes_by_rate.get(
tax_rate
)
if tax_amount is None:
tax_amount = (
taxable_amount
* tax_rate
/ Decimal("100")
).quantize(
Decimal("0.01"),
rounding=ROUND_HALF_UP,
)
calculated_vat_total += tax_amount
vat_break_down.append(
{
"vat_category_code": "S",
"vat_category_rate": _money(
tax_rate
),
"vat_category_tax_amount": _money(
tax_amount
),
"vat_category_taxable_amount": _money(
taxable_amount
),
"vat_identifier": "VAT",
}
)
erpnext_vat_total = _decimal(
_get(
invoice,
"total_taxes_and_charges",
)
)
if (
abs(
calculated_vat_total
- erpnext_vat_total
)
> Decimal("0.01")
):
raise EN16931BuildError(
"La ventilation de TVA ne correspond pas "
"au total des taxes ERPNext."
)
grand_total = _decimal(
_get(invoice, "grand_total")
)
rounded_total = _decimal(
_get(invoice, "rounded_total")
)
if rounded_total == 0:
rounded_total = grand_total
rounding_amount = (
rounded_total - grand_total
)
total_advance = _decimal(
_get(invoice, "total_advance")
)
amount_due = (
rounded_total - total_advance
)
totals: dict[str, Any] = {
"amount_due_for_payment": _money(
amount_due
),
"sum_invoice_lines_amount": _money(
invoice_lines_total
),
"total_vat_amount": {
"currency_code": str(
_get(invoice, "currency")
),
"value": _money(
calculated_vat_total
),
},
"total_with_vat": _money(
grand_total
),
"total_without_vat": _money(
net_total
),
}
if rounding_amount != 0:
totals["rounding_amount"] = _money(
rounding_amount
)
if total_advance != 0:
totals["paid_amount"] = _money(
total_advance
)
return {
"number": str(
_get(invoice, "name")
),
"type_code": 380,
"issue_date": str(
_get(invoice, "posting_date")
),
"payment_due_date": str(
_get(invoice, "due_date")
),
"currency_code": str(
_get(invoice, "currency")
),
"process_control": {
"business_process_type": "M1",
"specification_identifier": (
"urn:cen.eu:en16931:2017"
),
},
"seller": seller,
"buyer": buyer,
"lines": lines,
"totals": totals,
"vat_break_down": vat_break_down,
}
enuxia_einvoice/payload.py
+348
View File
@@ -0,0 +1,348 @@
# Copyright (c) 2026, Enuxia and contributors
# For license information, please see license.txt
from __future__ import annotations
from typing import Any
import frappe
from frappe import _
from enuxia_einvoice.builders.en16931 import (
EN16931BuildError,
build_en16931_invoice,
)
from enuxia_einvoice.directory import (
split_directory_identifier,
)
from enuxia_einvoice.providers.superpdp.client import (
SuperPDPClient,
)
def _country_code(country: str | None) -> str:
country = str(country or "").strip()
if not country:
return ""
code = frappe.db.get_value(
"Country",
country,
"code",
)
return str(
code or country
).upper()
def _get_address(
address_name: str | None,
) -> dict[str, str]:
if not address_name:
frappe.throw(
_("Une adresse obligatoire est manquante.")
)
address = frappe.get_doc(
"Address",
address_name,
)
result = {
"address_line1": str(
address.address_line1 or ""
),
"address_line2": str(
address.address_line2 or ""
),
"address_line3": "",
"city": str(
address.city or ""
),
"post_code": str(
address.pincode or ""
),
"country_code": _country_code(
address.country
),
"country_subdivision": str(
address.state or ""
),
}
return {
key: value
for key, value in result.items()
if value
}
def _get_account(company: str):
if not frappe.db.exists(
"Electronic Invoicing Account",
company,
):
frappe.throw(
_(
"Aucun compte de facturation électronique "
"n’est configuré pour cette société."
)
)
account = frappe.get_doc(
"Electronic Invoicing Account",
company,
)
if not account.enabled:
frappe.throw(
_("Le compte SuperPDP est désactivé.")
)
return account
def _get_superpdp_sender(
account,
) -> tuple[dict[str, Any], dict[str, str]]:
client_id = str(
account.client_id or ""
).strip()
client_secret = account.get_password(
"client_secret",
raise_exception=False,
)
if not client_id or not client_secret:
frappe.throw(
_(
"Les identifiants OAuth SuperPDP "
"sont incomplets."
)
)
client = SuperPDPClient(
client_id=client_id,
client_secret=client_secret,
)
token = client.authenticate_client_credentials()
company = client.get_current_company(
token.access_token
)
entries_payload = client.list_directory_entries(
token.access_token
)
entries = entries_payload.get("data") or []
usable_entries = [
entry
for entry in entries
if (
isinstance(entry, dict)
and entry.get("status") == "created"
and entry.get("is_replyto") is not True
and entry.get("identifier")
)
]
if not usable_entries:
frappe.throw(
_(
"Aucune adresse électronique active "
"n’est disponible pour le vendeur."
)
)
if len(usable_entries) > 1:
frappe.throw(
_(
"Plusieurs adresses vendeur sont disponibles. "
"La sélection explicite sera nécessaire."
)
)
electronic_address = split_directory_identifier(
usable_entries[0]["identifier"]
)
return company, electronic_address
@frappe.whitelist()
def generate_sales_invoice_en16931(
invoice_name: str,
) -> dict[str, Any]:
"""Générer le JSON EN16931 sans créer ni envoyer de facture."""
invoice = frappe.get_doc(
"Sales Invoice",
invoice_name,
)
invoice.check_permission("read")
if invoice.docstatus != 1:
frappe.throw(
_("La facture doit être soumise.")
)
customer = frappe.get_doc(
"Customer",
invoice.customer,
)
erpnext_company = frappe.get_doc(
"Company",
invoice.company,
)
account = _get_account(
invoice.company
)
remote_company, sender_address = (
_get_superpdp_sender(account)
)
expected_environment = (
"sandbox"
if account.environment == "Sandbox"
else "production"
)
if remote_company.get("env") != expected_environment:
frappe.throw(
_(
"L’environnement du compte ERPNext "
"ne correspond pas au compte SuperPDP."
)
)
seller_vat = str(
invoice.company_tax_id
or erpnext_company.tax_id
or ""
).strip()
buyer_vat = str(
invoice.tax_id
or customer.tax_id
or ""
).strip()
if not seller_vat:
frappe.throw(
_(
"Le numéro de TVA de la société "
"émettrice est manquant."
)
)
if not buyer_vat:
frappe.throw(
_(
"Le numéro de TVA du client est manquant."
)
)
buyer_scheme = str(
customer.einvoice_routing_scheme
or ""
).strip()
buyer_address = str(
customer.einvoice_routing_address
or ""
).strip()
buyer_number = str(
customer.einvoice_siren
or ""
).strip()
if (
not buyer_scheme
or not buyer_address
or not buyer_number
):
frappe.throw(
_(
"Le routage électronique du client "
"est incomplet."
)
)
seller = {
"electronic_address": {
"scheme": sender_address["scheme"],
"value": sender_address["value"],
},
"legal_registration_identifier": {
"scheme": "0002",
"value": str(
remote_company.get("number") or ""
),
},
"name": str(
remote_company.get("formal_name") or ""
),
"postal_address": {
"address_line1": str(
remote_company.get("address") or ""
),
"city": str(
remote_company.get("city") or ""
),
"post_code": str(
remote_company.get("postcode") or ""
),
"country_code": str(
remote_company.get("country") or ""
),
},
"vat_identifier": seller_vat,
}
buyer = {
"electronic_address": {
"scheme": buyer_scheme,
"value": buyer_address,
},
"identifiers": [
{
"scheme": buyer_scheme,
"value": buyer_number,
}
],
"legal_registration_identifier": {
"scheme": "0002",
"value": buyer_number,
},
"name": str(
invoice.customer_name
or customer.customer_name
or customer.name
),
"postal_address": _get_address(
invoice.customer_address
),
"vat_identifier": buyer_vat,
}
try:
return build_en16931_invoice(
invoice.as_dict(),
seller=seller,
buyer=buyer,
)
except EN16931BuildError as exc:
frappe.throw(
str(exc),
title=_("Génération EN16931 impossible"),
)
enuxia_einvoice/providers/superpdp/client.py
+38
View File
@@ -174,6 +174,44 @@ class SuperPDPClient:
return self._read_response(response)
def list_directory_entries(
self,
access_token: str,
) -> dict[str, Any]:
"""Lister les adresses de la société associée au token."""
response = self.session.get(
f"{self.base_url}/v1.beta/directory_entries",
headers=self._authorization_headers(
access_token
),
timeout=self.timeout,
)
return self._read_response(response)
def generate_test_en16931_invoice(
self,
access_token: str,
) -> dict[str, Any]:
"""Récupérer un exemple EN16931 valide en Sandbox."""
response = self.session.get(
(
f"{self.base_url}/v1.beta/invoices/"
"generate_test_invoice"
),
headers=self._authorization_headers(
access_token
),
params={
"format": "en16931",
},
timeout=self.timeout,
)
return self._read_response(response)
@staticmethod
def _authorization_headers(
access_token: str,
enuxia_einvoice/providers/superpdp/test_client.py
+87
View File
@@ -227,5 +227,92 @@ class TestSuperPDPClient(unittest.TestCase):
)
def test_generate_test_en16931_invoice(self) -> None:
response = Mock()
response.ok = True
response.status_code = 200
response.json.return_value = {
"number": "TEST-2026-001",
"type_code": 380,
"currency_code": "EUR",
}
self.session.get.return_value = response
result = (
self.client.generate_test_en16931_invoice(
"secret-access-token"
)
)
self.assertEqual(
result["number"],
"TEST-2026-001",
)
self.session.get.assert_called_once_with(
(
"https://api.superpdp.tech/"
"v1.beta/invoices/"
"generate_test_invoice"
),
headers={
"Authorization": "Bearer secret-access-token",
},
params={
"format": "en16931",
},
timeout=(5, 30),
)
def test_list_directory_entries(self) -> None:
response = Mock()
response.ok = True
response.status_code = 200
response.json.return_value = {
"data": [
{
"id": 4919,
"directory": "peppol",
"identifier": "0225:315143296_4920",
"status": "created",
"is_replyto": False,
},
{
"id": 4920,
"directory": "peppol",
"identifier": (
"0225:315143296_4920_replyto"
),
"status": "created",
"is_replyto": True,
},
]
}
self.session.get.return_value = response
result = self.client.list_directory_entries(
"secret-access-token"
)
self.assertEqual(len(result["data"]), 2)
self.assertFalse(
result["data"][0]["is_replyto"]
)
self.session.get.assert_called_once_with(
(
"https://api.superpdp.tech/"
"v1.beta/directory_entries"
),
headers={
"Authorization": "Bearer secret-access-token",
},
timeout=(5, 30),
)
if __name__ == "__main__":
unittest.main()
enuxia_einvoice/sandbox.py
+148
View File
@@ -0,0 +1,148 @@
# Copyright (c) 2026, Enuxia and contributors
# For license information, please see license.txt
from __future__ import annotations
from typing import Any
import frappe
from frappe import _
from enuxia_einvoice.providers.superpdp.client import (
SuperPDPClient,
)
def _get_sandbox_account():
accounts = frappe.get_all(
"Electronic Invoicing Account",
filters={
"enabled": 1,
"provider": "SuperPDP",
"environment": "Sandbox",
},
pluck="name",
limit=2,
)
if not accounts:
frappe.throw(
_("Aucun compte SuperPDP Sandbox actif.")
)
if len(accounts) > 1:
frappe.throw(
_(
"Plusieurs comptes SuperPDP Sandbox "
"sont actifs."
)
)
return frappe.get_doc(
"Electronic Invoicing Account",
accounts[0],
)
@frappe.whitelist()
def get_superpdp_test_invoice() -> dict[str, Any]:
"""Obtenir un exemple EN16931 sans envoyer de facture."""
account = _get_sandbox_account()
if (
account.authentication_flow
!= "Client Credentials"
):
frappe.throw(
_(
"Cette opération nécessite "
"Client Credentials."
)
)
client_id = str(
account.client_id or ""
).strip()
client_secret = account.get_password(
"client_secret",
raise_exception=False,
)
if not client_id or not client_secret:
frappe.throw(
_(
"Les identifiants OAuth SuperPDP "
"sont incomplets."
)
)
client = SuperPDPClient(
client_id=client_id,
client_secret=client_secret,
)
token = client.authenticate_client_credentials()
return client.generate_test_en16931_invoice(
token.access_token
)
@frappe.whitelist()
def get_superpdp_sender_context() -> dict[str, Any]:
"""Afficher l’identité et les adresses du vendeur Sandbox."""
account = _get_sandbox_account()
client_id = str(
account.client_id or ""
).strip()
client_secret = account.get_password(
"client_secret",
raise_exception=False,
)
if not client_id or not client_secret:
frappe.throw(
_(
"Les identifiants OAuth SuperPDP "
"sont incomplets."
)
)
client = SuperPDPClient(
client_id=client_id,
client_secret=client_secret,
)
token = client.authenticate_client_credentials()
company = client.get_current_company(
token.access_token
)
entries_payload = client.list_directory_entries(
token.access_token
)
entries = entries_payload.get("data") or []
usable_entries = [
entry
for entry in entries
if (
isinstance(entry, dict)
and entry.get("status") == "created"
and entry.get("is_replyto") is not True
and entry.get("identifier")
)
]
return {
"company": company,
"usable_entries": usable_entries,
"all_entries": entries,
}
enuxia_einvoice/tests/test_en16931.py
+135
View File
@@ -0,0 +1,135 @@
from __future__ import annotations
import unittest
from enuxia_einvoice.builders.en16931 import (
EN16931BuildError,
build_en16931_invoice,
)
class TestEN16931Builder(unittest.TestCase):
def invoice(self):
return {
"name": "ACC-SINV-2026-00001",
"docstatus": 1,
"is_return": 0,
"posting_date": "2026-06-19",
"due_date": "2026-06-24",
"currency": "EUR",
"net_total": 50,
"total_taxes_and_charges": 10,
"grand_total": 60,
"rounded_total": 60,
"total_advance": 0,
"discount_amount": 0,
"items": [
{
"item_code": "VODKA",
"item_name": "VODKA",
"description": "",
"qty": 1,
"uom": "Litre",
"net_rate": 50,
"net_amount": 50,
"item_tax_rate": (
'{"445720 - TVA 20%": 20}'
),
}
],
"taxes": [
{
"charge_type": "On Net Total",
"rate": 20,
"tax_amount": 10,
"included_in_print_rate": 0,
}
],
}
def seller(self):
return {
"name": "Burger Queen",
"electronic_address": {
"scheme": "0225",
"value": "315143296_4920",
},
"legal_registration_identifier": {
"scheme": "0002",
"value": "000000002",
},
"postal_address": {
"country_code": "FR",
},
"vat_identifier": "FR18000000002",
}
def buyer(self):
return {
"name": "Tricatel",
"electronic_address": {
"scheme": "0225",
"value": "315143296_4919",
},
"legal_registration_identifier": {
"scheme": "0002",
"value": "000000001",
},
"postal_address": {
"country_code": "FR",
},
"vat_identifier": "FR15000000001",
}
def test_build_current_invoice(self):
result = build_en16931_invoice(
self.invoice(),
seller=self.seller(),
buyer=self.buyer(),
)
self.assertEqual(
result["number"],
"ACC-SINV-2026-00001",
)
self.assertEqual(
result["lines"][0][
"invoiced_quantity_code"
],
"LTR",
)
self.assertEqual(
result["totals"][
"total_without_vat"
],
"50.00",
)
self.assertEqual(
result["totals"][
"total_vat_amount"
]["value"],
"10.00",
)
self.assertEqual(
result["totals"][
"total_with_vat"
],
"60.00",
)
def test_unknown_uom_is_rejected(self):
invoice = self.invoice()
invoice["items"][0]["uom"] = "Palette spéciale"
with self.assertRaises(
EN16931BuildError
):
build_en16931_invoice(
invoice,
seller=self.seller(),
buyer=self.buyer(),
)
if __name__ == "__main__":
unittest.main()