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

feat: inspect incoming SuperPDP invoices

Browse Source
This commit is contained in:
Julien Denizot
2026-06-19 12:40:18 +00:00
parent 9a53d1f2c3
commit 11f7c0804e
4 changed files with 530 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
enuxia_einvoice/inbound.py
+294
View File
@@ -0,0 +1,294 @@
# 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.sending import (
extract_latest_provider_status,
)
from enuxia_einvoice.validation import (
get_authenticated_client,
)
def _get(
value: Any,
fieldname: str,
default: Any = None,
) -> Any:
if not isinstance(value, dict):
return default
result = value.get(fieldname)
return default if result is None else result
def _party_summary(
party: Any,
) -> dict[str, Any]:
if not isinstance(party, dict):
party = {}
legal_identifier = _get(
party,
"legal_registration_identifier",
{},
)
electronic_address = _get(
party,
"electronic_address",
{},
)
postal_address = _get(
party,
"postal_address",
{},
)
return {
"name": str(
_get(party, "name", "")
),
"trading_name": str(
_get(party, "trading_name", "")
),
"vat_identifier": str(
_get(party, "vat_identifier", "")
),
"legal_identifier_scheme": str(
_get(
legal_identifier,
"scheme",
"",
)
),
"legal_identifier_value": str(
_get(
legal_identifier,
"value",
"",
)
),
"electronic_address_scheme": str(
_get(
electronic_address,
"scheme",
"",
)
),
"electronic_address_value": str(
_get(
electronic_address,
"value",
"",
)
),
"postal_address": postal_address,
}
def summarize_incoming_invoice(
remote_invoice: Any,
) -> dict[str, Any]:
"""Résumer une facture entrante sans modifier ERPNext."""
if not isinstance(remote_invoice, dict):
return {}
en_invoice = _get(
remote_invoice,
"en_invoice",
{},
)
totals = _get(
en_invoice,
"totals",
{},
)
lines = _get(
en_invoice,
"lines",
[],
)
events = _get(
remote_invoice,
"events",
[],
)
if not isinstance(lines, list):
lines = []
if not isinstance(events, list):
events = []
return {
"remote_id": remote_invoice.get("id"),
"direction": str(
remote_invoice.get("direction") or ""
),
"created_at": str(
remote_invoice.get("created_at") or ""
),
"external_id": str(
remote_invoice.get("external_id") or ""
),
"provider_status": (
extract_latest_provider_status(
remote_invoice
)
),
"invoice_number": str(
_get(en_invoice, "number", "")
),
"type_code": _get(
en_invoice,
"type_code",
),
"issue_date": str(
_get(en_invoice, "issue_date", "")
),
"payment_due_date": str(
_get(
en_invoice,
"payment_due_date",
"",
)
),
"currency_code": str(
_get(
en_invoice,
"currency_code",
"",
)
),
"seller": _party_summary(
_get(en_invoice, "seller", {})
),
"buyer": _party_summary(
_get(en_invoice, "buyer", {})
),
"total_without_vat": _get(
totals,
"total_without_vat",
"",
),
"total_vat_amount": _get(
totals,
"total_vat_amount",
"",
),
"total_with_vat": _get(
totals,
"total_with_vat",
"",
),
"amount_due_for_payment": _get(
totals,
"amount_due_for_payment",
"",
),
"line_count": len(lines),
"event_count": len(events),
}
@frappe.whitelist()
def inspect_incoming_invoices(
company: str,
limit: int = 20,
starting_after_id: int | None = None,
) -> dict[str, Any]:
"""Lister les factures reçues sans rien importer."""
frappe.only_for("System Manager")
requested_limit = max(
1,
min(int(limit), 100),
)
client, access_token = (
get_authenticated_client(company)
)
payload = client.list_invoices(
access_token,
direction="in",
order="asc",
starting_after_id=starting_after_id,
limit=requested_limit,
expand=[
"en_invoice",
"en_invoice.seller",
"en_invoice.buyer",
"en_invoice.lines",
],
)
raw_invoices = payload.get("data") or []
invoices = [
summarize_incoming_invoice(invoice)
for invoice in raw_invoices
if isinstance(invoice, dict)
]
return {
"company": company,
"count": payload.get("count", 0),
"returned": len(invoices),
"has_before": bool(
payload.get("has_before")
),
"has_after": bool(
payload.get("has_after")
),
"last_remote_id": (
invoices[-1]["remote_id"]
if invoices
else starting_after_id
),
"invoices": invoices,
}
@frappe.whitelist()
def inspect_incoming_invoice(
company: str,
invoice_id: int,
) -> dict[str, Any]:
"""Récupérer le détail EN16931 d’une facture reçue."""
frappe.only_for("System Manager")
client, access_token = (
get_authenticated_client(company)
)
invoice = client.get_invoice(
access_token,
invoice_id=int(invoice_id),
invoice_format="en16931",
)
if invoice.get("direction") != "in":
frappe.throw(
_(
"La facture SuperPDP demandée "
"n’est pas une facture entrante."
)
)
return invoice
enuxia_einvoice/providers/superpdp/client.py
+54 -3
View File
@@ -279,22 +279,48 @@ class SuperPDPClient:
self,
access_token: str,
*,
direction: str = "out",
direction: str | None = "out",
date: str | None = None,
order: str = "desc",
starting_after_id: int | None = None,
ending_before_id: int | None = None,
limit: int = 1000,
expand: list[str] | None = None,
) -> dict[str, Any]:
"""Lister les factures pour retrouver un envoi existant."""
"""Lister les factures avec pagination et expansion."""
if (
starting_after_id is not None
and ending_before_id is not None
):
raise ValueError(
"Utilisez un seul curseur de pagination."
)
params: dict[str, Any] = {
"direction": direction,
"order": order,
"limit": limit,
}
if direction:
params["direction"] = direction
if date:
params["date"] = date
if starting_after_id is not None:
params["starting_after_id"] = (
starting_after_id
)
if ending_before_id is not None:
params["ending_before_id"] = (
ending_before_id
)
if expand:
params["expand[]"] = expand
response = self.session.get(
f"{self.base_url}/v1.beta/invoices",
headers=self._authorization_headers(
@@ -306,6 +332,31 @@ class SuperPDPClient:
return self._read_response(response)
def get_invoice(
self,
access_token: str,
*,
invoice_id: int,
invoice_format: str = "en16931",
) -> dict[str, Any]:
"""Récupérer une facture précise."""
response = self.session.get(
(
f"{self.base_url}/v1.beta/invoices/"
f"{invoice_id}"
),
headers=self._authorization_headers(
access_token
),
params={
"format": invoice_format,
},
timeout=self.timeout,
)
return self._read_response(response)
def create_invoice(
self,
access_token: str,
enuxia_einvoice/providers/superpdp/test_client.py
+101
View File
@@ -535,5 +535,106 @@ class TestSuperPDPClient(unittest.TestCase):
)
def test_list_incoming_invoices_with_expansion(
self,
) -> None:
response = Mock()
response.ok = True
response.status_code = 200
response.json.return_value = {
"count": 1,
"data": [
{
"id": 80123,
"direction": "in",
"en_invoice": {
"number": "SUP-2026-001",
},
}
],
"has_before": False,
"has_after": False,
}
self.session.get.return_value = response
result = self.client.list_invoices(
"secret-access-token",
direction="in",
order="asc",
starting_after_id=80000,
limit=20,
expand=[
"en_invoice",
"en_invoice.seller",
"en_invoice.buyer",
"en_invoice.lines",
],
)
self.assertEqual(
result["data"][0]["direction"],
"in",
)
self.session.get.assert_called_once_with(
"https://api.superpdp.tech/v1.beta/invoices",
headers={
"Authorization": "Bearer secret-access-token",
},
params={
"direction": "in",
"order": "asc",
"starting_after_id": 80000,
"limit": 20,
"expand[]": [
"en_invoice",
"en_invoice.seller",
"en_invoice.buyer",
"en_invoice.lines",
],
},
timeout=(5, 30),
)
def test_get_invoice_as_en16931(self) -> None:
response = Mock()
response.ok = True
response.status_code = 200
response.json.return_value = {
"id": 80123,
"direction": "in",
"en_invoice": {
"number": "SUP-2026-001",
},
}
self.session.get.return_value = response
result = self.client.get_invoice(
"secret-access-token",
invoice_id=80123,
)
self.assertEqual(
result["en_invoice"]["number"],
"SUP-2026-001",
)
self.session.get.assert_called_once_with(
(
"https://api.superpdp.tech/"
"v1.beta/invoices/80123"
),
headers={
"Authorization": "Bearer secret-access-token",
},
params={
"format": "en16931",
},
timeout=(5, 30),
)
if __name__ == "__main__":
unittest.main()
enuxia_einvoice/tests/test_inbound.py
+81
View File
@@ -0,0 +1,81 @@
from __future__ import annotations
import unittest
from enuxia_einvoice.inbound import (
summarize_incoming_invoice,
)
class TestInboundInvoiceSummary(unittest.TestCase):
def test_summarize_incoming_invoice(self) -> None:
result = summarize_incoming_invoice(
{
"id": 80123,
"direction": "in",
"created_at": "2026-06-19T12:00:00Z",
"events": [
{
"id": 1,
"status_code": "fr:202",
}
],
"en_invoice": {
"number": "SUP-2026-001",
"type_code": 380,
"issue_date": "2026-06-19",
"payment_due_date": "2026-07-19",
"currency_code": "EUR",
"seller": {
"name": "Tricatel",
"vat_identifier": "FR15000000001",
"legal_registration_identifier": {
"scheme": "0002",
"value": "000000001",
},
},
"buyer": {
"name": "Burger Queen",
},
"lines": [
{
"identifier": "001",
}
],
"totals": {
"total_without_vat": "50.00",
"total_with_vat": "60.00",
"amount_due_for_payment": "60.00",
},
},
}
)
self.assertEqual(
result["remote_id"],
80123,
)
self.assertEqual(
result["invoice_number"],
"SUP-2026-001",
)
self.assertEqual(
result["seller"]["name"],
"Tricatel",
)
self.assertEqual(
result["total_with_vat"],
"60.00",
)
self.assertEqual(
result["line_count"],
1,
)
self.assertEqual(
result["provider_status"],
"fr:202",
)
if __name__ == "__main__":
unittest.main()