Files
Odoo-Modules/Fusion Accounting/models/ubl_generator.py
2026-02-22 01:22:18 -05:00

634 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
Fusion Accounting - UBL 2.1 Invoice / CreditNote Generator & Parser
Generates OASIS UBL 2.1 compliant XML documents from ``account.move``
records and parses incoming UBL XML back into invoice value dictionaries.
References
----------
* OASIS UBL 2.1 specification
https://docs.oasis-open.org/ubl/os-UBL-2.1/UBL-2.1.html
* Namespace URIs used below are taken directly from the published
OASIS schemas.
Original implementation by Nexa Systems Inc.
"""
import logging
from datetime import date
from lxml import etree
from odoo import api, fields, models, _
from odoo.exceptions import UserError
from odoo.tools import float_round
_log = logging.getLogger(__name__)
# ======================================================================
# XML Namespace constants (OASIS UBL 2.1)
# ======================================================================
NS_UBL_INVOICE = "urn:oasis:names:specification:ubl:schema:xsd:Invoice-2"
NS_UBL_CREDIT_NOTE = (
"urn:oasis:names:specification:ubl:schema:xsd:CreditNote-2"
)
NS_CAC = (
"urn:oasis:names:specification:ubl:schema:xsd:"
"CommonAggregateComponents-2"
)
NS_CBC = (
"urn:oasis:names:specification:ubl:schema:xsd:"
"CommonBasicComponents-2"
)
NSMAP_INVOICE = {
None: NS_UBL_INVOICE,
"cac": NS_CAC,
"cbc": NS_CBC,
}
NSMAP_CREDIT_NOTE = {
None: NS_UBL_CREDIT_NOTE,
"cac": NS_CAC,
"cbc": NS_CBC,
}
# UBL type code mapping (UNCL 1001)
MOVE_TYPE_CODE_MAP = {
"out_invoice": "380", # Commercial Invoice
"out_refund": "381", # Credit Note
"in_invoice": "380",
"in_refund": "381",
}
class FusionUBLGenerator(models.AbstractModel):
"""
Generates and parses OASIS UBL 2.1 Invoice and CreditNote documents.
This is implemented as an Odoo abstract model so it participates in
the ORM registry and has access to ``self.env`` for related lookups.
It does **not** create its own database table.
"""
_name = "fusion.ubl.generator"
_description = "Fusion UBL 2.1 Generator"
# ==================================================================
# Public API
# ==================================================================
def generate_ubl_invoice(self, move):
"""Build a UBL 2.1 XML document for a single ``account.move``.
Args:
move: An ``account.move`` singleton (invoice or credit note).
Returns:
bytes: UTF-8 encoded XML conforming to UBL 2.1.
Raises:
UserError: If required data is missing on the move.
"""
move.ensure_one()
self._validate_move(move)
is_credit_note = move.move_type in ("out_refund", "in_refund")
root = self._build_root_element(is_credit_note)
self._add_header(root, move, is_credit_note)
self._add_supplier_party(root, move)
self._add_customer_party(root, move)
self._add_payment_means(root, move)
self._add_tax_total(root, move)
self._add_legal_monetary_total(root, move, is_credit_note)
self._add_invoice_lines(root, move, is_credit_note)
return etree.tostring(
root,
xml_declaration=True,
encoding="UTF-8",
pretty_print=True,
)
def parse_ubl_invoice(self, xml_bytes):
"""Parse a UBL 2.1 Invoice or CreditNote XML into a values dict.
The returned dictionary is structured for direct use with
``account.move.create()`` (with minor caller-side adjustments
for partner resolution, etc.).
Args:
xml_bytes (bytes): Raw UBL XML.
Returns:
dict: Invoice values including ``invoice_line_ids`` as a
list of ``Command.create()`` tuples.
"""
root = etree.fromstring(xml_bytes)
# Detect document type from root tag
tag = etree.QName(root).localname
is_credit_note = tag == "CreditNote"
ns = {"cac": NS_CAC, "cbc": NS_CBC}
# ------ Header ------
invoice_date_str = self._xpath_text(root, "cbc:IssueDate", ns)
due_date_str = self._xpath_text(root, "cbc:DueDate", ns)
currency_code = self._xpath_text(
root, "cbc:DocumentCurrencyCode", ns
)
ref = self._xpath_text(root, "cbc:ID", ns)
# ------ Supplier / Customer ------
supplier_name = self._xpath_text(
root,
"cac:AccountingSupplierParty/cac:Party/cac:PartyName/cbc:Name",
ns,
)
supplier_vat = self._xpath_text(
root,
"cac:AccountingSupplierParty/cac:Party/"
"cac:PartyTaxScheme/cbc:CompanyID",
ns,
)
customer_name = self._xpath_text(
root,
"cac:AccountingCustomerParty/cac:Party/cac:PartyName/cbc:Name",
ns,
)
customer_vat = self._xpath_text(
root,
"cac:AccountingCustomerParty/cac:Party/"
"cac:PartyTaxScheme/cbc:CompanyID",
ns,
)
# ------ Lines ------
line_tag = "CreditNoteLine" if is_credit_note else "InvoiceLine"
line_nodes = root.findall(f"cac:{line_tag}", ns)
lines = []
for ln in line_nodes:
qty_tag = (
"cbc:CreditedQuantity" if is_credit_note
else "cbc:InvoicedQuantity"
)
lines.append({
"name": self._xpath_text(
ln, "cac:Item/cbc:Name", ns
) or "",
"quantity": float(
self._xpath_text(ln, qty_tag, ns) or "1"
),
"price_unit": float(
self._xpath_text(
ln, "cac:Price/cbc:PriceAmount", ns
) or "0"
),
})
move_type = "out_refund" if is_credit_note else "out_invoice"
return {
"move_type": move_type,
"ref": ref,
"invoice_date": invoice_date_str,
"invoice_date_due": due_date_str,
"currency_id": currency_code,
"supplier_name": supplier_name,
"supplier_vat": supplier_vat,
"customer_name": customer_name,
"customer_vat": customer_vat,
"invoice_line_ids": lines,
}
# ==================================================================
# Internal XML construction helpers
# ==================================================================
def _validate_move(self, move):
"""Ensure the move has the minimum data needed for UBL export."""
if not move.partner_id:
raise UserError(
_("Cannot generate UBL: invoice '%s' has no partner.",
move.name or _("Draft"))
)
if not move.company_id.vat and not move.company_id.company_registry:
_log.warning(
"Company '%s' has no VAT or registry number; "
"the UBL document may be rejected by recipients.",
move.company_id.name,
)
def _build_root_element(self, is_credit_note):
"""Create the root ``<Invoice>`` or ``<CreditNote>`` element."""
if is_credit_note:
return etree.Element(
f"{{{NS_UBL_CREDIT_NOTE}}}CreditNote",
nsmap=NSMAP_CREDIT_NOTE,
)
return etree.Element(
f"{{{NS_UBL_INVOICE}}}Invoice",
nsmap=NSMAP_INVOICE,
)
# ----- Header -----
def _add_header(self, root, move, is_credit_note):
"""Populate the document header (ID, dates, currency, etc.)."""
cbc = NS_CBC
self._sub(root, f"{{{cbc}}}UBLVersionID", "2.1")
self._sub(root, f"{{{cbc}}}CustomizationID",
"urn:cen.eu:en16931:2017")
self._sub(root, f"{{{cbc}}}ProfileID",
"urn:fdc:peppol.eu:2017:poacc:billing:01:1.0")
self._sub(root, f"{{{cbc}}}ID", move.name or "DRAFT")
issue_date = move.invoice_date or fields.Date.context_today(move)
self._sub(root, f"{{{cbc}}}IssueDate", str(issue_date))
if move.invoice_date_due:
self._sub(root, f"{{{cbc}}}DueDate", str(move.invoice_date_due))
type_code = MOVE_TYPE_CODE_MAP.get(move.move_type, "380")
self._sub(root, f"{{{cbc}}}InvoiceTypeCode", type_code)
if move.narration:
# Strip HTML tags for the Note element
import re
plain = re.sub(r"<[^>]+>", "", move.narration)
self._sub(root, f"{{{cbc}}}Note", plain)
self._sub(
root,
f"{{{cbc}}}DocumentCurrencyCode",
move.currency_id.name or "USD",
)
# ----- Parties -----
def _add_supplier_party(self, root, move):
"""Add ``AccountingSupplierParty`` from the company."""
cac, cbc = NS_CAC, NS_CBC
company = move.company_id
partner = company.partner_id
supplier = self._sub(root, f"{{{cac}}}AccountingSupplierParty")
party = self._sub(supplier, f"{{{cac}}}Party")
# Endpoint (electronic address)
if company.vat:
endpoint = self._sub(party, f"{{{cbc}}}EndpointID", company.vat)
endpoint.set("schemeID", "9925")
# Party name
party_name_el = self._sub(party, f"{{{cac}}}PartyName")
self._sub(party_name_el, f"{{{cbc}}}Name", company.name)
# Postal address
self._add_postal_address(party, partner)
# Tax scheme
if company.vat:
tax_scheme_el = self._sub(party, f"{{{cac}}}PartyTaxScheme")
self._sub(tax_scheme_el, f"{{{cbc}}}CompanyID", company.vat)
scheme = self._sub(tax_scheme_el, f"{{{cac}}}TaxScheme")
self._sub(scheme, f"{{{cbc}}}ID", "VAT")
# Legal entity
legal = self._sub(party, f"{{{cac}}}PartyLegalEntity")
self._sub(legal, f"{{{cbc}}}RegistrationName", company.name)
if company.company_registry:
self._sub(
legal, f"{{{cbc}}}CompanyID", company.company_registry
)
def _add_customer_party(self, root, move):
"""Add ``AccountingCustomerParty`` from the invoice partner."""
cac, cbc = NS_CAC, NS_CBC
partner = move.partner_id
customer = self._sub(root, f"{{{cac}}}AccountingCustomerParty")
party = self._sub(customer, f"{{{cac}}}Party")
# Endpoint
if partner.vat:
endpoint = self._sub(party, f"{{{cbc}}}EndpointID", partner.vat)
endpoint.set("schemeID", "9925")
# Party name
party_name_el = self._sub(party, f"{{{cac}}}PartyName")
self._sub(
party_name_el, f"{{{cbc}}}Name",
partner.commercial_partner_id.name or partner.name,
)
# Postal address
self._add_postal_address(party, partner)
# Tax scheme
if partner.vat:
tax_scheme_el = self._sub(party, f"{{{cac}}}PartyTaxScheme")
self._sub(tax_scheme_el, f"{{{cbc}}}CompanyID", partner.vat)
scheme = self._sub(tax_scheme_el, f"{{{cac}}}TaxScheme")
self._sub(scheme, f"{{{cbc}}}ID", "VAT")
# Legal entity
legal = self._sub(party, f"{{{cac}}}PartyLegalEntity")
self._sub(
legal, f"{{{cbc}}}RegistrationName",
partner.commercial_partner_id.name or partner.name,
)
def _add_postal_address(self, parent, partner):
"""Append a ``cac:PostalAddress`` block for the given partner."""
cac, cbc = NS_CAC, NS_CBC
address = self._sub(parent, f"{{{cac}}}PostalAddress")
if partner.street:
self._sub(address, f"{{{cbc}}}StreetName", partner.street)
if partner.street2:
self._sub(
address, f"{{{cbc}}}AdditionalStreetName", partner.street2
)
if partner.city:
self._sub(address, f"{{{cbc}}}CityName", partner.city)
if partner.zip:
self._sub(address, f"{{{cbc}}}PostalZone", partner.zip)
if partner.state_id:
self._sub(
address,
f"{{{cbc}}}CountrySubentity",
partner.state_id.name,
)
if partner.country_id:
country = self._sub(address, f"{{{cac}}}Country")
self._sub(
country,
f"{{{cbc}}}IdentificationCode",
partner.country_id.code,
)
# ----- Payment Means -----
def _add_payment_means(self, root, move):
"""Add ``PaymentMeans`` with bank account details if available."""
cac, cbc = NS_CAC, NS_CBC
pm = self._sub(root, f"{{{cac}}}PaymentMeans")
# Code 30 = Credit transfer (most common)
self._sub(pm, f"{{{cbc}}}PaymentMeansCode", "30")
if move.partner_bank_id:
account = self._sub(pm, f"{{{cac}}}PayeeFinancialAccount")
self._sub(
account, f"{{{cbc}}}ID",
move.partner_bank_id.acc_number or "",
)
if move.partner_bank_id.bank_id:
branch = self._sub(
account, f"{{{cac}}}FinancialInstitutionBranch"
)
self._sub(
branch, f"{{{cbc}}}ID",
move.partner_bank_id.bank_id.bic or "",
)
# ----- Tax Total -----
def _add_tax_total(self, root, move):
"""Add ``TaxTotal`` with per-tax breakdown."""
cac, cbc = NS_CAC, NS_CBC
currency = move.currency_id.name or "USD"
tax_total = self._sub(root, f"{{{cac}}}TaxTotal")
tax_amount_el = self._sub(
tax_total, f"{{{cbc}}}TaxAmount",
self._fmt(move.amount_tax),
)
tax_amount_el.set("currencyID", currency)
# Build per-tax subtotals from the tax lines
tax_groups = {}
for line in move.line_ids.filtered(
lambda l: l.tax_line_id and l.tax_line_id.amount_type != "group"
):
tax = line.tax_line_id
key = (tax.id, tax.name, tax.amount)
if key not in tax_groups:
tax_groups[key] = {
"tax": tax,
"tax_amount": 0.0,
"base_amount": 0.0,
}
tax_groups[key]["tax_amount"] += abs(line.balance)
# Compute base amounts from invoice lines
for line in move.invoice_line_ids:
for tax in line.tax_ids:
key = (tax.id, tax.name, tax.amount)
if key in tax_groups:
tax_groups[key]["base_amount"] += abs(line.balance)
for _key, data in tax_groups.items():
subtotal = self._sub(tax_total, f"{{{cac}}}TaxSubtotal")
taxable_el = self._sub(
subtotal, f"{{{cbc}}}TaxableAmount",
self._fmt(data["base_amount"]),
)
taxable_el.set("currencyID", currency)
sub_tax_el = self._sub(
subtotal, f"{{{cbc}}}TaxAmount",
self._fmt(data["tax_amount"]),
)
sub_tax_el.set("currencyID", currency)
cat = self._sub(subtotal, f"{{{cac}}}TaxCategory")
self._sub(cat, f"{{{cbc}}}ID", self._tax_category(data["tax"]))
self._sub(
cat, f"{{{cbc}}}Percent", self._fmt(abs(data["tax"].amount))
)
scheme = self._sub(cat, f"{{{cac}}}TaxScheme")
self._sub(scheme, f"{{{cbc}}}ID", "VAT")
# ----- Legal Monetary Total -----
def _add_legal_monetary_total(self, root, move, is_credit_note):
"""Add ``LegalMonetaryTotal`` summarising the invoice amounts."""
cac, cbc = NS_CAC, NS_CBC
currency = move.currency_id.name or "USD"
lmt = self._sub(root, f"{{{cac}}}LegalMonetaryTotal")
line_ext_el = self._sub(
lmt, f"{{{cbc}}}LineExtensionAmount",
self._fmt(move.amount_untaxed),
)
line_ext_el.set("currencyID", currency)
tax_excl_el = self._sub(
lmt, f"{{{cbc}}}TaxExclusiveAmount",
self._fmt(move.amount_untaxed),
)
tax_excl_el.set("currencyID", currency)
tax_incl_el = self._sub(
lmt, f"{{{cbc}}}TaxInclusiveAmount",
self._fmt(move.amount_total),
)
tax_incl_el.set("currencyID", currency)
payable_el = self._sub(
lmt, f"{{{cbc}}}PayableAmount",
self._fmt(move.amount_residual),
)
payable_el.set("currencyID", currency)
# ----- Invoice Lines -----
def _add_invoice_lines(self, root, move, is_credit_note):
"""Append one ``InvoiceLine`` / ``CreditNoteLine`` per product line."""
cac, cbc = NS_CAC, NS_CBC
currency = move.currency_id.name or "USD"
for idx, line in enumerate(move.invoice_line_ids, start=1):
if line.display_type in ("line_section", "line_note"):
continue
line_tag = (
f"{{{cac}}}CreditNoteLine"
if is_credit_note
else f"{{{cac}}}InvoiceLine"
)
inv_line = self._sub(root, line_tag)
self._sub(inv_line, f"{{{cbc}}}ID", str(idx))
qty_tag = (
f"{{{cbc}}}CreditedQuantity"
if is_credit_note
else f"{{{cbc}}}InvoicedQuantity"
)
qty_el = self._sub(
inv_line, qty_tag, self._fmt(line.quantity)
)
qty_el.set("unitCode", self._uom_unece(line))
ext_el = self._sub(
inv_line, f"{{{cbc}}}LineExtensionAmount",
self._fmt(line.price_subtotal),
)
ext_el.set("currencyID", currency)
# Per-line tax info
for tax in line.tax_ids:
tax_el = self._sub(inv_line, f"{{{cac}}}TaxTotal")
tax_amt_el = self._sub(
tax_el, f"{{{cbc}}}TaxAmount",
self._fmt(line.price_total - line.price_subtotal),
)
tax_amt_el.set("currencyID", currency)
# Item
item = self._sub(inv_line, f"{{{cac}}}Item")
self._sub(
item, f"{{{cbc}}}Name",
line.name or line.product_id.name or _("(Unnamed)"),
)
if line.product_id and line.product_id.default_code:
seller_id_el = self._sub(
item, f"{{{cac}}}SellersItemIdentification"
)
self._sub(
seller_id_el, f"{{{cbc}}}ID",
line.product_id.default_code,
)
# Classified tax category per line item
for tax in line.tax_ids:
ctc = self._sub(item, f"{{{cac}}}ClassifiedTaxCategory")
self._sub(
ctc, f"{{{cbc}}}ID", self._tax_category(tax)
)
self._sub(
ctc, f"{{{cbc}}}Percent", self._fmt(abs(tax.amount))
)
ts = self._sub(ctc, f"{{{cac}}}TaxScheme")
self._sub(ts, f"{{{cbc}}}ID", "VAT")
# Price
price_el = self._sub(inv_line, f"{{{cac}}}Price")
price_amt_el = self._sub(
price_el, f"{{{cbc}}}PriceAmount",
self._fmt(line.price_unit),
)
price_amt_el.set("currencyID", currency)
# ==================================================================
# Utility helpers
# ==================================================================
@staticmethod
def _sub(parent, tag, text=None):
"""Create a sub-element, optionally setting its text content."""
el = etree.SubElement(parent, tag)
if text is not None:
el.text = str(text)
return el
@staticmethod
def _fmt(value, precision=2):
"""Format a numeric value with the given decimal precision."""
return f"{float_round(float(value or 0), precision_digits=precision):.{precision}f}"
@staticmethod
def _tax_category(tax):
"""Map an Odoo tax to a UBL tax category code (UNCL 5305).
Standard rate -> S, Zero -> Z, Exempt -> E, Reverse charge -> AE.
"""
amount = abs(tax.amount)
if amount == 0:
return "Z"
tax_name_lower = (tax.name or "").lower()
if "exempt" in tax_name_lower:
return "E"
if "reverse" in tax_name_lower:
return "AE"
return "S"
@staticmethod
def _uom_unece(line):
"""Return the UN/ECE Rec 20 unit code for the invoice line.
Falls back to ``C62`` (one / unit) when no mapping exists.
"""
uom = line.product_uom_id
if not uom:
return "C62"
unece_code = getattr(uom, "unece_code", None)
if unece_code:
return unece_code
# Heuristic fallback for common units
name = (uom.name or "").lower()
mapping = {
"unit": "C62",
"units": "C62",
"piece": "C62",
"pieces": "C62",
"pce": "C62",
"kg": "KGM",
"kilogram": "KGM",
"g": "GRM",
"gram": "GRM",
"l": "LTR",
"liter": "LTR",
"litre": "LTR",
"m": "MTR",
"meter": "MTR",
"metre": "MTR",
"hour": "HUR",
"hours": "HUR",
"day": "DAY",
"days": "DAY",
}
return mapping.get(name, "C62")
@staticmethod
def _xpath_text(node, xpath, ns):
"""Return the text of the first matching element, or ``None``."""
found = node.find(xpath, ns)
return found.text if found is not None else None