""" Fusion Accounting - Cross-Industry Invoice (CII) / Factur-X Generator & Parser Generates UN/CEFACT Cross-Industry Invoice (CII) compliant XML documents and supports embedding the XML inside a PDF/A-3 container to produce Factur-X / ZUGFeRD hybrid invoices. References ---------- * UN/CEFACT XML Schemas (D16B) https://unece.org/trade/uncefact/xml-schemas * Factur-X / ZUGFeRD specification https://fnfe-mpe.org/factur-x/ * EN 16931-1:2017 – European e-Invoicing semantic data model Namespace URIs used below are taken directly from the published UN/CEFACT schemas. Original implementation by Nexa Systems Inc. """ import io 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 (UN/CEFACT CII D16B) # ====================================================================== NS_RSM = ( "urn:un:unece:uncefact:data:standard:" "CrossIndustryInvoice:100" ) NS_RAM = ( "urn:un:unece:uncefact:data:standard:" "ReusableAggregateBusinessInformationEntity:100" ) NS_QDT = ( "urn:un:unece:uncefact:data:standard:" "QualifiedDataType:100" ) NS_UDT = ( "urn:un:unece:uncefact:data:standard:" "UnqualifiedDataType:100" ) NSMAP_CII = { "rsm": NS_RSM, "ram": NS_RAM, "qdt": NS_QDT, "udt": NS_UDT, } # Factur-X profile URNs FACTURX_PROFILES = { "minimum": ( "urn:factur-x.eu:1p0:minimum" ), "basicwl": ( "urn:factur-x.eu:1p0:basicwl" ), "basic": ( "urn:factur-x.eu:1p0:basic" ), "en16931": ( "urn:cen.eu:en16931:2017#compliant#" "urn:factur-x.eu:1p0:en16931" ), "extended": ( "urn:factur-x.eu:1p0:extended" ), } # CII type code mapping (UNTDID 1001) CII_TYPE_CODE_MAP = { "out_invoice": "380", # Commercial Invoice "out_refund": "381", # Credit Note "in_invoice": "380", "in_refund": "381", } class FusionCIIGenerator(models.AbstractModel): """ Generates and parses UN/CEFACT Cross-Industry Invoice documents and optionally embeds the XML within a PDF/A-3 container for Factur-X compliance. Implemented as an Odoo abstract model for ORM registry access. """ _name = "fusion.cii.generator" _description = "Fusion CII / Factur-X Generator" # ================================================================== # Public API # ================================================================== def generate_cii_invoice(self, move, profile="en16931"): """Build a CII XML document for a single ``account.move``. Args: move: An ``account.move`` singleton. profile (str): Factur-X conformance profile. One of ``'minimum'``, ``'basic'``, ``'en16931'`` (default), ``'extended'``. Returns: bytes: UTF-8 encoded CII XML. """ move.ensure_one() self._validate_move(move) root = etree.Element( f"{{{NS_RSM}}}CrossIndustryInvoice", nsmap=NSMAP_CII ) self._add_exchange_context(root, profile) header = self._add_header_trade(root, move) agreement = self._add_agreement_trade(root, move) delivery = self._add_delivery_trade(root, move) settlement = self._add_settlement_trade(root, move) self._add_line_items(root, move) return etree.tostring( root, xml_declaration=True, encoding="UTF-8", pretty_print=True, ) def parse_cii_invoice(self, xml_bytes): """Parse a CII XML document into an invoice values dictionary. Args: xml_bytes (bytes): Raw CII XML content. Returns: dict: Invoice values suitable for ``account.move.create()``. """ root = etree.fromstring(xml_bytes) ns = { "rsm": NS_RSM, "ram": NS_RAM, "udt": NS_UDT, } # Header header_path = ( "rsm:SupplyChainTradeTransaction/" "ram:ApplicableHeaderTradeSettlement" ) doc_path = ( "rsm:ExchangedDocument" ) ref = self._xpath_text(root, f"{doc_path}/ram:ID", ns) type_code = self._xpath_text(root, f"{doc_path}/ram:TypeCode", ns) issue_date = self._xpath_text( root, f"{doc_path}/ram:IssueDateTime/udt:DateTimeString", ns, ) currency = self._xpath_text( root, f"{header_path}/ram:InvoiceCurrencyCode", ns ) due_date = self._xpath_text( root, f"{header_path}/ram:SpecifiedTradePaymentTerms/" "ram:DueDateDateTime/udt:DateTimeString", ns, ) # Parties agreement_path = ( "rsm:SupplyChainTradeTransaction/" "ram:ApplicableHeaderTradeAgreement" ) supplier_name = self._xpath_text( root, f"{agreement_path}/ram:SellerTradeParty/ram:Name", ns, ) supplier_vat = self._xpath_text( root, f"{agreement_path}/ram:SellerTradeParty/" "ram:SpecifiedTaxRegistration/ram:ID", ns, ) customer_name = self._xpath_text( root, f"{agreement_path}/ram:BuyerTradeParty/ram:Name", ns, ) customer_vat = self._xpath_text( root, f"{agreement_path}/ram:BuyerTradeParty/" "ram:SpecifiedTaxRegistration/ram:ID", ns, ) # Lines line_path = ( "rsm:SupplyChainTradeTransaction/" "ram:IncludedSupplyChainTradeLineItem" ) line_nodes = root.findall(line_path, ns) lines = [] for ln in line_nodes: name = self._xpath_text( ln, "ram:SpecifiedTradeProduct/ram:Name", ns, ) or "" qty = float( self._xpath_text( ln, "ram:SpecifiedLineTradeDelivery/" "ram:BilledQuantity", ns, ) or "1" ) price = float( self._xpath_text( ln, "ram:SpecifiedLineTradeAgreement/" "ram:NetPriceProductTradePrice/" "ram:ChargeAmount", ns, ) or "0" ) lines.append({ "name": name, "quantity": qty, "price_unit": price, }) is_credit_note = type_code == "381" move_type = "out_refund" if is_credit_note else "out_invoice" # Normalise dates from CII format (YYYYMMDD) to ISO if issue_date and len(issue_date) == 8: issue_date = f"{issue_date[:4]}-{issue_date[4:6]}-{issue_date[6:]}" if due_date and len(due_date) == 8: due_date = f"{due_date[:4]}-{due_date[4:6]}-{due_date[6:]}" return { "move_type": move_type, "ref": ref, "invoice_date": issue_date, "invoice_date_due": due_date, "currency_id": currency, "supplier_name": supplier_name, "supplier_vat": supplier_vat, "customer_name": customer_name, "customer_vat": customer_vat, "invoice_line_ids": lines, } def embed_in_pdf(self, pdf_bytes, xml_bytes, profile="en16931"): """Embed CII XML into a PDF to produce a Factur-X / ZUGFeRD file. This creates a PDF/A-3 compliant document with the XML attached as an Associated File (AF) according to the Factur-X specification. Args: pdf_bytes (bytes): The original invoice PDF content. xml_bytes (bytes): The CII XML to embed. profile (str): Factur-X profile name for metadata. Returns: bytes: The resulting PDF/A-3 with embedded XML. Note: This method requires the ``pypdf`` library. If it is not installed the original PDF is returned unchanged with a warning logged. """ try: from pypdf import PdfReader, PdfWriter except ImportError: _log.warning( "pypdf is not installed; returning PDF without embedded XML. " "Install pypdf to enable Factur-X PDF/A-3 embedding." ) return pdf_bytes reader = PdfReader(io.BytesIO(pdf_bytes)) writer = PdfWriter() # Copy all pages from the source PDF for page in reader.pages: writer.add_page(page) # Copy metadata if reader.metadata: writer.add_metadata(reader.metadata) # Attach the XML as an embedded file writer.add_attachment( fname="factur-x.xml", data=xml_bytes, ) # Update document info with Factur-X conformance level profile_label = profile.upper() if profile != "en16931" else "EN 16931" writer.add_metadata({ "/Subject": f"Factur-X {profile_label}", }) output = io.BytesIO() writer.write(output) return output.getvalue() # ================================================================== # Internal – XML construction helpers # ================================================================== def _validate_move(self, move): """Ensure the move has the minimum data needed for CII export.""" if not move.partner_id: raise UserError( _("Cannot generate CII: invoice '%s' has no partner.", move.name or _("Draft")) ) def _add_exchange_context(self, root, profile): """Add ``ExchangedDocumentContext`` with the Factur-X profile.""" ram = NS_RAM rsm = NS_RSM ctx = self._sub(root, f"{{{rsm}}}ExchangedDocumentContext") guide = self._sub(ctx, f"{{{ram}}}GuidelineSpecifiedDocumentContextParameter") profile_urn = FACTURX_PROFILES.get(profile, FACTURX_PROFILES["en16931"]) self._sub(guide, f"{{{ram}}}ID", profile_urn) def _add_header_trade(self, root, move): """Add ``ExchangedDocument`` with ID, type code, and issue date.""" rsm = NS_RSM ram = NS_RAM udt = NS_UDT doc = self._sub(root, f"{{{rsm}}}ExchangedDocument") self._sub(doc, f"{{{ram}}}ID", move.name or "DRAFT") type_code = CII_TYPE_CODE_MAP.get(move.move_type, "380") self._sub(doc, f"{{{ram}}}TypeCode", type_code) issue_dt = self._sub(doc, f"{{{ram}}}IssueDateTime") issue_date = move.invoice_date or fields.Date.context_today(move) date_str_el = self._sub( issue_dt, f"{{{udt}}}DateTimeString", issue_date.strftime("%Y%m%d"), ) date_str_el.set("format", "102") if move.narration: import re plain = re.sub(r"<[^>]+>", "", move.narration) note = self._sub(doc, f"{{{ram}}}IncludedNote") self._sub(note, f"{{{ram}}}Content", plain) return doc def _add_agreement_trade(self, root, move): """Add ``ApplicableHeaderTradeAgreement`` with seller/buyer parties.""" rsm = NS_RSM ram = NS_RAM txn = root.find(f"{{{rsm}}}SupplyChainTradeTransaction") if txn is None: txn = self._sub(root, f"{{{rsm}}}SupplyChainTradeTransaction") agreement = self._sub(txn, f"{{{ram}}}ApplicableHeaderTradeAgreement") # Seller seller = self._sub(agreement, f"{{{ram}}}SellerTradeParty") self._add_trade_party(seller, move.company_id.partner_id, move.company_id) # Buyer buyer = self._sub(agreement, f"{{{ram}}}BuyerTradeParty") self._add_trade_party(buyer, move.partner_id) return agreement def _add_delivery_trade(self, root, move): """Add ``ApplicableHeaderTradeDelivery``.""" rsm = NS_RSM ram = NS_RAM udt = NS_UDT txn = root.find(f"{{{rsm}}}SupplyChainTradeTransaction") delivery = self._sub(txn, f"{{{ram}}}ApplicableHeaderTradeDelivery") # Actual delivery date (use invoice date as fallback) event = self._sub(delivery, f"{{{ram}}}ActualDeliverySupplyChainEvent") occ = self._sub(event, f"{{{ram}}}OccurrenceDateTime") del_date = move.invoice_date or fields.Date.context_today(move) date_el = self._sub( occ, f"{{{udt}}}DateTimeString", del_date.strftime("%Y%m%d") ) date_el.set("format", "102") return delivery def _add_settlement_trade(self, root, move): """Add ``ApplicableHeaderTradeSettlement`` with tax, totals, and terms.""" rsm = NS_RSM ram = NS_RAM udt = NS_UDT txn = root.find(f"{{{rsm}}}SupplyChainTradeTransaction") settlement = self._sub( txn, f"{{{ram}}}ApplicableHeaderTradeSettlement" ) currency = move.currency_id.name or "USD" self._sub(settlement, f"{{{ram}}}InvoiceCurrencyCode", currency) # Payment means pm = self._sub( settlement, f"{{{ram}}}SpecifiedTradeSettlementPaymentMeans" ) self._sub(pm, f"{{{ram}}}TypeCode", "30") # Credit transfer if move.partner_bank_id: account = self._sub( pm, f"{{{ram}}}PayeePartyCreditorFinancialAccount" ) self._sub( account, f"{{{ram}}}IBANID", move.partner_bank_id.acc_number or "", ) # Tax breakdown self._add_cii_tax(settlement, move, currency) # Payment terms if move.invoice_date_due: terms = self._sub( settlement, f"{{{ram}}}SpecifiedTradePaymentTerms" ) due_dt = self._sub(terms, f"{{{ram}}}DueDateDateTime") due_el = self._sub( due_dt, f"{{{udt}}}DateTimeString", move.invoice_date_due.strftime("%Y%m%d"), ) due_el.set("format", "102") # Monetary summation summation = self._sub( settlement, f"{{{ram}}}SpecifiedTradeSettlementHeaderMonetarySummation", ) self._monetary_sub( summation, f"{{{ram}}}LineTotalAmount", move.amount_untaxed, currency, ) self._monetary_sub( summation, f"{{{ram}}}TaxBasisTotalAmount", move.amount_untaxed, currency, ) self._monetary_sub( summation, f"{{{ram}}}TaxTotalAmount", move.amount_tax, currency, ) self._monetary_sub( summation, f"{{{ram}}}GrandTotalAmount", move.amount_total, currency, ) self._monetary_sub( summation, f"{{{ram}}}DuePayableAmount", move.amount_residual, currency, ) return settlement def _add_cii_tax(self, settlement, move, currency): """Add per-tax ``ApplicableTradeTax`` elements.""" ram = NS_RAM # Group 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) for inv_line in move.invoice_line_ids: for tax in inv_line.tax_ids: key = (tax.id, tax.name, tax.amount) if key in tax_groups: tax_groups[key]["base_amount"] += abs(inv_line.balance) for _key, data in tax_groups.items(): tax_el = self._sub(settlement, f"{{{ram}}}ApplicableTradeTax") self._monetary_sub( tax_el, f"{{{ram}}}CalculatedAmount", data["tax_amount"], currency, ) self._sub(tax_el, f"{{{ram}}}TypeCode", "VAT") self._monetary_sub( tax_el, f"{{{ram}}}BasisAmount", data["base_amount"], currency, ) self._sub( tax_el, f"{{{ram}}}CategoryCode", self._tax_category(data["tax"]), ) self._sub( tax_el, f"{{{ram}}}RateApplicablePercent", self._fmt(abs(data["tax"].amount)), ) def _add_line_items(self, root, move): """Append ``IncludedSupplyChainTradeLineItem`` elements.""" rsm = NS_RSM ram = NS_RAM txn = root.find(f"{{{rsm}}}SupplyChainTradeTransaction") 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 item_el = self._sub( txn, f"{{{ram}}}IncludedSupplyChainTradeLineItem" ) # Line document line_doc = self._sub( item_el, f"{{{ram}}}AssociatedDocumentLineDocument", ) self._sub(line_doc, f"{{{ram}}}LineID", str(idx)) # Product product = self._sub( item_el, f"{{{ram}}}SpecifiedTradeProduct" ) if line.product_id and line.product_id.default_code: self._sub( product, f"{{{ram}}}SellerAssignedID", line.product_id.default_code, ) self._sub( product, f"{{{ram}}}Name", line.name or line.product_id.name or _("(Unnamed)"), ) # Line trade agreement (price) line_agreement = self._sub( item_el, f"{{{ram}}}SpecifiedLineTradeAgreement" ) net_price = self._sub( line_agreement, f"{{{ram}}}NetPriceProductTradePrice", ) self._monetary_sub( net_price, f"{{{ram}}}ChargeAmount", line.price_unit, currency, ) # Line trade delivery (quantity) line_delivery = self._sub( item_el, f"{{{ram}}}SpecifiedLineTradeDelivery" ) qty_el = self._sub( line_delivery, f"{{{ram}}}BilledQuantity", self._fmt(line.quantity), ) qty_el.set("unitCode", self._uom_unece(line)) # Line trade settlement (tax, total) line_settlement = self._sub( item_el, f"{{{ram}}}SpecifiedLineTradeSettlement" ) for tax in line.tax_ids: trade_tax = self._sub( line_settlement, f"{{{ram}}}ApplicableTradeTax" ) self._sub(trade_tax, f"{{{ram}}}TypeCode", "VAT") self._sub( trade_tax, f"{{{ram}}}CategoryCode", self._tax_category(tax), ) self._sub( trade_tax, f"{{{ram}}}RateApplicablePercent", self._fmt(abs(tax.amount)), ) line_summation = self._sub( line_settlement, f"{{{ram}}}SpecifiedTradeSettlementLineMonetarySummation", ) self._monetary_sub( line_summation, f"{{{ram}}}LineTotalAmount", line.price_subtotal, currency, ) # ------------------------------------------------------------------ # Trade party helper # ------------------------------------------------------------------ def _add_trade_party(self, parent, partner, company=None): """Populate a trade party element with name, address, and tax ID.""" ram = NS_RAM self._sub( parent, f"{{{ram}}}Name", company.name if company else partner.name, ) # Postal address address = self._sub(parent, f"{{{ram}}}PostalTradeAddress") if partner.zip: self._sub(address, f"{{{ram}}}PostcodeCode", partner.zip) if partner.street: self._sub(address, f"{{{ram}}}LineOne", partner.street) if partner.street2: self._sub(address, f"{{{ram}}}LineTwo", partner.street2) if partner.city: self._sub(address, f"{{{ram}}}CityName", partner.city) if partner.country_id: self._sub( address, f"{{{ram}}}CountryID", partner.country_id.code ) # Tax registration vat = company.vat if company else partner.vat if vat: tax_reg = self._sub( parent, f"{{{ram}}}SpecifiedTaxRegistration" ) tax_id = self._sub(tax_reg, f"{{{ram}}}ID", vat) tax_id.set("schemeID", "VA") # ================================================================== # 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 _monetary_sub(parent, tag, value, currency): """Create a monetary amount sub-element with ``currencyID``.""" formatted = f"{float_round(float(value or 0), precision_digits=2):.2f}" el = etree.SubElement(parent, tag) el.text = formatted el.set("currencyID", currency) 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 CII/UBL tax category code (UNCL 5305).""" 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.""" uom = line.product_uom_id if not uom: return "C62" unece_code = getattr(uom, "unece_code", None) if unece_code: return unece_code 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