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

716 lines
23 KiB
Python
Raw Permalink 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 - 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