""" Fusion Accounting - Account Move EDI Extension Extends the ``account.move`` model with fields and methods for Electronic Data Interchange (EDI) document management. Adds an EDI tab to the invoice form, buttons to generate/export electronic documents, and an import wizard that can parse UBL 2.1 or CII XML files into new invoice records. Original implementation by Nexa Systems Inc. """ import base64 import logging from lxml import etree from odoo import api, fields, models, Command, _ from odoo.exceptions import UserError _log = logging.getLogger(__name__) class FusionAccountMoveEDI(models.Model): """ Adds EDI lifecycle tracking and import/export capabilities to journal entries. """ _inherit = "account.move" # ------------------------------------------------------------------ # Fields # ------------------------------------------------------------------ edi_document_ids = fields.One2many( comodel_name="fusion.edi.document", inverse_name="move_id", string="EDI Documents", copy=False, help="Electronic documents generated for this journal entry.", ) edi_document_count = fields.Integer( string="EDI Count", compute="_compute_edi_document_count", ) edi_state = fields.Selection( selection=[ ("to_send", "To Send"), ("sent", "Sent"), ("to_cancel", "To Cancel"), ("cancelled", "Cancelled"), ], string="EDI Status", compute="_compute_edi_state", store=True, help=( "Aggregate EDI state derived from linked EDI documents. " "Shows the most urgent state across all formats." ), ) edi_error_message = fields.Text( string="EDI Error", compute="_compute_edi_error_message", help="Concatenated error messages from all EDI documents.", ) edi_blocking_level = fields.Selection( selection=[ ("info", "Info"), ("warning", "Warning"), ("error", "Error"), ], string="EDI Error Severity", compute="_compute_edi_error_message", ) # ------------------------------------------------------------------ # Computed fields # ------------------------------------------------------------------ @api.depends("edi_document_ids") def _compute_edi_document_count(self): for move in self: move.edi_document_count = len(move.edi_document_ids) @api.depends( "edi_document_ids.state", "edi_document_ids.error_message", ) def _compute_edi_state(self): """Derive an aggregate state from all linked EDI documents. Priority order (highest urgency first): to_send > to_cancel > sent > cancelled If there are no EDI documents the field is left empty. """ priority = { "to_send": 0, "to_cancel": 1, "sent": 2, "cancelled": 3, } for move in self: docs = move.edi_document_ids if not docs: move.edi_state = False continue move.edi_state = min( docs.mapped("state"), key=lambda s: priority.get(s, 99), ) @api.depends("edi_document_ids.error_message", "edi_document_ids.blocking_level") def _compute_edi_error_message(self): for move in self: errors = move.edi_document_ids.filtered("error_message") if errors: move.edi_error_message = "\n".join( f"[{doc.edi_format_id.name}] {doc.error_message}" for doc in errors ) # Take the highest severity levels = errors.mapped("blocking_level") if "error" in levels: move.edi_blocking_level = "error" elif "warning" in levels: move.edi_blocking_level = "warning" else: move.edi_blocking_level = "info" else: move.edi_error_message = False move.edi_blocking_level = False # ------------------------------------------------------------------ # Button Actions # ------------------------------------------------------------------ def action_send_edi(self): """Create EDI documents for all active formats and send them. For each active ``fusion.edi.format`` that is applicable to this move, creates a ``fusion.edi.document`` in *to_send* state (if one does not already exist) and then triggers generation. """ self.ensure_one() if self.state != "posted": raise UserError( _("Only posted journal entries can generate EDI documents.") ) formats = self.env["fusion.edi.format"].search([ ("active", "=", True), ]) for fmt in formats: # Check applicability try: fmt._check_applicability(self) except UserError: continue existing = self.edi_document_ids.filtered( lambda d: d.edi_format_id == fmt and d.state != "cancelled" ) if existing: continue self.env["fusion.edi.document"].create({ "move_id": self.id, "edi_format_id": fmt.id, "state": "to_send", }) # Trigger generation on all pending documents pending = self.edi_document_ids.filtered( lambda d: d.state == "to_send" ) if pending: pending.action_send() def action_export_edi_xml(self): """Export the first available EDI attachment for download. Opens a download action for the XML file so the user can save it locally. Returns: dict: An ``ir.actions.act_url`` action pointing to the attachment download URL. """ self.ensure_one() sent_docs = self.edi_document_ids.filtered( lambda d: d.state == "sent" and d.attachment_id ) if not sent_docs: raise UserError( _("No sent EDI document with an attachment is available. " "Please generate EDI documents first.") ) attachment = sent_docs[0].attachment_id return { "type": "ir.actions.act_url", "url": f"/web/content/{attachment.id}?download=true", "target": "new", } def action_view_edi_documents(self): """Open the list of EDI documents for this journal entry. Returns: dict: A window action displaying related EDI documents. """ self.ensure_one() return { "type": "ir.actions.act_window", "name": _("EDI Documents"), "res_model": "fusion.edi.document", "domain": [("move_id", "=", self.id)], "view_mode": "list,form", "context": {"default_move_id": self.id}, } # ------------------------------------------------------------------ # Import # ------------------------------------------------------------------ def action_import_edi_xml(self): """Open a file upload wizard to import a UBL or CII XML invoice. Returns: dict: A window action for the import wizard. """ return { "type": "ir.actions.act_window", "name": _("Import EDI Invoice"), "res_model": "fusion.edi.import.wizard", "view_mode": "form", "target": "new", "context": { "default_move_type": self.env.context.get( "default_move_type", "out_invoice" ), }, } @api.model def create_invoice_from_xml(self, xml_bytes): """Parse an XML file (UBL or CII) and create an invoice. Auto-detects the XML format by inspecting the root element namespace. Args: xml_bytes (bytes): Raw XML content. Returns: account.move: The newly created invoice record. Raises: UserError: When the XML format is not recognised. """ root = etree.fromstring(xml_bytes) ns = etree.QName(root).namespace # Detect format if "CrossIndustryInvoice" in (ns or ""): fmt_code = "cii" parser = self.env["fusion.cii.generator"] values = parser.parse_cii_invoice(xml_bytes) elif "Invoice" in (ns or "") or "CreditNote" in (ns or ""): fmt_code = "ubl_21" parser = self.env["fusion.ubl.generator"] values = parser.parse_ubl_invoice(xml_bytes) else: raise UserError( _("Unrecognised XML format. Expected UBL 2.1 or CII.") ) return self._create_move_from_parsed(values, fmt_code) @api.model def _create_move_from_parsed(self, values, fmt_code): """Transform parsed EDI values into an ``account.move`` record. Handles partner lookup/creation, currency resolution, and line item creation. Args: values (dict): Parsed invoice data from a generator's ``parse_*`` method. fmt_code (str): The EDI format code for logging. Returns: account.move: The newly created draft invoice. """ Partner = self.env["res.partner"] Currency = self.env["res.currency"] # Resolve partner partner = Partner customer_vat = values.get("customer_vat") customer_name = values.get("customer_name") supplier_name = values.get("supplier_name") # For incoming invoices the "supplier" is our vendor if values.get("move_type") in ("in_invoice", "in_refund"): search_name = supplier_name search_vat = values.get("supplier_vat") else: search_name = customer_name search_vat = customer_vat if search_vat: partner = Partner.search([("vat", "=", search_vat)], limit=1) if not partner and search_name: partner = Partner.search( [("name", "ilike", search_name)], limit=1 ) # Resolve currency currency = Currency currency_code = values.get("currency_id") if currency_code: currency = Currency.search( [("name", "=", currency_code)], limit=1 ) # Build line commands line_commands = [] for line_vals in values.get("invoice_line_ids", []): line_commands.append(Command.create({ "name": line_vals.get("name", ""), "quantity": line_vals.get("quantity", 1), "price_unit": line_vals.get("price_unit", 0), })) move_vals = { "move_type": values.get("move_type", "out_invoice"), "ref": values.get("ref"), "invoice_date": values.get("invoice_date"), "invoice_date_due": values.get("invoice_date_due"), "invoice_line_ids": line_commands, } if partner: move_vals["partner_id"] = partner.id if currency: move_vals["currency_id"] = currency.id move = self.create(move_vals) _log.info( "Created invoice %s from %s XML import.", move.name or "(draft)", fmt_code, ) return move