""" 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 ```` or ```` 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