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

feat: preview inbound purchase invoices

Browse Source
This commit is contained in:
Julien Denizot
2026-06-19 13:01:38 +00:00
parent 320dd80d0e
commit 78acf3a578
5 changed files with 1169 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
enuxia_einvoice/enuxia_e_invoice/doctype/electronic_invoice_exchange/electronic_invoice_exchange.js
+239
View File
@@ -11,6 +11,12 @@ frappe.ui.form.on(
set_review_indicator(frm);
frm.add_custom_button(
__("Prévisualiser l’import"),
() => preview_purchase_invoice(frm),
__("Réception fournisseur")
);
if (!frm.doc.supplier) {
frm.add_custom_button(
__("Rapprochement automatique"),
@@ -187,3 +193,236 @@ function clear_supplier_match(frm) {
}
);
}
function preview_purchase_invoice(frm) {
frappe.call({
method:
"enuxia_einvoice.purchase_preview."
+ "preview_purchase_invoice",
args: {
exchange_name: frm.doc.name,
},
freeze: true,
freeze_message: __(
"Construction de la prévisualisation…"
),
callback(response) {
const preview = response.message || {};
show_purchase_preview_dialog(
preview
);
},
});
}
function show_purchase_preview_dialog(
preview
) {
const escape = (value) =>
frappe.utils.escape_html(
String(value ?? "")
);
const lineRows = (
preview.lines || []
).map((line) => `
<tr>
<td>${escape(line.line_number)}</td>
<td>
<strong>${escape(line.item_name)}</strong>
${
line.description
? `<br><small>${escape(line.description)}</small>`
: ""
}
</td>
<td>${escape(line.quantity)}</td>
<td>
${escape(line.unit_code)}
<br>
<small>
ERPNext :
${escape(line.resolved_uom || "Non rapprochée")}
</small>
</td>
<td>
${escape(
line.suggested_item_code
|| "Non rapproché"
)}
</td>
<td style="text-align:right">
${escape(line.unit_price)}
</td>
<td style="text-align:right">
${escape(line.net_amount)}
</td>
<td style="text-align:right">
${escape(line.vat_rate)} %
</td>
</tr>
`).join("");
const checks = (
preview.checks || []
).map((check) => `
<li>
<span class="indicator-pill ${
check.ok ? "green" : "red"
}">
${check.ok ? __("OK") : __("Écart")}
</span>
${escape(check.label)}
</li>
`).join("");
const blockers = (
preview.blockers || []
).map(
(message) =>
`<li>${escape(message)}</li>`
).join("");
const warnings = (
preview.warnings || []
).map(
(message) =>
`<li>${escape(message)}</li>`
).join("");
const readiness = preview.can_create_draft
? `
<div class="alert alert-success">
${__(
"La facture peut techniquement être préparée en brouillon."
)}
</div>
`
: `
<div class="alert alert-warning">
${__(
"La création du brouillon reste bloquée."
)}
</div>
`;
const html = `
${readiness}
<div class="mb-3">
<strong>${__(
"Facture fournisseur"
)} :</strong>
${escape(preview.invoice_number)}
<br>
<strong>${__(
"Fournisseur"
)} :</strong>
${escape(
preview.supplier || "Non rapproché"
)}
<br>
<strong>${__("Date")} :</strong>
${escape(preview.issue_date)}
&nbsp;—&nbsp;
<strong>${__(
"Échéance"
)} :</strong>
${escape(preview.payment_due_date)}
</div>
<div class="table-responsive">
<table class="table table-bordered">
<thead>
<tr>
<th>#</th>
<th>${__("Désignation")}</th>
<th>${__("Quantité")}</th>
<th>${__("Unité")}</th>
<th>${__("Article proposé")}</th>
<th>${__("Prix unitaire")}</th>
<th>${__("Montant HT")}</th>
<th>${__("TVA")}</th>
</tr>
</thead>
<tbody>
${lineRows}
</tbody>
</table>
</div>
<div class="row">
<div class="col-sm-6">
<h5>${__("Contrôles")}</h5>
<ul>${checks}</ul>
${
blockers
? `
<h5>${__("Blocages")}</h5>
<ul>${blockers}</ul>
`
: ""
}
${
warnings
? `
<h5>${__("Avertissements")}</h5>
<ul>${warnings}</ul>
`
: ""
}
</div>
<div class="col-sm-6 text-right">
<p>
<strong>${__("Total HT")} :</strong>
${escape(preview.totals?.net_total)}
${escape(preview.currency)}
</p>
<p>
<strong>${__("TVA")} :</strong>
${escape(preview.totals?.vat_total)}
${escape(preview.currency)}
</p>
<p>
<strong>${__("Total TTC")} :</strong>
${escape(preview.totals?.grand_total)}
${escape(preview.currency)}
</p>
<p>
<strong>${__("Net à payer")} :</strong>
${escape(preview.totals?.amount_due)}
${escape(preview.currency)}
</p>
</div>
</div>
`;
const dialog = new frappe.ui.Dialog({
title: __(
"Prévisualisation de la facture d’achat"
),
size: "extra-large",
fields: [
{
fieldname: "preview",
fieldtype: "HTML",
options: html,
},
],
primary_action_label: __("Fermer"),
primary_action() {
dialog.hide();
},
});
dialog.show();
}
enuxia_einvoice/enuxia_e_invoice/doctype/electronic_invoice_exchange/electronic_invoice_exchange.json
+9 -1
View File
@@ -6,6 +6,7 @@
"doctype": "DocType",
"engine": "InnoDB",
"field_order": [
"display_title",
"identity_section",
"direction",
"company",
@@ -53,6 +54,13 @@
"last_error"
],
"fields": [
{
"fieldname": "display_title",
"label": "Titre",
"fieldtype": "Data",
"read_only": 1,
"hidden": 1
},
{
"fieldname": "identity_section",
"label": "Identification",
@@ -381,7 +389,7 @@
"sort_field": "creation",
"sort_order": "DESC",
"states": [],
"title_field": "source_document",
"title_field": "display_title",
"track_changes": 1,
"search_fields": "source_document,external_id,remote_id,party",
"allow_rename": 0,
enuxia_einvoice/enuxia_e_invoice/doctype/electronic_invoice_exchange/electronic_invoice_exchange.py
+17
View File
@@ -42,11 +42,28 @@ class ElectronicInvoiceExchange(Document):
def before_validate(self) -> None:
if self.direction == "Inbound":
invoice_number = (
self.remote_invoice_number
or self.remote_id
or self.external_id
or _("Facture fournisseur")
)
self.display_title = _(
"Facture fournisseur {0}"
).format(invoice_number)
if not self.review_status:
self.review_status = "To Review"
return
self.display_title = (
self.source_document
or self.external_id
or _("Facture électronique sortante")
)
# Les statuts d’examen et de rapprochement fournisseur
# ne concernent jamais les échanges sortants.
self.review_status = None
enuxia_einvoice/purchase_preview.py
+719
View File
@@ -0,0 +1,719 @@
# Copyright (c) 2026, Enuxia and contributors
# For license information, please see license.txt
from __future__ import annotations
import json
from decimal import Decimal, InvalidOperation
from typing import Any
import frappe
from frappe import _
AMOUNT_TOLERANCE = Decimal("0.01")
UNIT_CODE_CANDIDATES = {
"C62": ["Nos", "Unit"],
"KGM": ["Kg", "Kilogram"],
"GRM": ["Gram"],
"LTR": ["Litre", "Liter"],
"MLT": ["Millilitre", "Milliliter"],
"HUR": ["Hour"],
"DAY": ["Day"],
"MTR": ["Meter", "Metre"],
"MTK": ["Square Meter"],
"MTQ": ["Cubic Meter"],
}
class PurchasePreviewError(ValueError):
"""La prévisualisation de la facture d’achat a échoué."""
def _dictionary(
value: Any,
) -> dict[str, Any]:
return value if isinstance(value, dict) else {}
def _list(
value: Any,
) -> list[Any]:
return value if isinstance(value, list) else []
def _decimal(
value: Any,
*,
fieldname: str,
) -> Decimal:
if isinstance(value, dict):
value = value.get("value")
try:
return Decimal(
str(
value
if value not in (None, "")
else "0"
)
)
except (
InvalidOperation,
TypeError,
ValueError,
) as error:
raise PurchasePreviewError(
f"Valeur numérique invalide pour {fieldname}."
) from error
def _decimal_text(
value: Decimal,
places: str = "0.01",
) -> str:
return format(
value.quantize(Decimal(places)),
"f",
)
def _identifier_value(
value: Any,
) -> str:
if isinstance(value, dict):
return str(
value.get("value")
or value.get("identifier")
or ""
).strip()
return str(value or "").strip()
def _first_identifier(
line: dict[str, Any],
item_information: dict[str, Any],
) -> str:
keys = (
"seller_item_identifier",
"sellers_item_identifier",
"standard_item_identifier",
"item_identifier",
)
for container in (
item_information,
line,
):
for key in keys:
identifier = _identifier_value(
container.get(key)
)
if identifier:
return identifier
return ""
def build_purchase_invoice_preview(
remote_invoice: dict[str, Any],
*,
supplier: str = "",
) -> dict[str, Any]:
"""Construire une prévisualisation sans accès à la base."""
if not isinstance(remote_invoice, dict):
raise PurchasePreviewError(
"Le document EN16931 est invalide."
)
en_invoice = _dictionary(
remote_invoice.get("en_invoice")
)
if not en_invoice:
en_invoice = remote_invoice
raw_lines = _list(
en_invoice.get("lines")
)
if not raw_lines:
raise PurchasePreviewError(
"La facture reçue ne contient aucune ligne."
)
totals = _dictionary(
en_invoice.get("totals")
)
preview_lines: list[dict[str, Any]] = []
line_total = Decimal("0")
calculation_warnings: list[str] = []
for index, raw_line in enumerate(
raw_lines,
start=1,
):
line = _dictionary(raw_line)
item_information = _dictionary(
line.get("item_information")
)
price_details = _dictionary(
line.get("price_details")
)
vat_information = _dictionary(
line.get("vat_information")
)
quantity = _decimal(
line.get("invoiced_quantity"),
fieldname=f"quantité ligne {index}",
)
unit_price = _decimal(
price_details.get("item_net_price"),
fieldname=f"prix unitaire ligne {index}",
)
net_amount = _decimal(
line.get("net_amount"),
fieldname=f"montant net ligne {index}",
)
calculated_amount = (
quantity * unit_price
)
amount_difference = (
net_amount - calculated_amount
)
if abs(amount_difference) > AMOUNT_TOLERANCE:
calculation_warnings.append(
(
f"Ligne {index} : quantité × prix "
f"diffère du montant net de "
f"{_decimal_text(amount_difference)}."
)
)
vat_rate = _decimal(
vat_information.get(
"invoiced_item_vat_rate"
),
fieldname=f"taux de TVA ligne {index}",
)
line_total += net_amount
unit_code = str(
line.get(
"invoiced_quantity_code"
)
or ""
).strip()
item_name = str(
item_information.get("name")
or ""
).strip()
description = str(
item_information.get("description")
or ""
).strip()
preview_lines.append(
{
"line_number": index,
"identifier": str(
line.get("identifier")
or index
),
"supplier_item_identifier": (
_first_identifier(
line,
item_information,
)
),
"item_name": item_name,
"description": description,
"quantity": _decimal_text(
quantity,
"0.0001",
),
"unit_code": unit_code,
"uom_candidates": (
UNIT_CODE_CANDIDATES.get(
unit_code,
[],
)
),
"unit_price": _decimal_text(
unit_price,
"0.000001",
),
"net_amount": _decimal_text(
net_amount,
),
"calculated_amount": (
_decimal_text(
calculated_amount,
)
),
"amount_difference": (
_decimal_text(
amount_difference,
)
),
"vat_category": str(
vat_information.get(
"invoiced_item_vat_category_code"
)
or ""
),
"vat_rate": _decimal_text(
vat_rate,
),
}
)
net_total = _decimal(
totals.get("total_without_vat"),
fieldname="total HT",
)
vat_total = _decimal(
totals.get("total_vat_amount"),
fieldname="total TVA",
)
grand_total = _decimal(
totals.get("total_with_vat"),
fieldname="total TTC",
)
amount_due = _decimal(
totals.get("amount_due_for_payment"),
fieldname="net à payer",
)
vat_breakdown_total = sum(
(
_decimal(
breakdown.get(
"vat_category_tax_amount"
),
fieldname="ventilation TVA",
)
for breakdown in _list(
en_invoice.get(
"vat_break_down"
)
)
if isinstance(
breakdown,
dict,
)
),
Decimal("0"),
)
checks = [
{
"code": "LINE_TOTAL",
"label": (
"Somme des lignes égale au total HT"
),
"ok": (
abs(line_total - net_total)
<= AMOUNT_TOLERANCE
),
"expected": _decimal_text(
net_total
),
"actual": _decimal_text(
line_total
),
},
{
"code": "VAT_TOTAL",
"label": (
"Ventilation TVA égale au total TVA"
),
"ok": (
abs(
vat_breakdown_total
- vat_total
)
<= AMOUNT_TOLERANCE
),
"expected": _decimal_text(
vat_total
),
"actual": _decimal_text(
vat_breakdown_total
),
},
{
"code": "GRAND_TOTAL",
"label": (
"Total HT + TVA égal au total TTC"
),
"ok": (
abs(
net_total
+ vat_total
- grand_total
)
<= AMOUNT_TOLERANCE
),
"expected": _decimal_text(
grand_total
),
"actual": _decimal_text(
net_total + vat_total
),
},
]
return {
"remote_id": remote_invoice.get("id"),
"invoice_number": str(
en_invoice.get("number")
or ""
),
"supplier": supplier,
"issue_date": str(
en_invoice.get("issue_date")
or ""
),
"payment_due_date": str(
en_invoice.get(
"payment_due_date"
)
or ""
),
"currency": str(
en_invoice.get(
"currency_code"
)
or ""
),
"line_count": len(
preview_lines
),
"lines": preview_lines,
"totals": {
"net_total": _decimal_text(
net_total
),
"vat_total": _decimal_text(
vat_total
),
"grand_total": _decimal_text(
grand_total
),
"amount_due": _decimal_text(
amount_due
),
},
"checks": checks,
"calculation_warnings": (
calculation_warnings
),
}
def _load_semantic_payload(
file_url: str,
) -> dict[str, Any]:
file_name = frappe.db.get_value(
"File",
{
"file_url": file_url,
},
"name",
)
if not file_name:
frappe.throw(
_(
"Le fichier EN16931 archivé "
"est introuvable."
)
)
file_document = frappe.get_doc(
"File",
file_name,
)
content = file_document.get_content()
if isinstance(content, bytes):
content = content.decode("utf-8")
try:
result = json.loads(content)
except (
TypeError,
ValueError,
json.JSONDecodeError,
):
frappe.throw(
_(
"Le fichier EN16931 archivé "
"ne contient pas un JSON valide."
)
)
if not isinstance(result, dict):
frappe.throw(
_("Le contenu EN16931 est invalide.")
)
return result
def _resolve_uom(
unit_code: str,
) -> str:
candidates = [
unit_code,
*UNIT_CODE_CANDIDATES.get(
unit_code,
[],
),
]
for candidate in candidates:
if (
candidate
and frappe.db.exists(
"UOM",
candidate,
)
):
return candidate
return ""
def _suggest_item(
item_name: str,
) -> tuple[str, str]:
item_name = str(
item_name or ""
).strip()
if not item_name:
return "", ""
if frappe.db.exists(
"Item",
item_name,
):
return item_name, "Exact Item Code"
matches = frappe.get_all(
"Item",
filters={
"item_name": item_name,
"disabled": 0,
},
pluck="name",
limit=2,
)
if len(matches) == 1:
return (
matches[0],
"Exact Item Name",
)
return "", ""
def _find_existing_purchase_invoice(
*,
supplier: str,
bill_no: str,
) -> str:
if not supplier or not bill_no:
return ""
matches = frappe.get_all(
"Purchase Invoice",
filters=[
[
"supplier",
"=",
supplier,
],
[
"bill_no",
"=",
bill_no,
],
[
"docstatus",
"!=",
2,
],
],
pluck="name",
limit=1,
)
return (
str(matches[0])
if matches
else ""
)
@frappe.whitelist()
def preview_purchase_invoice(
exchange_name: str,
) -> dict[str, Any]:
"""Prévisualiser une facture d’achat sans la créer."""
exchange = frappe.get_doc(
"Electronic Invoice Exchange",
exchange_name,
)
exchange.check_permission("read")
if exchange.direction != "Inbound":
frappe.throw(
_(
"La prévisualisation d’achat est "
"réservée aux échanges entrants."
)
)
if not exchange.semantic_payload_file:
frappe.throw(
_(
"Aucune donnée EN16931 n’est "
"archivée sur cet échange."
)
)
payload = _load_semantic_payload(
exchange.semantic_payload_file
)
try:
preview = build_purchase_invoice_preview(
payload,
supplier=exchange.supplier or "",
)
except PurchasePreviewError as error:
frappe.throw(str(error))
blockers: list[str] = []
warnings = list(
preview["calculation_warnings"]
)
if not exchange.supplier:
blockers.append(
"Aucun fournisseur ERPNext n’est rapproché."
)
for line in preview["lines"]:
resolved_uom = _resolve_uom(
line["unit_code"]
)
item_code, item_match_method = (
_suggest_item(
line["item_name"]
)
)
line["resolved_uom"] = resolved_uom
line["suggested_item_code"] = (
item_code
)
line["item_match_method"] = (
item_match_method
)
if not resolved_uom:
blockers.append(
(
f"Ligne {line['line_number']} : "
f"aucune UOM ERPNext ne correspond "
f"au code {line['unit_code'] or 'vide'}."
)
)
if not item_code:
blockers.append(
(
f"Ligne {line['line_number']} : "
f"aucun article ERPNext n’est "
f"rapproché avec « {line['item_name']} »."
)
)
for check in preview["checks"]:
if not check["ok"]:
blockers.append(
(
f"{check['label']} : "
f"attendu {check['expected']}, "
f"obtenu {check['actual']}."
)
)
existing_purchase_invoice = (
_find_existing_purchase_invoice(
supplier=exchange.supplier or "",
bill_no=preview["invoice_number"],
)
)
if existing_purchase_invoice:
blockers.append(
(
"Une facture d’achat ERPNext possède "
"déjà ce fournisseur et ce numéro."
)
)
preview.update(
{
"exchange": exchange.name,
"review_status": (
exchange.review_status
),
"original_file": (
exchange.payload_file
),
"semantic_file": (
exchange.semantic_payload_file
),
"existing_purchase_invoice": (
existing_purchase_invoice
),
"blockers": blockers,
"warnings": warnings,
"can_create_draft": (
not blockers
),
}
)
return preview
enuxia_einvoice/tests/test_purchase_preview.py
+185
View File
@@ -0,0 +1,185 @@
from __future__ import annotations
import copy
import unittest
from enuxia_einvoice.purchase_preview import (
build_purchase_invoice_preview,
)
class TestPurchaseInvoicePreview(
unittest.TestCase
):
def remote_invoice(self):
return {
"id": 79712,
"direction": "in",
"en_invoice": {
"number": (
"F20260619_123507_609"
),
"issue_date": "2025-06-30",
"payment_due_date": (
"2025-07-30"
),
"currency_code": "EUR",
"lines": [
{
"identifier": "001",
"invoiced_quantity": (
"28.5200"
),
"invoiced_quantity_code": (
"KGM"
),
"item_information": {
"name": (
"Poulet aux hormones"
),
},
"net_amount": "60.46",
"price_details": {
"item_net_price": (
"2.120000"
),
},
"vat_information": {
"invoiced_item_vat_category_code": (
"S"
),
"invoiced_item_vat_rate": (
"5.50"
),
},
},
{
"identifier": "002",
"invoiced_quantity": (
"1.0000"
),
"invoiced_quantity_code": (
"C62"
),
"item_information": {
"name": (
"Conseil en stratégie"
),
},
"net_amount": "1500.00",
"price_details": {
"item_net_price": (
"1500.000000"
),
},
"vat_information": {
"invoiced_item_vat_category_code": (
"S"
),
"invoiced_item_vat_rate": (
"20.00"
),
},
},
],
"totals": {
"total_without_vat": (
"1560.46"
),
"total_vat_amount": {
"value": "303.33",
"currency_code": "EUR",
},
"total_with_vat": (
"1863.79"
),
"amount_due_for_payment": (
"1863.79"
),
},
"vat_break_down": [
{
"vat_category_tax_amount": (
"3.33"
),
},
{
"vat_category_tax_amount": (
"300.00"
),
},
],
},
}
def test_build_preview(self) -> None:
preview = (
build_purchase_invoice_preview(
self.remote_invoice(),
supplier="Tricatel",
)
)
self.assertEqual(
preview["invoice_number"],
"F20260619_123507_609",
)
self.assertEqual(
preview["line_count"],
2,
)
self.assertEqual(
preview["totals"]["grand_total"],
"1863.79",
)
self.assertTrue(
all(
check["ok"]
for check in preview["checks"]
)
)
self.assertEqual(
preview["lines"][0][
"unit_code"
],
"KGM",
)
def test_detect_header_total_mismatch(
self,
) -> None:
invoice = copy.deepcopy(
self.remote_invoice()
)
invoice["en_invoice"]["totals"][
"total_without_vat"
] = "1600.00"
preview = (
build_purchase_invoice_preview(
invoice,
supplier="Tricatel",
)
)
checks = {
check["code"]: check["ok"]
for check in preview["checks"]
}
self.assertFalse(
checks["LINE_TOTAL"]
)
self.assertFalse(
checks["GRAND_TOTAL"]
)
if __name__ == "__main__":
unittest.main()