feat: support sandbox Peppol customer routing

This commit is contained in:
Julien Denizot
2026-06-19 10:17:30 +00:00
parent 4ded977452
commit 35409ca935
6 changed files with 991 additions and 46 deletions
+113 -27
View File
@@ -29,15 +29,25 @@ ERR_SIRET_SIREN_MISMATCH = (
"Le SIRET ne correspond pas au SIREN renseigné. "
"Ses 9 premiers chiffres doivent être identiques au SIREN."
)
ERR_SANDBOX_NUMBER = (
"L'identifiant Sandbox doit contenir entre 1 et 64 caractères "
"alphanumériques. Les caractères point, tiret et underscore "
"sont également autorisés."
)
ERR_NUMBER_SCHEME = (
"Le type d'identifiant entreprise n'est pas pris en charge."
)
ERR_PEPPOL_IDENTIFIER = (
"L'adresse Peppol doit respecter le format "
"SCHÉMA:IDENTIFIANT, par exemple 0225:315143296_4919."
)
class FrenchIdentifierError(ValueError):
"""Erreur de validation d'un identifiant français."""
"""Erreur de validation d'un identifiant d'entreprise."""
def normalize_registration_number(value: Any) -> str:
"""Retirer les séparateurs usuels sans accepter d'autres caractères."""
normalized = re.sub(
r"[\s.\-]",
"",
@@ -51,8 +61,6 @@ def normalize_registration_number(value: Any) -> str:
def is_luhn_valid(value: str) -> bool:
"""Valider un numéro avec l'algorithme de Luhn."""
if not value or not value.isdigit():
return False
@@ -74,8 +82,6 @@ def is_luhn_valid(value: str) -> bool:
def validate_siren(value: Any) -> str:
"""Normaliser et valider un SIREN français."""
siren = normalize_registration_number(value)
if len(siren) != 9:
@@ -92,8 +98,6 @@ def validate_siret(
*,
expected_siren: str | None = None,
) -> str:
"""Normaliser et valider un SIRET français."""
siret = normalize_registration_number(value)
if len(siret) != 14:
@@ -110,12 +114,63 @@ def validate_siret(
return siret
def validate_sandbox_company_number(value: Any) -> str:
number = str(value or "").strip()
if not re.fullmatch(
r"[A-Za-z0-9._-]{1,64}",
number,
):
raise FrenchIdentifierError(
ERR_SANDBOX_NUMBER
)
return number
def validate_company_number(
value: Any,
*,
number_scheme: str,
) -> str:
if number_scheme == "fr_siren":
return validate_siren(value)
if number_scheme == "sandbox":
return validate_sandbox_company_number(value)
raise FrenchIdentifierError(
ERR_NUMBER_SCHEME
)
def validate_peppol_identifier(
value: Any,
) -> dict[str, str]:
identifier = str(value or "").strip()
match = re.fullmatch(
r"(?P<scheme>[0-9]{4}):"
r"(?P<value>[A-Za-z0-9._-]{1,128})",
identifier,
)
if not match:
raise FrenchIdentifierError(
ERR_PEPPOL_IDENTIFIER
)
return {
"identifier": identifier,
"scheme": match.group("scheme"),
"value": match.group("value"),
}
def validate_customer(
doc: Any,
method: str | None = None,
) -> None:
"""Valider les données de facturation électronique d'un client."""
del method
if not doc.get("einvoice_enabled"):
@@ -130,15 +185,33 @@ def validate_customer(
title=_("Type de client incompatible"),
)
number_scheme = (
doc.get("einvoice_number_scheme")
or "fr_siren"
)
try:
doc.einvoice_siren = validate_siren(
doc.get("einvoice_siren")
doc.einvoice_siren = validate_company_number(
doc.get("einvoice_siren"),
number_scheme=number_scheme,
)
if doc.get("einvoice_siret"):
doc.einvoice_siret = validate_siret(
doc.einvoice_siret,
expected_siren=doc.einvoice_siren,
if number_scheme == "fr_siren":
doc.einvoice_sandbox_peppol_identifier = ""
if doc.get("einvoice_siret"):
doc.einvoice_siret = validate_siret(
doc.einvoice_siret,
expected_siren=doc.einvoice_siren,
)
elif number_scheme == "sandbox":
doc.einvoice_siret = ""
peppol = validate_peppol_identifier(
doc.get(
"einvoice_sandbox_peppol_identifier"
)
)
except FrenchIdentifierError as exc:
@@ -149,12 +222,18 @@ def validate_customer(
_reset_directory_data_if_identifiers_changed(doc)
if number_scheme == "sandbox":
doc.einvoice_directory_status = "Configured"
doc.einvoice_routing_scheme = peppol["scheme"]
doc.einvoice_routing_address = peppol["value"]
doc.einvoice_receiving_platform = "Peppol"
doc.einvoice_directory_entry_id = ""
doc.einvoice_last_directory_sync = None
def _reset_directory_data_if_identifiers_changed(
doc: Any,
) -> None:
"""Invalider le routage après modification du SIREN ou du SIRET."""
if doc.is_new():
return
@@ -162,8 +241,10 @@ def _reset_directory_data_if_identifiers_changed(
"Customer",
doc.name,
[
"einvoice_number_scheme",
"einvoice_siren",
"einvoice_siret",
"einvoice_sandbox_peppol_identifier",
],
as_dict=True,
)
@@ -171,16 +252,21 @@ def _reset_directory_data_if_identifiers_changed(
if not previous:
return
previous_siren = previous.einvoice_siren or ""
previous_siret = previous.einvoice_siret or ""
previous_values = (
previous.einvoice_number_scheme or "",
previous.einvoice_siren or "",
previous.einvoice_siret or "",
previous.einvoice_sandbox_peppol_identifier or "",
)
current_siren = doc.einvoice_siren or ""
current_siret = doc.einvoice_siret or ""
current_values = (
doc.einvoice_number_scheme or "",
doc.einvoice_siren or "",
doc.einvoice_siret or "",
doc.einvoice_sandbox_peppol_identifier or "",
)
if (
previous_siren == current_siren
and previous_siret == current_siret
):
if previous_values == current_values:
return
doc.einvoice_directory_status = "Not Checked"
+580
View File
@@ -0,0 +1,580 @@
# 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 frappe.utils import now_datetime
from enuxia_einvoice.customer import (
FrenchIdentifierError,
validate_company_number,
)
from enuxia_einvoice.providers.superpdp.client import (
SuperPDPClient,
SuperPDPError,
)
def split_directory_identifier(
identifier: Any,
) -> dict[str, str]:
"""Séparer une adresse telle que 0225:732829320."""
raw_identifier = str(identifier or "").strip()
if not raw_identifier:
return {
"identifier": "",
"scheme": "",
"value": "",
}
scheme, separator, value = raw_identifier.partition(":")
if not separator:
return {
"identifier": raw_identifier,
"scheme": "",
"value": raw_identifier,
}
return {
"identifier": raw_identifier,
"scheme": scheme.strip(),
"value": value.strip(),
}
def extract_active_entries(
payload: Any,
) -> list[dict[str, Any]]:
"""Extraire, normaliser et dédupliquer les adresses actives."""
if not isinstance(payload, dict):
return []
raw_entries = payload.get("data")
if not isinstance(raw_entries, list):
return []
entries: list[dict[str, Any]] = []
seen_identifiers: set[str] = set()
for raw_entry in raw_entries:
if not isinstance(raw_entry, dict):
continue
if raw_entry.get("is_active") is not True:
continue
if raw_entry.get("is_replyto") is True:
continue
address = split_directory_identifier(
raw_entry.get("identifier")
)
identifier = address["identifier"]
if not identifier or identifier in seen_identifiers:
continue
seen_identifiers.add(identifier)
company = raw_entry.get("company")
if not isinstance(company, dict):
company = {}
entries.append(
{
**address,
"company_name": (
company.get("formal_name") or ""
),
"company_number": (
company.get("number") or ""
),
"company_city": (
company.get("city") or ""
),
}
)
return sorted(
entries,
key=lambda entry: entry["identifier"],
)
def _get_customer(customer_name: str):
customer = frappe.get_doc(
"Customer",
customer_name,
)
customer.check_permission("write")
if not customer.get("einvoice_enabled"):
frappe.throw(
_(
"Activez la facturation électronique "
"sur ce client avant de consulter l'annuaire."
),
title=_("Facturation électronique désactivée"),
)
if customer.customer_type != "Company":
frappe.throw(
_(
"La recherche dans l'annuaire est réservée "
"aux clients de type Société."
),
title=_("Type de client incompatible"),
)
try:
number_scheme = (
customer.get("einvoice_number_scheme")
or "fr_siren"
)
if number_scheme != "fr_siren":
frappe.throw(
_(
"La recherche dans l'annuaire français "
"est disponible uniquement pour les SIREN. "
"En Sandbox, renseignez directement "
"l'adresse Peppol fournie par SuperPDP."
),
title=_("Recherche non disponible en Sandbox"),
)
siren = validate_company_number(
customer.get("einvoice_siren"),
number_scheme=number_scheme,
)
except FrenchIdentifierError as exc:
frappe.throw(
str(exc),
title=_("SIREN invalide"),
)
return customer, siren
def _get_enabled_account(
account_name: str | None = None,
):
if account_name:
account = frappe.get_doc(
"Electronic Invoicing Account",
account_name,
)
if not account.enabled:
frappe.throw(
_("Le compte SuperPDP sélectionné est désactivé.")
)
return account
accounts = frappe.get_all(
"Electronic Invoicing Account",
filters={
"enabled": 1,
"provider": "SuperPDP",
},
pluck="name",
limit=2,
)
if not accounts:
frappe.throw(
_(
"Aucun compte SuperPDP actif n'est configuré."
),
title=_("Compte SuperPDP manquant"),
)
if len(accounts) > 1:
frappe.throw(
_(
"Plusieurs comptes SuperPDP sont actifs. "
"Le choix de la société émettrice sera ajouté "
"lors du workflow de facture."
),
title=_("Plusieurs comptes disponibles"),
)
return frappe.get_doc(
"Electronic Invoicing Account",
accounts[0],
)
def _get_authenticated_client(account):
if account.authentication_flow != "Client Credentials":
frappe.throw(
_(
"La consultation de l'annuaire prend actuellement "
"en charge uniquement 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(
_(
"Le compte SuperPDP actif ne possède pas "
"de Client ID ou de Client Secret."
),
title=_("Configuration SuperPDP incomplète"),
)
client = SuperPDPClient(
client_id=client_id,
client_secret=client_secret,
)
token = client.authenticate_client_credentials()
session_info = client.get_current_session(
token.access_token
)
company_status = str(
session_info.get(
"company_verification_status"
)
or ""
).lower()
if company_status != "verified":
frappe.throw(
_(
"La société associée au compte SuperPDP "
"n'est pas encore vérifiée."
),
title=_("Compte SuperPDP non vérifié"),
)
return client, token.access_token
def _find_exact_company(
payload: Any,
siren: str,
) -> dict[str, Any] | None:
if not isinstance(payload, dict):
return None
companies = payload.get("data")
if not isinstance(companies, list):
return None
for company in companies:
if not isinstance(company, dict):
continue
if str(company.get("number") or "") == siren:
return company
return None
def _current_identifier(customer) -> str:
scheme = str(
customer.get("einvoice_routing_scheme") or ""
).strip()
value = str(
customer.get("einvoice_routing_address") or ""
).strip()
if not value:
return ""
return f"{scheme}:{value}" if scheme else value
def _save_directory_state(
customer_name: str,
*,
status: str,
entry: dict[str, Any] | None = None,
) -> None:
values: dict[str, Any] = {
"einvoice_directory_status": status,
"einvoice_last_directory_sync": now_datetime(),
}
if entry:
values.update(
{
"einvoice_routing_scheme": (
entry.get("scheme") or ""
),
"einvoice_routing_address": (
entry.get("value") or ""
),
# Ces données ne sont pas exposées
# par french_directory/entries.
"einvoice_receiving_platform": "",
"einvoice_directory_entry_id": "",
}
)
else:
values.update(
{
"einvoice_routing_scheme": "",
"einvoice_routing_address": "",
"einvoice_receiving_platform": "",
"einvoice_directory_entry_id": "",
}
)
frappe.db.set_value(
"Customer",
customer_name,
values,
update_modified=True,
)
@frappe.whitelist()
def search_customer_directory(
customer_name: str,
account_name: str | None = None,
) -> dict[str, Any]:
"""Rechercher et éventuellement sélectionner une adresse."""
customer, siren = _get_customer(customer_name)
account = _get_enabled_account(account_name)
try:
client, access_token = (
_get_authenticated_client(account)
)
companies_payload = (
client.search_french_directory_companies(
access_token,
number=siren,
)
)
company = _find_exact_company(
companies_payload,
siren,
)
if not company:
_save_directory_state(
customer.name,
status="Not Found",
)
return {
"ok": True,
"found": False,
"selected": False,
"message": _(
"Aucune entreprise correspondant à ce SIREN "
"n'a été trouvée dans l'annuaire."
),
"entries": [],
}
entries_payload = (
client.list_french_directory_entries(
access_token,
number=siren,
)
)
entries = extract_active_entries(
entries_payload
)
if not entries:
_save_directory_state(
customer.name,
status="Not Found",
)
return {
"ok": True,
"found": True,
"selected": False,
"message": _(
"L'entreprise existe dans l'annuaire, "
"mais aucune adresse active n'est disponible."
),
"company": company,
"entries": [],
}
current_identifier = _current_identifier(
customer
)
selected_entry = next(
(
entry
for entry in entries
if entry["identifier"]
== current_identifier
),
None,
)
if not selected_entry and len(entries) == 1:
selected_entry = entries[0]
if selected_entry:
_save_directory_state(
customer.name,
status="Found",
entry=selected_entry,
)
return {
"ok": True,
"found": True,
"selected": True,
"message": _(
"L'adresse de facturation électronique "
"a été enregistrée."
),
"company": company,
"entry": selected_entry,
"entries": entries,
}
_save_directory_state(
customer.name,
status="Found",
)
return {
"ok": True,
"found": True,
"selected": False,
"message": _(
"Plusieurs adresses actives ont été trouvées. "
"Sélectionnez l'adresse à utiliser."
),
"company": company,
"entries": entries,
}
except SuperPDPError as exc:
_save_directory_state(
customer.name,
status="Error",
)
return {
"ok": False,
"found": False,
"selected": False,
"message": exc.message,
"status_code": exc.status_code,
"api_code": exc.api_code,
"entries": [],
}
@frappe.whitelist()
def select_customer_directory_entry(
customer_name: str,
identifier: str,
account_name: str | None = None,
) -> dict[str, Any]:
"""Valider puis enregistrer une adresse choisie."""
customer, siren = _get_customer(customer_name)
account = _get_enabled_account(account_name)
requested_identifier = str(
identifier or ""
).strip()
try:
client, access_token = (
_get_authenticated_client(account)
)
entries_payload = (
client.list_french_directory_entries(
access_token,
number=siren,
)
)
entries = extract_active_entries(
entries_payload
)
selected_entry = next(
(
entry
for entry in entries
if entry["identifier"]
== requested_identifier
),
None,
)
if not selected_entry:
frappe.throw(
_(
"L'adresse sélectionnée n'est plus active "
"dans l'annuaire."
),
title=_("Adresse indisponible"),
)
_save_directory_state(
customer.name,
status="Found",
entry=selected_entry,
)
return {
"ok": True,
"selected": True,
"message": _(
"L'adresse de facturation électronique "
"a été enregistrée."
),
"entry": selected_entry,
}
except SuperPDPError as exc:
_save_directory_state(
customer.name,
status="Error",
)
return {
"ok": False,
"selected": False,
"message": exc.message,
"status_code": exc.status_code,
"api_code": exc.api_code,
}
+157 -1
View File
@@ -4,18 +4,174 @@
frappe.ui.form.on("Customer", {
refresh(frm) {
toggle_electronic_invoicing_section(frm);
add_directory_button(frm);
},
customer_type(frm) {
toggle_electronic_invoicing_section(frm);
},
einvoice_enabled(frm) {
add_directory_button(frm);
},
});
function toggle_electronic_invoicing_section(frm) {
const is_company = frm.doc.customer_type === "Company";
const is_company =
frm.doc.customer_type === "Company";
frm.toggle_display(
"einvoice_section",
is_company
);
}
function add_directory_button(frm) {
if (
frm.is_new()
|| frm.doc.customer_type !== "Company"
|| !frm.doc.einvoice_enabled
|| frm.doc.einvoice_number_scheme !== "fr_siren"
) {
return;
}
frm.add_custom_button(
__("Rechercher dans lannuaire"),
async () => {
if (frm.is_dirty()) {
frappe.msgprint(
__(
"Enregistrez le client avant "
+ "de consulter lannuaire."
)
);
return;
}
const response = await frappe.call({
method:
"enuxia_einvoice.directory."
+ "search_customer_directory",
args: {
customer_name: frm.doc.name,
},
freeze: true,
freeze_message: __(
"Recherche dans lannuaire…"
),
});
const result = response.message || {};
const entries = result.entries || [];
await frm.reload_doc();
if (!result.ok) {
frappe.msgprint({
title: __("Erreur SuperPDP"),
indicator: "red",
message: frappe.utils.escape_html(
result.message
|| __("Erreur inconnue.")
),
});
return;
}
if (result.selected) {
frappe.show_alert({
message: result.message,
indicator: "green",
});
return;
}
if (entries.length <= 1) {
frappe.msgprint({
title: __("Annuaire"),
indicator: result.found
? "orange"
: "red",
message: frappe.utils.escape_html(
result.message
),
});
return;
}
show_entry_selector(
frm,
entries,
result.company || {}
);
},
__("Facturation électronique")
);
}
function show_entry_selector(
frm,
entries,
company
) {
const identifiers = entries.map(
(entry) => entry.identifier
);
frappe.prompt(
[
{
fieldname: "identifier",
label: __(
"Adresse électronique"
),
fieldtype: "Select",
options: identifiers.join("\n"),
reqd: 1,
},
],
async (values) => {
const response = await frappe.call({
method:
"enuxia_einvoice.directory."
+ "select_customer_directory_entry",
args: {
customer_name: frm.doc.name,
identifier: values.identifier,
},
freeze: true,
freeze_message: __(
"Enregistrement de ladresse…"
),
});
const result = response.message || {};
await frm.reload_doc();
frappe.msgprint({
title: result.ok
? __("Adresse enregistrée")
: __("Erreur SuperPDP"),
indicator: result.ok
? "green"
: "red",
message: frappe.utils.escape_html(
result.message
|| __("Réponse inconnue.")
),
});
},
company.formal_name
? __(
"Adresses disponibles pour {0}",
[company.formal_name]
)
: __("Choisir une adresse"),
__("Utiliser cette adresse")
);
}
+41 -8
View File
@@ -29,25 +29,58 @@ CUSTOM_FIELDS = {
),
},
{
"fieldname": "einvoice_siren",
"label": "SIREN",
"fieldtype": "Data",
"fieldname": "einvoice_number_scheme",
"label": "Type didentifiant entreprise",
"fieldtype": "Select",
"insert_after": "einvoice_enabled",
"length": 9,
"options": "fr_siren\nsandbox",
"default": "fr_siren",
"depends_on": "eval:doc.einvoice_enabled",
"description": (
"Identifiant à 9 chiffres de l'entreprise française."
"Utilisez fr_siren en production et sandbox "
"pour les entreprises fictives SuperPDP."
),
},
{
"fieldname": "einvoice_siren",
"label": "Identifiant entreprise",
"fieldtype": "Data",
"insert_after": "einvoice_number_scheme",
"length": 64,
"depends_on": "eval:doc.einvoice_enabled",
"description": (
"SIREN en production ou numéro fictif "
"de lentreprise en bac à sable."
),
"in_list_view": 0,
"in_standard_filter": 1,
},
{
"fieldname": "einvoice_sandbox_peppol_identifier",
"label": "Adresse Peppol Sandbox",
"fieldtype": "Data",
"insert_after": "einvoice_siren",
"length": 140,
"depends_on": (
"eval:doc.einvoice_enabled "
"&& doc.einvoice_number_scheme == 'sandbox'"
),
"mandatory_depends_on": (
"eval:doc.einvoice_enabled "
"&& doc.einvoice_number_scheme == 'sandbox'"
),
"description": (
"Adresse complète fournie par SuperPDP, "
"par exemple 0225:315143296_4919."
),
},
{
"fieldname": "einvoice_siret",
"label": "SIRET",
"fieldtype": "Data",
"insert_after": "einvoice_siren",
"insert_after": "einvoice_sandbox_peppol_identifier",
"length": 14,
"depends_on": "eval:doc.einvoice_enabled",
"depends_on": "eval:doc.einvoice_enabled && doc.einvoice_number_scheme == 'fr_siren'",
"description": (
"Identifiant à 14 chiffres de l'établissement concerné."
),
@@ -62,7 +95,7 @@ CUSTOM_FIELDS = {
"label": "Statut dans l'annuaire",
"fieldtype": "Select",
"insert_after": "einvoice_routing_column",
"options": "Not Checked\nFound\nNot Found\nError",
"options": "Not Checked\nConfigured\nFound\nNot Found\nError",
"default": "Not Checked",
"read_only": 1,
"depends_on": "eval:doc.einvoice_enabled",
+37 -10
View File
@@ -1,6 +1,3 @@
# Copyright (c) 2026, Enuxia and contributors
# For license information, please see license.txt
from __future__ import annotations
import unittest
@@ -9,6 +6,9 @@ from enuxia_einvoice.customer import (
FrenchIdentifierError,
is_luhn_valid,
normalize_registration_number,
validate_company_number,
validate_peppol_identifier,
validate_sandbox_company_number,
validate_siren,
validate_siret,
)
@@ -23,13 +23,6 @@ class TestFrenchCustomerIdentifiers(unittest.TestCase):
"732829320",
)
self.assertEqual(
normalize_registration_number(
"732.829.320"
),
"732829320",
)
def test_reject_letters(self) -> None:
with self.assertRaises(FrenchIdentifierError):
normalize_registration_number(
@@ -74,6 +67,40 @@ class TestFrenchCustomerIdentifiers(unittest.TestCase):
is_luhn_valid("732829321")
)
def test_valid_sandbox_company_number(self) -> None:
self.assertEqual(
validate_sandbox_company_number(
"000000001"
),
"000000001",
)
def test_company_number_scheme(self) -> None:
self.assertEqual(
validate_company_number(
"000000001",
number_scheme="sandbox",
),
"000000001",
)
def test_valid_peppol_identifier(self) -> None:
result = validate_peppol_identifier(
"0225:315143296_4919"
)
self.assertEqual(result["scheme"], "0225")
self.assertEqual(
result["value"],
"315143296_4919",
)
def test_invalid_peppol_identifier(self) -> None:
with self.assertRaises(FrenchIdentifierError):
validate_peppol_identifier(
"315143296_4919"
)
if __name__ == "__main__":
unittest.main()
+63
View File
@@ -0,0 +1,63 @@
# Copyright (c) 2026, Enuxia and contributors
# For license information, please see license.txt
from __future__ import annotations
import unittest
from enuxia_einvoice.directory import (
extract_active_entries,
split_directory_identifier,
)
class TestFrenchDirectoryHelpers(unittest.TestCase):
def test_split_peppol_identifier(self) -> None:
result = split_directory_identifier(
"0225:732829320"
)
self.assertEqual(result["scheme"], "0225")
self.assertEqual(result["value"], "732829320")
def test_split_identifier_without_scheme(self) -> None:
result = split_directory_identifier(
"732829320"
)
self.assertEqual(result["scheme"], "")
self.assertEqual(result["value"], "732829320")
def test_extract_only_active_entries(self) -> None:
result = extract_active_entries(
{
"data": [
{
"identifier": "0225:732829320",
"is_active": True,
"company": {
"formal_name": "Test",
"number": "732829320",
},
},
{
"identifier": "0225:73282932000074",
"is_active": False,
},
{
"identifier": "0225:732829320",
"is_active": True,
},
]
}
)
self.assertEqual(len(result), 1)
self.assertEqual(
result[0]["identifier"],
"0225:732829320",
)
if __name__ == "__main__":
unittest.main()