353 lines
12 KiB
Python
353 lines
12 KiB
Python
"""
|
|
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
|