import io
import logging
import re
from collections import defaultdict
from markupsafe import Markup
from stdnum.be import vat as be_vat
from odoo import Command, _, api, fields, models
from odoo.exceptions import UserError
from odoo.fields import Domain
from odoo.tools import formatLang, frozendict, html2plaintext, html_escape, pdf, unique
from odoo.addons.account_edi_ubl_cii.models.account_edi_common import (
EAS_MAPPING,
FloatFmt,
GST_COUNTRY_CODES,
UOM_TO_UNECE_CODE,
)
from odoo.addons.base.models.res_partner_bank import sanitize_account_number
from odoo.addons.account_edi_ubl_cii.tools.ubl_20_optional_fields import PEPPOL_INVOICE_OPTIONAL_FIELDS, PEPPOL_INVOICE_OPTIONAL_LINE_FIELDS, PEPPOL_CREDIT_NOTE_OPTIONAL_FIELDS, PEPPOL_CREDIT_NOTE_OPTIONAL_LINE_FIELDS
from odoo.addons.account_edi_ubl_cii.tools import Invoice, CreditNote, DebitNote
_logger = logging.getLogger(__name__)
class AccountEdiUBL(models.AbstractModel):
_name = "account.edi.ubl"
_inherit = 'account.edi.common'
_description = "Base helpers for UBL"
# -------------------------------------------------------------------------
# BASE LINES HELPERS
# -------------------------------------------------------------------------
def _ubl_is_recycling_contribution_tax(self, tax_data):
""" Indicate if the 'tax_data' passed as parameter is a recycling contribution tax.
:param tax_data: One of the tax data in base_line['tax_details']['taxes_data'].
:return: True if tax_data['tax'] is a recycling contribution tax, False otherwise.
"""
if not tax_data:
return False
tax = tax_data['tax']
return tax.amount_type == 'fixed' and tax.include_base_amount
def _ubl_is_excise_tax(self, tax_data):
""" Indicate if the 'tax_data' passed as parameter is an excise tax.
:param tax_data: One of the tax data in base_line['tax_details']['taxes_data'].
:return: True if tax_data['tax'] is an excise tax, False otherwise.
"""
if not tax_data:
return False
tax = tax_data['tax']
return tax.amount_type == 'code' and tax.include_base_amount
def _ubl_is_reverse_charge_tax(self, tax_data):
""" Indicate if the 'tax_data' passed as parameter is an intracommunity reverse charge purchase tax.
:param tax_data: One of the tax data in base_line['tax_details']['taxes_data'].
:return: True if tax_data['tax'] is an intracommunity reverse charge purchase tax, False otherwise.
"""
if not tax_data:
return False
tax = tax_data['tax']
return tax.amount_type == 'percent' and tax.has_negative_factor
def _ubl_is_early_payment_base_line(self, base_line):
""" Indicate if the 'base_line' passed as parameter has been generated by an 'mixed' early payment.
:param base_line: A base line (see '_prepare_base_line_for_taxes_computation').
:return: True if the 'base_line' is a 'mixed' early payment line, False otherwise.
"""
return base_line['special_type'] == 'early_payment'
def _ubl_is_global_discount_base_line(self, base_line):
""" Indicate if the 'base_line' passed as parameter is a global discount line.
:param base_line: A base line (see '_prepare_base_line_for_taxes_computation').
:return: True if the 'base_line' is a global discount line, False otherwise.
"""
return base_line['special_type'] == 'global_discount'
def _ubl_is_cash_rounding_base_line(self, base_line):
""" Indicate if the 'base_line' passed as parameter has been generated by a cash rounding method.
:param base_line: A base line (see '_prepare_base_line_for_taxes_computation').
:return: True if the 'base_line' is a cash rounding line, False otherwise.
"""
return base_line['special_type'] == 'cash_rounding'
def _ubl_default_tax_category_grouping_key(self, base_line, tax_data, vals, currency):
""" Give the values about the tax category for a given tax.
:param base_line: A base line (see '_prepare_base_line_for_taxes_computation').
:param tax_data: One of the tax data in base_line['tax_details']['taxes_data'].
:param vals: Some custom data.
:param currency: The currency for which the grouping key is expressed.
:return: A dictionary that could be used as a grouping key for the taxes helpers.
"""
customer = vals['customer']
supplier = vals['supplier']
if tax_data and (
tax_data['tax'].amount_type != 'percent'
or self._ubl_is_recycling_contribution_tax(tax_data)
or self._ubl_is_excise_tax(tax_data)
):
return
else:
supplier_country_code = supplier.commercial_partner_id.country_id.code
if supplier_country_code in GST_COUNTRY_CODES:
scheme_id = 'GST'
else:
scheme_id = 'VAT'
if self._ubl_is_reverse_charge_tax(tax_data):
# Reverse-charge taxes with +100/-100% repartition lines are used in vendor bills.
# In self-billed invoices, we report them from the seller's perspective, as 0% taxes.
tax = tax_data['tax']
return {
'tax_category_code': self._get_tax_category_code(customer.commercial_partner_id, supplier, tax),
**self._get_tax_exemption_reason(customer.commercial_partner_id, supplier, tax),
'percent': 0.0,
'scheme_id': scheme_id,
'is_withholding': False,
'currency': currency,
}
elif tax_data:
tax = tax_data['tax']
tax_category_code = self._get_tax_category_code(customer.commercial_partner_id, supplier, tax)
if tax_category_code == 'O':
# [BR-O-05] An Invoice line (BG-25) where the VAT category code (BT-151) is
# "Not subject to VAT" shall not contain an Invoiced item VAT rate (BT-152).
# [BR-O-06] A Document level allowance (BG-20) where VAT category code (BT-95) is
# "Not subject to VAT" shall not contain a Document level allowance VAT rate (BT-96).
# [BR-O-07] A Document level charge (BG-21) where the VAT category code (BT-102) is
# "Not subject to VAT" shall not contain a Document level charge VAT rate (BT-103).
percent = None
else:
percent = tax.amount if not tax.has_negative_factor else 0.0
return {
'tax_category_code': tax_category_code,
**self.with_context(tax_exemption_reason_invoice=vals.get('invoice'))._get_tax_exemption_reason(customer.commercial_partner_id, supplier, tax),
'percent': percent,
'scheme_id': scheme_id,
'is_withholding': tax.amount < 0.0,
'currency': currency,
}
else:
return {
'tax_category_code': self._get_tax_category_code(customer.commercial_partner_id, supplier, self.env['account.tax']),
**self._get_tax_exemption_reason(customer.commercial_partner_id, supplier, self.env['account.tax']),
'percent': 0.0,
'scheme_id': scheme_id,
'is_withholding': False,
'currency': currency,
}
def _ubl_default_tax_subtotal_tax_category_grouping_key(self, tax_grouping_key, vals):
""" Give the values about how taxes are grouped together in TaxTotal -> TaxSubtotal -> TaxCategory
(or WithholdingTaxTotal depending on 'is_withholding').
:param tax_grouping_key: The grouping key returned by '_ubl_default_tax_category_grouping_key'.
:param vals: Some custom data.
:return: A dictionary that could be used as a grouping key for the taxes helpers.
"""
return dict(tax_grouping_key)
def _ubl_default_tax_subtotal_grouping_key(self, tax_category_grouping_key, vals):
""" Give the values about how taxes are grouped together in TaxTotal -> TaxSubtotal
(or WithholdingTaxTotal depending on 'is_withholding').
:param tax_category_grouping_key: The grouping key returned by '_ubl_default_tax_subtotal_tax_category_grouping_key'.
:param vals: Some custom data.
:return: A dictionary that could be used as a grouping key for the taxes helpers.
"""
return dict(tax_category_grouping_key)
def _ubl_default_tax_total_grouping_key(self, tax_subtotal_grouping_key, vals):
""" Give the values about how taxes are grouped together in TaxTotal
(or WithholdingTaxTotal depending on 'is_withholding').
:param tax_subtotal_grouping_key: The grouping key returned by '_ubl_default_tax_subtotal_grouping_key'.
:param vals: Some custom data.
:return: A dictionary that could be used as a grouping key for the taxes helpers.
"""
return {
'is_withholding': tax_subtotal_grouping_key['is_withholding'],
'currency': tax_subtotal_grouping_key['currency'],
}
def _ubl_default_allowance_charge_early_payment_grouping_key(self, base_line, tax_data, vals, currency):
""" Give the grouping key when generating the allowance/charge from an early payment discount.
:param base_line: A base line (see '_prepare_base_line_for_taxes_computation').
:param tax_data: One of the tax data in base_line['tax_details']['taxes_data'].
:param vals: Some custom data.
:param currency: The currency for which the grouping key is expressed.
:return: A dictionary that could be used as a grouping key for the taxes helpers.
"""
if not self._ubl_is_early_payment_base_line(base_line):
return
tax_grouping_key = self._ubl_default_tax_category_grouping_key(base_line, tax_data, vals, currency)
if not tax_grouping_key or tax_grouping_key['is_withholding']:
return
# We do not want to group the positive and negative lines together;
# the following changes the grouping key to be like in 18.0 to 18.3.
if tax_grouping_key['tax_category_code'] == 'E' and tax_data and tax_data['tax']:
tax_grouping_key['tax_exemption_reason'] = None
return tax_grouping_key
def _ubl_default_allowance_charge_global_discount_grouping_key(self, base_line, tax_data, vals, currency):
""" Give the grouping key when generating the allowance/charge from a global discount line.
:param base_line: A base line (see '_prepare_base_line_for_taxes_computation').
:param tax_data: One of the tax data in base_line['tax_details']['taxes_data'].
:param vals: Some custom data.
:param currency: The currency for which the grouping key is expressed.
:return: A dictionary that could be used as a grouping key for the taxes helpers.
"""
if not self._ubl_is_global_discount_base_line(base_line):
return
tax_grouping_key = self._ubl_default_tax_category_grouping_key(base_line, tax_data, vals, currency)
if not tax_grouping_key or tax_grouping_key['is_withholding']:
return
return tax_grouping_key
def _ubl_default_payable_amount_tax_withholding_grouping_key(self, base_line, tax_data, vals, currency):
""" Give the grouping key when moving the tax withholding amounts to PrepaidAmount.
:param base_line: A base line (see '_prepare_base_line_for_taxes_computation').
:param tax_data: One of the tax data in base_line['tax_details']['taxes_data'].
:param vals: Some custom data.
:param currency: The currency for which the grouping key is expressed.
:return: A dictionary that could be used as a grouping key for the taxes helpers.
"""
if not tax_data:
return
tax_grouping_key = self._ubl_default_tax_category_grouping_key(base_line, tax_data, vals, currency)
if not tax_grouping_key:
return
return tax_grouping_key['is_withholding']
def _ubl_default_base_line_item_classified_tax_category_grouping_key(self, base_line, tax_data, vals, currency):
""" Give the grouping key when computing taxes for Item -> ClassifiedTaxCategory.
:param base_line: A base line (see '_prepare_base_line_for_taxes_computation').
:param tax_data: One of the tax data in base_line['tax_details']['taxes_data'].
:param vals: Some custom data.
:param currency: The currency for which the grouping key is expressed.
:return: A dictionary that could be used as a grouping key for the taxes helpers.
"""
tax_grouping_key = self._ubl_default_tax_category_grouping_key(base_line, tax_data, vals, currency)
if not tax_grouping_key or tax_grouping_key['is_withholding']:
return
return tax_grouping_key
def _ubl_turn_base_lines_price_unit_as_always_positive(self, vals):
""" Helper to make sure the base_lines don't contain any negative price_unit.
:param vals: Some custom data.
"""
for base_line in vals['base_lines']:
if base_line['price_unit'] < 0.0:
base_line['quantity'] *= -1
base_line['price_unit'] *= -1
def _ubl_turn_emptying_taxes_as_new_base_lines(self, base_lines, company, vals):
""" Extract emptying taxes such as "Vidanges" on bottles from the current base lines and turn them into
additional base lines.
:param base_lines: The original 'base_lines' of the document.
:param company: The company owning the 'base_lines'.
:param vals: Some custom data.
"""
AccountTax = self.env['account.tax']
def exclude_function(base_line, tax_data):
if not tax_data:
return
tax = tax_data['tax']
return tax.amount_type in ('fixed', 'code') and not tax.include_base_amount
new_base_lines = AccountTax._dispatch_taxes_into_new_base_lines(base_lines, company, exclude_function)
# fixed tax are not affected by discount, so the removed_tax_data_base_lines should get their discount remove
# to have the total equal to the fixed tax
for new_base_line in new_base_lines:
for removed_taxes_data_base_line in new_base_line['removed_taxes_data_base_lines']:
removed_taxes_data_base_line['discount'] = 0
def aggregate_function(target_base_line, base_line):
target_base_line.setdefault('_aggregated_quantity', 0.0)
target_base_line['_aggregated_quantity'] += base_line['quantity']
def grouping_function(base_line):
return {'tax': base_line['_removed_tax_data']['tax']}
extra_base_lines = AccountTax._turn_removed_taxes_into_new_base_lines(
base_lines=new_base_lines,
company=company,
grouping_function=grouping_function,
aggregate_function=aggregate_function,
)
# Restore back the values per quantity.
for base_line in extra_base_lines:
base_line['quantity'] = base_line['_aggregated_quantity']
if base_line['_aggregated_quantity']:
base_line['price_unit'] /= base_line['_aggregated_quantity']
base_line['product_id'] = self.env['product.product']
return new_base_lines + extra_base_lines
# -------------------------------------------------------------------------
# EXPORT: Collecting data
# -------------------------------------------------------------------------
def _ubl_add_values_company(self, vals, company):
vals['company'] = company
def _ubl_add_values_currency(self, vals, currency):
vals['currency'] = currency
def _ubl_add_values_supplier(self, vals, supplier):
vals['supplier'] = supplier
def _ubl_add_values_customer(self, vals, customer):
vals['customer'] = customer
def _ubl_add_values_delivery(self, vals, delivery):
vals['delivery'] = delivery
# -------------------------------------------------------------------------
# EXPORT: Building nodes
# -------------------------------------------------------------------------
def _ubl_add_line_id_node(self, vals):
vals['line_node']['cbc:ID'] = {'_text': vals['line_vals']['index']}
def _ubl_add_line_note_nodes(self, vals):
vals['line_node']['cbc:Note'] = []
def _ubl_add_line_quantity_node(self, vals):
base_line = vals['line_vals']['base_line']
vals['line_node']['cbc:Quantity'] = {
'_text': base_line['quantity'],
'unitCode': self._get_uom_unece_code(base_line['product_uom_id']),
}
def _ubl_add_line_invoiced_quantity_node(self, vals):
base_line = vals['line_vals']['base_line']
vals['line_node']['cbc:InvoicedQuantity'] = {
'_text': base_line['quantity'],
'unitCode': self._get_uom_unece_code(base_line['product_uom_id']),
}
def _ubl_add_line_credited_quantity_node(self, vals):
base_line = vals['line_vals']['base_line']
vals['line_node']['cbc:CreditedQuantity'] = {
'_text': base_line['quantity'],
'unitCode': self._get_uom_unece_code(base_line['product_uom_id']),
}
def _ubl_add_line_debited_quantity_node(self, vals):
base_line = vals['line_vals']['base_line']
vals['line_node']['cbc:DebitedQuantity'] = {
'_text': base_line['quantity'],
'unitCode': self._get_uom_unece_code(base_line['product_uom_id']),
}
def _ubl_add_line_item_name_description_nodes(self, vals):
item_node = vals['item_node']
base_line = vals['line_vals']['base_line']
line_name = name = base_line.get('name') or '' # Regular business line.
description = None
if product := base_line['product_id']:
name = product.display_name
description = line_name.replace(name, '').strip() # Remove the redundant product's name from the description.
elif base_line.get('_removed_tax_data'):
# Emptying tax extra line.
name = base_line['_removed_tax_data']['tax'].name
if description:
item_node['cbc:Description'] = {'_text': description}
else:
item_node['cbc:Description'] = None
if name:
item_node['cbc:Name'] = {'_text': name}
else:
item_node['cbc:Name'] = None
def _ubl_add_line_item_identification_nodes(self, vals):
item_node = vals['item_node']
base_line = vals['line_vals']['base_line']
product = base_line['product_id']
item_node['cac:BuyersItemIdentification'] = {}
if product.default_code:
item_node['cac:SellersItemIdentification'] = {
'cbc:ID': {'_text': product.default_code},
}
else:
item_node['cac:SellersItemIdentification'] = None
if product.barcode:
item_node['cac:StandardItemIdentification'] = {
'cbc:ID': {
'_text': product.barcode,
'schemeID': '0160', # GTIN
},
}
else:
item_node['cac:StandardItemIdentification'] = None
def _ubl_add_line_item_additional_item_property_nodes(self, vals):
item_node = vals['item_node']
base_line = vals['line_vals']['base_line']
product = base_line['product_id']
item_node['cac:AdditionalItemProperty'] = [
{
'cbc:Name': {'_text': value.attribute_id.name},
'cbc:Value': {'_text': value.name},
}
for value in product.product_template_attribute_value_ids
]
def _ubl_get_line_item_commodity_classification_node_from_intrastat_code(self, vals, intrastat_code):
return {
'cbc:ItemClassificationCode': {
'_text': intrastat_code.code,
'listID': 'HS',
'listVersionID': None,
}
}
def _ubl_get_line_item_commodity_classification_node_from_unspsc_code(self, vals, unspsc_code):
return {
'cbc:ItemClassificationCode': {
'_text': unspsc_code.code,
'listID': 'TST',
'listVersionID': None,
}
}
def _ubl_get_line_item_commodity_classification_node_from_cpv_code(self, vals, cpv_code):
return {
'cbc:ItemClassificationCode': {
'_text': cpv_code.code,
'listID': 'STI',
'listVersionID': None,
}
}
def _ubl_get_line_item_commodity_classification_node_from_cg_code(self, vals, cg_code):
return {
'cbc:ItemClassificationCode': {
'_text': cg_code.name,
'listID': 'CG',
'listVersionID': None,
}
}
def _ubl_add_line_item_commodity_classification_nodes(self, vals):
item_node = vals['item_node']
base_line = vals['line_vals']['base_line']
product = base_line['product_id']
nodes = item_node['cac:CommodityClassification'] = []
if self.module_installed('account_intrastat'):
intrastat_code = product.intrastat_code_id
if intrastat_code.code:
nodes.append(self._ubl_get_line_item_commodity_classification_node_from_intrastat_code(vals, intrastat_code))
if self.module_installed('product_unspsc'):
unspsc_code = product.unspsc_code_id
if unspsc_code.code:
nodes.append(self._ubl_get_line_item_commodity_classification_node_from_unspsc_code(vals, unspsc_code))
if self.module_installed('l10n_ro_edi'):
cpv_code = product.cpv_code_id
if cpv_code.code:
nodes.append(self._ubl_get_line_item_commodity_classification_node_from_cpv_code(vals, cpv_code))
if self.module_installed('l10n_hr_edi'):
cg_code = base_line.get('cg_item_classification_code') or product.l10n_hr_kpd_category_id
if cg_code.name:
nodes.append(self._ubl_get_line_item_commodity_classification_node_from_cg_code(vals, cg_code))
return nodes
def _ubl_get_line_item_node_classified_tax_category_node(self, vals, tax_category):
""" Generate the node 'cac:ClassifiedTaxCategory' in 'cac:Item'.
:param vals: Some custom data.
:param tax_category: An entry of vals['_ubl_values']['item_classified_tax_categories']
containing all the necessary data to build the node.
:return: A new node in 'cac:Item' -> 'cac:ClassifiedTaxCategory'.
"""
return {
'_currency': tax_category['currency'],
'cbc:ID': {'_text': tax_category['tax_category_code']},
'cbc:Name': {'_text': None},
'cbc:Percent': {'_text': tax_category['percent']},
'cbc:TaxExemptionReasonCode': {'_text': None},
'cbc:TaxExemptionReason': {'_text': None},
'cac:TaxScheme': {
'cbc:ID': {'_text': tax_category['scheme_id']},
}
}
def _ubl_add_line_item_classified_tax_category_nodes(self, vals, in_foreign_currency=True):
AccountTax = self.env['account.tax']
item_node = vals['item_node']
base_line = vals['line_vals']['base_line']
currency = base_line['currency_id'] if in_foreign_currency else vals['company_currency']
suffix = '_currency' if in_foreign_currency else ''
classified_tax_category_nodes = item_node['cac:ClassifiedTaxCategory'] = []
aggregated_values = AccountTax._aggregate_base_line_tax_details(
base_line=base_line,
grouping_function=lambda base_line, tax_data: self._ubl_default_base_line_item_classified_tax_category_grouping_key(
base_line=base_line,
tax_data=tax_data,
vals=vals,
currency=currency,
),
)
for grouping_key, values in aggregated_values.items():
if not grouping_key:
continue
classified_tax_category_nodes.append(self._ubl_get_line_item_node_classified_tax_category_node(vals, {
**grouping_key,
'base_amount': values[f'base_amount{suffix}'],
'tax_amount': values[f'tax_amount{suffix}'],
}))
def _ubl_add_line_item_node(self, vals):
node = vals['line_node']['cac:Item'] = {}
sub_vals = {**vals, 'item_node': node}
self._ubl_add_line_item_name_description_nodes(sub_vals)
self._ubl_add_line_item_identification_nodes(sub_vals)
self._ubl_add_line_item_additional_item_property_nodes(sub_vals)
self._ubl_add_line_item_commodity_classification_nodes(sub_vals)
self._ubl_add_line_item_classified_tax_category_nodes(sub_vals)
def _ubl_add_line_price_node(self, vals, in_foreign_currency=True):
line_node = vals['line_node']
base_line = vals['line_vals']['base_line']
currency = base_line['currency_id'] if in_foreign_currency else vals['company_currency']
raw_gross_total_excluded = self.env['account.tax']._get_gross_total_without_tax(
base_line=base_line,
company=vals['company'],
in_foreign_currency=in_foreign_currency,
)
raw_gross_price_unit = self.env['account.tax']._get_price_unit_without_tax(
base_line=base_line,
company=vals['company'],
raw_gross_total_excluded=raw_gross_total_excluded,
in_foreign_currency=in_foreign_currency,
)
line_node['cac:Price'] = {
'cbc:PriceAmount': {
'_text': FloatFmt(raw_gross_price_unit, min_dp=1, max_dp=10),
'currencyID': currency.name,
},
}
def _ubl_get_line_allowance_charge_recycling_contribution_node(self, vals, recycling_contribution_values):
currency = recycling_contribution_values['currency']
amount = recycling_contribution_values['amount']
tax = recycling_contribution_values['tax']
if 'bebat' in tax.name.lower():
charge_reason_code = 'CAV'
else:
charge_reason_code = 'AEO'
is_charge = recycling_contribution_values['is_charge']
return {
'_currency': currency,
'cbc:ChargeIndicator': {'_text': 'true' if is_charge else 'false'},
'cbc:AllowanceChargeReasonCode': {'_text': charge_reason_code if is_charge else '100'},
'cbc:AllowanceChargeReason': {'_text': tax.name},
'cbc:Amount': {
'_text': FloatFmt(abs(amount), max_dp=currency.decimal_places),
'currencyID': currency.name,
},
}
def _ubl_get_line_allowance_charge_excise_node(self, vals, excise_values):
currency = excise_values['currency']
amount = excise_values['amount']
tax = excise_values['tax']
is_charge = excise_values['is_charge']
return {
'_currency': currency,
'cbc:ChargeIndicator': {'_text': 'true' if is_charge else 'false'},
'cbc:AllowanceChargeReason': {'_text': tax.name},
'cbc:Amount': {
'_text': FloatFmt(abs(amount), max_dp=currency.decimal_places),
'currencyID': currency.name,
},
}
def _ubl_get_line_allowance_charge_discount_node(self, vals, discount_values):
currency = discount_values['currency']
amount = discount_values['amount']
base_amount = discount_values['base_amount']
percent = discount_values['percent']
is_charge = discount_values['is_charge']
return {
'_currency': currency,
'cbc:ChargeIndicator': {'_text': 'true' if is_charge else 'false'},
'cbc:MultiplierFactorNumeric': {'_text': abs(percent)},
'cbc:AllowanceChargeReasonCode': {'_text': 'ADK' if is_charge else '95'},
'cbc:AllowanceChargeReason': {'_text': _("Discount")},
'cbc:Amount': {
'_text': FloatFmt(abs(amount), max_dp=currency.decimal_places),
'currencyID': currency.name,
},
'cbc:BaseAmount': {
'_text': FloatFmt(abs(base_amount), max_dp=currency.decimal_places),
'currencyID': currency.name,
},
}
def _ubl_add_line_allowance_charge_nodes_for_discount(self, vals, in_foreign_currency=True):
line_node = vals['line_node']
base_line = vals['line_vals']['base_line']
currency = base_line['currency_id'] if in_foreign_currency else vals['company_currency']
suffix = '_currency' if in_foreign_currency else ''
tax_details = base_line['tax_details']
raw_discount_amount = tax_details[f'discount_amount{suffix}']
if currency.is_zero(raw_discount_amount):
return
allowance_charges_nodes = line_node['cac:AllowanceCharge']
allowance_charges_nodes.append(self._ubl_get_line_allowance_charge_discount_node(vals, {
'currency': currency,
'percent': base_line['discount'],
'is_charge': raw_discount_amount < 0.0,
'amount': raw_discount_amount,
'base_amount': tax_details[f'gross_total_excluded{suffix}'],
}))
def _ubl_add_line_allowance_charge_nodes_for_recycling_contribution_taxes(self, vals, in_foreign_currency=True):
line_node = vals['line_node']
base_line = vals['line_vals']['base_line']
currency = base_line['currency_id'] if in_foreign_currency else vals['company_currency']
suffix = '_currency' if in_foreign_currency else ''
allowance_charges_nodes = line_node['cac:AllowanceCharge']
for tax_data in base_line['tax_details']['taxes_data']:
if not self._ubl_is_recycling_contribution_tax(tax_data):
continue
allowance_charges_nodes.append(self._ubl_get_line_allowance_charge_recycling_contribution_node(vals, {
'tax': tax_data['tax'],
'is_charge': tax_data['tax_amount'] > 0.0,
'amount': tax_data[f'tax_amount{suffix}'],
'currency': currency,
}))
def _ubl_add_line_allowance_charge_nodes_for_excise_taxes(self, vals, in_foreign_currency=True):
line_node = vals['line_node']
base_line = vals['line_vals']['base_line']
currency = base_line['currency_id'] if in_foreign_currency else vals['company_currency']
suffix = '_currency' if in_foreign_currency else ''
allowance_charges_nodes = line_node['cac:AllowanceCharge']
for tax_data in base_line['tax_details']['taxes_data']:
if not self._ubl_is_excise_tax(tax_data):
continue
allowance_charges_nodes.append(self._ubl_get_line_allowance_charge_excise_node(vals, {
'tax': tax_data['tax'],
'is_charge': tax_data['tax_amount'] > 0.0,
'amount': tax_data[f'tax_amount{suffix}'],
'currency': currency,
}))
def _ubl_add_line_allowance_charge_nodes(self, vals):
vals['line_node']['cac:AllowanceCharge'] = []
def _ubl_add_line_extension_amount_node(self, vals, in_foreign_currency=True):
line_node = vals['line_node']
base_line = vals['line_vals']['base_line']
currency = base_line['currency_id'] if in_foreign_currency else vals['company_currency']
suffix = '_currency' if in_foreign_currency else ''
tax_details = base_line['tax_details']
gross_total_excluded = currency.round(tax_details[f'raw_gross_total_excluded{suffix}'])
for allowance_charge_node in line_node['cac:AllowanceCharge']:
sign = 1 if allowance_charge_node['cbc:ChargeIndicator']['_text'] == 'true' else -1
gross_total_excluded += sign * allowance_charge_node['cbc:Amount']['_text']
line_node['cbc:LineExtensionAmount'] = {
'_text': FloatFmt(gross_total_excluded, min_dp=currency.decimal_places),
'currencyID': currency.name,
}
def _ubl_add_line_period_nodes(self, vals):
# DEPRECATED
nodes = vals['line_node']['cac:InvoicePeriod'] = []
if self._is_document(vals, 'invoice', 'credit_note', 'self_invoice', 'self_credit_note'):
base_line = vals['line_vals']['base_line']
if base_line.get('deferred_start_date') or base_line.get('deferred_end_date'):
nodes.append({
'cbc:StartDate': {'_text': base_line['deferred_start_date']},
'cbc:EndDate': {'_text': base_line['deferred_end_date']},
})
def _ubl_add_line_pricing_reference_node(self, vals):
vals['line_node']['cac:PricingReference'] = {}
def _ubl_add_line_tax_totals_nodes(self, vals):
vals['line_node']['cac:TaxTotal'] = []
def _line_nodes_filter_base_lines(self, vals, filter_function=None):
index = 1
for base_line in vals['base_lines']:
if not filter_function or filter_function(base_line):
line_vals = {'base_line': base_line, 'index': index}
line_node = {}
index += 1
yield {**vals, 'line_vals': line_vals, 'line_node': line_node}
def _ubl_add_party_endpoint_id_node(self, vals):
vals['party_node']['cbc:EndpointID'] = {
'_text': None,
'schemeID': None,
}
def _ubl_add_party_identification_nodes_iso_6523_icd(self, vals):
nodes = vals['party_node']['cac:PartyIdentification']
partner = vals['party_vals']['partner']
commercial_partner = partner.commercial_partner_id
country_code = commercial_partner.country_code
if country_code == 'BE' and commercial_partner.company_registry:
nodes.append({
'cbc:ID': {
'_text': be_vat.compact(commercial_partner.company_registry),
'schemeID': '0208',
},
})
def _ubl_add_party_identification_nodes(self, vals):
vals['party_node']['cac:PartyIdentification'] = []
def _ubl_add_party_name_node(self, vals):
partner = vals['party_vals']['partner']
# When the selected partner is a contact or an invoice address, there is nothing ensuring the partner's name is set.
# In that case, fallback on the commercial partner's name.
if partner.name:
name = partner.display_name
else:
name = partner.commercial_partner_id.display_name
vals['party_node']['cac:PartyName'] = {
'cbc:Name': {'_text': name},
}
def _ubl_get_partner_address_node(self, vals, partner):
return {
'cbc:StreetName': {'_text': partner.street},
'cbc:AdditionalStreetName': {'_text': partner.street2},
'cbc:CityName': {'_text': partner.city},
'cbc:PostalZone': {'_text': partner.zip},
'cbc:CountrySubentity': {'_text': partner.state_id.name},
'cbc:CountrySubentityCode': {'_text': partner.state_id.code},
'cac:Country': {
'cbc:IdentificationCode': {'_text': partner.country_id.code},
'cbc:Name': {'_text': partner.country_id.name},
},
}
def _ubl_add_party_postal_address_node(self, vals):
partner = vals['party_vals']['partner']
vals['party_node']['cac:PostalAddress'] = self._ubl_get_partner_address_node(vals, partner)
def _ubl_add_party_tax_scheme_nodes_vat_gst(self, vals):
nodes = vals['party_node']['cac:PartyTaxScheme']
partner = vals['party_vals']['partner']
commercial_partner = partner.commercial_partner_id
country_code = commercial_partner.country_code
if not country_code:
return
if commercial_partner.vat and commercial_partner.vat != '/':
vat = commercial_partner.vat
country_code = commercial_partner.country_id.code
if country_code in GST_COUNTRY_CODES:
tax_scheme_id = 'GST'
else:
tax_scheme_id = 'VAT'
if country_code == 'HU' and not vat.upper().startswith('HU'):
vat = 'HU' + vat[:8]
if country_code == 'DK' and not vat.upper().startswith('DK'):
vat = f'DK{vat}'
if country_code == 'NO' and not vat.upper().startswith('NO'):
vat = f'NO{vat}'
if country_code == 'NO' and vat[-3:] != 'MVA':
vat += 'MVA'
nodes.append({
'cbc:CompanyID': {'_text': vat},
'cac:TaxScheme': {
'cbc:ID': {'_text': tax_scheme_id},
},
})
def _ubl_add_party_tax_scheme_nodes(self, vals):
vals['party_node']['cac:PartyTaxScheme'] = []
def _ubl_add_party_legal_entity_nodes_iso_6523_icd(self, vals):
nodes = vals['party_node']['cac:PartyLegalEntity']
partner = vals['party_vals']['partner']
commercial_partner = partner.commercial_partner_id
vat = commercial_partner.vat != '/' and commercial_partner.vat
if commercial_partner.peppol_eas in ('0106', '0190'):
nl_id = commercial_partner.peppol_endpoint
else:
nl_id = commercial_partner.company_registry
if commercial_partner.country_code == 'NL' and nl_id:
# For NL, VAT can be used as a Peppol endpoint, but KVK/OIN has to be used as PartyLegalEntity/CompanyID
# To implement a workaround on stable, company_registry field is used without recording whether
# the number is a KVK or OIN, and the length of the number (8 = KVK, 20 = OIN) is used to determine the type
nodes.append({
'cbc:RegistrationName': {'_text': commercial_partner.name},
'cbc:CompanyID': {
'_text': nl_id,
'schemeID': '0190' if len(nl_id) == 20 else '0106',
},
})
elif commercial_partner.country_code == 'LU' and commercial_partner.company_registry:
nodes.append({
'cbc:RegistrationName': {'_text': commercial_partner.name},
'cbc:CompanyID': {
'_text': commercial_partner.company_registry,
'schemeID': None,
},
})
elif commercial_partner.country_code == 'SE' and commercial_partner.company_registry:
nodes.append({
'cbc:RegistrationName': {'_text': commercial_partner.name},
'cbc:CompanyID': {
'_text': ''.join(char for char in commercial_partner.company_registry if char.isdigit()),
},
})
elif commercial_partner.country_code == 'BE' and commercial_partner.company_registry:
nodes.append({
'cbc:RegistrationName': {'_text': commercial_partner.name},
'cbc:CompanyID': {
'_text': be_vat.compact(commercial_partner.company_registry),
'schemeID': '0208',
},
})
elif (
commercial_partner.country_code == 'DK'
and commercial_partner.peppol_eas == '0184'
and commercial_partner.peppol_endpoint
):
nodes.append({
'cbc:RegistrationName': {'_text': commercial_partner.name},
'cbc:CompanyID': {
'_text': commercial_partner.peppol_endpoint,
'schemeID': '0184',
},
})
elif commercial_partner.country_code == 'AU' and vat:
nodes.append({
'cbc:RegistrationName': {'_text': commercial_partner.name},
'cbc:CompanyID': {
'_text': commercial_partner.vat,
'schemeID': '0151',
},
})
elif commercial_partner.country_code == 'NZ' and vat:
nodes.append({
'cbc:RegistrationName': {'_text': commercial_partner.name},
'cbc:CompanyID': {
'_text': commercial_partner.vat,
'schemeID': '0088',
},
})
elif (
commercial_partner.country_code == 'NO'
and (
no_id := (
(
'l10n_no_bronnoysund_number' in self.env['res.partner']._fields
and commercial_partner.l10n_no_bronnoysund_number
)
or commercial_partner.company_registry
)
)
):
nodes.append({
'cbc:RegistrationName': {'_text': commercial_partner.name},
'cbc:CompanyID': {
'_text': no_id,
'schemeID': '0192',
},
})
elif commercial_partner.country_code == 'NO' and commercial_partner.vat != '/':
if not vat.startswith('NO'):
vat = f'NO{vat}'
if not vat.endswith('MVA'):
vat += 'MVA'
nodes.append({
'cbc:RegistrationName': {'_text': commercial_partner.name},
'cbc:CompanyID': {
'_text': vat,
'schemeID': None,
},
})
elif commercial_partner.vat and commercial_partner.vat != '/':
nodes.append({
'cbc:RegistrationName': {'_text': commercial_partner.name},
'cbc:CompanyID': {
'_text': commercial_partner.vat,
'schemeID': None,
},
})
elif commercial_partner.peppol_endpoint:
nodes.append({
'cbc:RegistrationName': {'_text': commercial_partner.name},
'cbc:CompanyID': {
'_text': commercial_partner.peppol_endpoint,
'schemeID': None,
},
})
def _ubl_add_party_legal_entity_nodes(self, vals):
vals['party_node']['cac:PartyLegalEntity'] = []
def _ubl_add_party_contact_node(self, vals):
partner = vals['party_vals']['partner']
vals['party_node']['cac:Contact'] = {
'cbc:ID': {'_text': None},
'cbc:Name': {'_text': partner.name},
'cbc:Telephone': {'_text': partner.phone},
'cbc:ElectronicMail': {'_text': partner.email},
}
def _ubl_add_accounting_supplier_party_endpoint_id_node(self, vals):
self._ubl_add_party_endpoint_id_node(vals)
def _ubl_add_accounting_supplier_party_identification_nodes(self, vals):
self._ubl_add_party_identification_nodes(vals)
def _ubl_add_accounting_supplier_party_name_node(self, vals):
self._ubl_add_party_name_node(vals)
def _ubl_add_accounting_supplier_party_postal_address_node(self, vals):
self._ubl_add_party_postal_address_node(vals)
def _ubl_add_accounting_supplier_party_tax_scheme_nodes(self, vals):
self._ubl_add_party_tax_scheme_nodes(vals)
def _ubl_add_accounting_supplier_party_legal_entity_nodes(self, vals):
self._ubl_add_party_legal_entity_nodes(vals)
def _ubl_add_accounting_supplier_party_contact_node(self, vals):
self._ubl_add_party_contact_node(vals)
def _ubl_add_accounting_supplier_party_node(self, vals):
node = vals['document_node']['cac:AccountingSupplierParty'] = {'cac:Party': {}}
party_node = node['cac:Party']
sub_vals = {
**vals,
'party_vals': {'partner': vals['supplier']},
'party_node': party_node,
}
self._ubl_add_accounting_supplier_party_endpoint_id_node(sub_vals)
self._ubl_add_accounting_supplier_party_identification_nodes(sub_vals)
self._ubl_add_accounting_supplier_party_name_node(sub_vals)
self._ubl_add_accounting_supplier_party_postal_address_node(sub_vals)
self._ubl_add_accounting_supplier_party_tax_scheme_nodes(sub_vals)
self._ubl_add_accounting_supplier_party_legal_entity_nodes(sub_vals)
self._ubl_add_accounting_supplier_party_contact_node(sub_vals)
def _ubl_add_accounting_customer_party_endpoint_id_node(self, vals):
self._ubl_add_party_endpoint_id_node(vals)
def _ubl_add_accounting_customer_party_identification_nodes(self, vals):
self._ubl_add_party_identification_nodes(vals)
def _ubl_add_accounting_customer_party_name_node(self, vals):
self._ubl_add_party_name_node(vals)
def _ubl_add_accounting_customer_party_postal_address_node(self, vals):
self._ubl_add_party_postal_address_node(vals)
def _ubl_add_accounting_customer_party_tax_scheme_nodes(self, vals):
self._ubl_add_party_tax_scheme_nodes(vals)
def _ubl_add_accounting_customer_party_legal_entity_nodes(self, vals):
self._ubl_add_party_legal_entity_nodes(vals)
def _ubl_add_accounting_customer_party_contact_node(self, vals):
self._ubl_add_party_contact_node(vals)
def _ubl_add_accounting_customer_party_node(self, vals):
node = vals['document_node']['cac:AccountingCustomerParty'] = {'cac:Party': {}}
party_node = node['cac:Party']
sub_vals = {
**vals,
'party_vals': {'partner': vals['customer']},
'party_node': party_node,
}
self._ubl_add_accounting_customer_party_endpoint_id_node(sub_vals)
self._ubl_add_accounting_customer_party_identification_nodes(sub_vals)
self._ubl_add_accounting_customer_party_name_node(sub_vals)
self._ubl_add_accounting_customer_party_postal_address_node(sub_vals)
self._ubl_add_accounting_customer_party_tax_scheme_nodes(sub_vals)
self._ubl_add_accounting_customer_party_legal_entity_nodes(sub_vals)
self._ubl_add_accounting_customer_party_contact_node(sub_vals)
def _ubl_add_seller_supplier_party_node(self, vals):
node = vals['document_node']['cac:SellerSupplierParty'] = {'cac:Party': {}}
party_node = node['cac:Party']
sub_vals = {
**vals,
'party_vals': {'partner': vals['supplier']},
'party_node': party_node,
}
self._ubl_add_accounting_supplier_party_endpoint_id_node(sub_vals)
self._ubl_add_accounting_supplier_party_identification_nodes(sub_vals)
self._ubl_add_accounting_supplier_party_name_node(sub_vals)
self._ubl_add_accounting_supplier_party_postal_address_node(sub_vals)
self._ubl_add_accounting_supplier_party_tax_scheme_nodes(sub_vals)
self._ubl_add_accounting_supplier_party_legal_entity_nodes(sub_vals)
self._ubl_add_accounting_supplier_party_contact_node(sub_vals)
def _ubl_add_buyer_customer_party_node(self, vals):
node = vals['document_node']['cac:BuyerCustomerParty'] = {'cac:Party': {}}
party_node = node['cac:Party']
sub_vals = {
**vals,
'party_vals': {'partner': vals['customer']},
'party_node': party_node,
}
self._ubl_add_accounting_customer_party_endpoint_id_node(sub_vals)
self._ubl_add_accounting_customer_party_identification_nodes(sub_vals)
self._ubl_add_accounting_customer_party_name_node(sub_vals)
self._ubl_add_accounting_customer_party_postal_address_node(sub_vals)
self._ubl_add_accounting_customer_party_tax_scheme_nodes(sub_vals)
self._ubl_add_accounting_customer_party_legal_entity_nodes(sub_vals)
self._ubl_add_accounting_customer_party_contact_node(sub_vals)
def _ubl_add_delivery_party_endpoint_id_node(self, vals):
self._ubl_add_party_endpoint_id_node(vals)
def _ubl_add_delivery_party_identification_nodes(self, vals):
self._ubl_add_party_identification_nodes(vals)
def _ubl_add_delivery_party_name_node(self, vals):
self._ubl_add_party_name_node(vals)
def _ubl_add_delivery_party_postal_address_node(self, vals):
self._ubl_add_party_postal_address_node(vals)
def _ubl_add_delivery_party_tax_scheme_nodes(self, vals):
self._ubl_add_party_tax_scheme_nodes(vals)
def _ubl_add_delivery_party_legal_entity_nodes(self, vals):
self._ubl_add_party_legal_entity_nodes(vals)
def _ubl_add_delivery_party_contact_node(self, vals):
self._ubl_add_party_contact_node(vals)
def _ubl_get_delivery_node_from_delivery_address(self, vals):
delivery_partner = vals['delivery']
node = {
'cbc:ActualDeliveryDate': {'_text': None},
'cac:DeliveryLocation': {
'cbc:ID': {
'schemeID': None,
'_text': None,
},
'cac:Address': self._ubl_get_partner_address_node(vals, delivery_partner),
},
}
if delivery_partner.global_location_number:
node['cac:DeliveryLocation']['cbc:ID']['schemeID'] = '0088'
node['cac:DeliveryLocation']['cbc:ID']['_text'] = delivery_partner.global_location_number
if self._is_document(vals, 'invoice', 'credit_note', 'self_invoice', 'self_credit_note'):
invoice = vals['invoice']
if invoice.delivery_date:
node['cbc:ActualDeliveryDate']['_text'] = invoice.delivery_date
party_node = node['cac:DeliveryParty'] = {}
sub_vals = {
**vals,
'party_vals': {'partner': vals['delivery']},
'party_node': party_node,
}
self._ubl_add_delivery_party_endpoint_id_node(sub_vals)
self._ubl_add_delivery_party_identification_nodes(sub_vals)
self._ubl_add_delivery_party_name_node(sub_vals)
self._ubl_add_delivery_party_postal_address_node(sub_vals)
self._ubl_add_delivery_party_tax_scheme_nodes(sub_vals)
self._ubl_add_delivery_party_legal_entity_nodes(sub_vals)
self._ubl_add_delivery_party_contact_node(sub_vals)
return node
def _ubl_add_delivery_nodes(self, vals):
nodes = vals['document_node']['cac:Delivery'] = []
if vals.get('delivery'):
nodes.append(self._ubl_get_delivery_node_from_delivery_address(vals))
def _ubl_add_invoice_line_node(self, vals):
self._ubl_add_line_id_node(vals)
self._ubl_add_line_note_nodes(vals)
self._ubl_add_line_invoiced_quantity_node(vals)
self._ubl_add_line_allowance_charge_nodes(vals)
self._ubl_add_line_extension_amount_node(vals)
self._ubl_add_line_pricing_reference_node(vals)
self._ubl_add_line_tax_totals_nodes(vals)
self._ubl_add_line_item_node(vals)
self._ubl_add_line_price_node(vals)
def _ubl_add_invoice_line_nodes(self, vals, filter_function=None):
nodes = vals['document_node']['cac:InvoiceLine'] = []
for sub_vals in self._line_nodes_filter_base_lines(vals, filter_function=filter_function):
self._ubl_add_invoice_line_node(sub_vals)
nodes.append(sub_vals['line_node'])
def _ubl_add_credit_note_line_node(self, vals):
self._ubl_add_line_id_node(vals)
self._ubl_add_line_note_nodes(vals)
self._ubl_add_line_credited_quantity_node(vals)
self._ubl_add_line_allowance_charge_nodes(vals)
self._ubl_add_line_extension_amount_node(vals)
self._ubl_add_line_pricing_reference_node(vals)
self._ubl_add_line_tax_totals_nodes(vals)
self._ubl_add_line_item_node(vals)
self._ubl_add_line_price_node(vals)
def _ubl_add_credit_note_line_nodes(self, vals, filter_function=None):
nodes = vals['document_node']['cac:CreditNoteLine'] = []
for sub_vals in self._line_nodes_filter_base_lines(vals, filter_function=filter_function):
self._ubl_add_credit_note_line_node(sub_vals)
nodes.append(sub_vals['line_node'])
def _ubl_add_debit_note_line_node(self, vals):
self._ubl_add_line_id_node(vals)
self._ubl_add_line_note_nodes(vals)
self._ubl_add_line_debited_quantity_node(vals)
self._ubl_add_line_allowance_charge_nodes(vals)
self._ubl_add_line_extension_amount_node(vals)
self._ubl_add_line_pricing_reference_node(vals)
self._ubl_add_line_tax_totals_nodes(vals)
self._ubl_add_line_item_node(vals)
self._ubl_add_line_price_node(vals)
def _ubl_add_debit_note_line_nodes(self, vals, filter_function=None):
nodes = vals['document_node']['cac:DebitNoteLine'] = []
for sub_vals in self._line_nodes_filter_base_lines(vals, filter_function=filter_function):
self._ubl_add_debit_note_line_node(sub_vals)
nodes.append(sub_vals['line_node'])
def _ubl_add_version_id_node(self, vals):
vals['document_node']['cbc:UBLVersionID'] = {'_text': None}
def _ubl_add_customization_id_node(self, vals):
vals['document_node']['cbc:CustomizationID'] = {'_text': None}
def _ubl_add_profile_id_node(self, vals):
vals['document_node']['cbc:ProfileID'] = {'_text': None}
def _ubl_add_id_node(self, vals):
vals['document_node']['cbc:ID'] = {'_text': None}
if self._is_document(vals, 'invoice', 'credit_note', 'self_invoice', 'self_credit_note'):
vals['document_node']['cbc:ID']['_text'] = vals['invoice'].name
def _ubl_add_copy_indicator_node(self, vals):
vals['document_node']['cbc:CopyIndicator'] = {'_text': None}
def _ubl_add_issue_date_node(self, vals):
issue_date_node = vals['document_node']['cbc:IssueDate'] = {'_text': None}
vals['document_node']['cbc:IssueTime'] = {'_text': None}
if self._is_document(vals, 'invoice', 'credit_note', 'self_invoice', 'self_credit_note'):
issue_date_node['_text'] = vals['invoice'].invoice_date
def _ubl_add_due_date_node(self, vals):
due_date_node = vals['document_node']['cbc:DueDate'] = {'_text': None}
if self._is_document(vals, 'invoice', 'credit_note', 'self_invoice', 'self_credit_note'):
due_date_node['_text'] = vals['invoice'].invoice_date_due
def _ubl_add_invoice_type_code_node(self, vals):
vals['document_node']['cbc:InvoiceTypeCode'] = {'_text': None}
def _ubl_add_credit_note_type_code_node(self, vals):
vals['document_node']['cbc:CreditNoteTypeCode'] = {'_text': None}
def _ubl_add_order_type_code_node(self, vals):
vals['document_node']['cbc:OrderTypeCode'] = {'_text': None}
def _ubl_add_notes_nodes(self, vals):
vals['document_node']['cbc:Note'] = []
def _ubl_add_document_currency_code_node_foreign_currency(self, vals):
vals['document_node']['cbc:DocumentCurrencyCode'] = {'_text': vals['currency'].name}
def _ubl_add_document_currency_code_node_company_currency(self, vals):
vals['document_node']['cbc:DocumentCurrencyCode'] = {'_text': vals['company'].currency_id.name}
def _ubl_add_document_currency_code_node(self, vals):
vals['document_node']['cbc:DocumentCurrencyCode'] = {'_text': None}
def _ubl_add_tax_currency_code_node_company_currency_if_foreign_currency(self, vals):
company = vals['company']
currency = vals['currency_id']
vals['document_node']['cbc:TaxCurrencyCode'] = {'_text': None if currency == company.currency_id else company.currency_id.name}
def _ubl_add_tax_currency_code_node_company_currency(self, vals):
vals['document_node']['cbc:TaxCurrencyCode'] = {'_text': vals['company'].currency_id.name}
def _ubl_add_tax_currency_code_node_empty(self, vals):
vals['document_node']['cbc:TaxCurrencyCode'] = {'_text': None}
def _ubl_add_tax_currency_code_node(self, vals):
vals['document_node']['cbc:TaxCurrencyCode'] = {'_text': None}
def _ubl_add_buyer_reference_node(self, vals):
vals['document_node']['cbc:BuyerReference'] = {'_text': None}
def _ubl_add_invoice_period_nodes(self, vals):
vals['document_node']['cac:InvoicePeriod'] = {}
def _ubl_add_order_reference_node(self, vals):
order_ref_node = vals['document_node']['cac:OrderReference'] = {
'cbc:ID': {'_text': None},
'cbc:SalesOrderID': {
'_text': None,
},
}
if self._is_document(vals, 'invoice', 'credit_note', 'self_invoice', 'self_credit_note'):
invoice = vals['invoice']
# Purchase order reference
# An identifier of a referenced purchase order, issued by the Buyer.
# Suppose the following case:
# - Buyer does a RFQ to the Seller.
# - Seller confirms with a SO.
# - Buyer converts the RFQ to a PO.
# => There is no automatic tracking of this information.
# Instead, the user can encode this information on 'Customer Reference' a.k.a the 'ref' field.
# Since ID is required, the fallback is also fine and avoid to force the encoding of this
# manual information.
order_ref_node['cbc:ID']['_text'] = invoice.ref or invoice.name
# Sales order reference
# An identifier of a referenced sales order issued by the Seller.
if self.module_installed('sale'):
so_names = set(invoice.invoice_line_ids.sale_line_ids.order_id.mapped('name'))
if so_names:
order_ref_node['cbc:SalesOrderID']['_text'] = ",".join(so_names)
def _ubl_add_billing_reference_nodes(self, vals):
vals['document_node']['cac:BillingReference'] = []
def _ubl_get_partner_bank_address_node(self, vals, partner_bank):
return {
'cbc:StreetName': {'_text': partner_bank.street},
'cbc:AdditionalStreetName': {'_text': partner_bank.street2},
'cbc:CityName': {'_text': partner_bank.city},
'cbc:PostalZone': {'_text': partner_bank.zip},
'cbc:CountrySubentity': {'_text': partner_bank.state_id.name},
'cbc:CountrySubentityCode': {'_text': partner_bank.state_id.code},
'cac:Country': {
'cbc:IdentificationCode': {'_text': partner_bank.country_id.code},
'cbc:Name': {'_text': partner_bank.country_id.name},
},
}
def _ubl_get_payment_means_payee_financial_account_institution_branch_node_from_partner_bank(self, vals, partner_bank):
if not partner_bank.bank_bic:
return None
return {
'cbc:ID': {
'_text': partner_bank.bank_bic,
'schemeID': 'BIC',
},
'cac:FinancialInstitution': {
'cbc:ID': {
'_text': partner_bank.bank_bic,
'schemeID': 'BIC',
},
'cbc:Name': {'_text': partner_bank.bank_name},
'cac:Address': self._ubl_get_partner_bank_address_node(vals, partner_bank)
}
}
def _ubl_get_payment_means_payee_financial_account_node_from_partner_bank(self, vals, partner_bank):
return {
'cbc:ID': {'_text': sanitize_account_number(partner_bank.account_number)},
'cac:FinancialInstitutionBranch': self._ubl_get_payment_means_payee_financial_account_institution_branch_node_from_partner_bank(vals, partner_bank),
}
def _ubl_add_payment_means_nodes(self, vals):
vals['document_node']['cac:PaymentMeans'] = []
def _ubl_get_payment_terms_node_from_payment_term(self, vals, payment_term):
note = payment_term.note and html2plaintext(payment_term.note) or None
if not note:
return
return {
'cbc:Note': {'_text': note}
}
def _ubl_add_payment_terms_nodes(self, vals):
nodes = vals['document_node']['cac:PaymentTerms'] = []
if self._is_document(vals, 'invoice', 'credit_note', 'self_invoice', 'self_credit_note'):
invoice = vals['invoice']
if payment_terms_node := self._ubl_get_payment_terms_node_from_payment_term(vals, invoice.invoice_payment_term_id):
nodes.append(payment_terms_node)
def _ubl_get_allowance_charge_early_payment_tax_category_node(self, vals, tax_category):
return {
'_currency': tax_category['currency'],
'cbc:ID': {'_text': tax_category['tax_category_code']},
'cbc:Percent': {'_text': tax_category['percent']},
'cac:TaxScheme': {
'cbc:ID': {'_text': tax_category['scheme_id']},
}
}
def _ubl_get_allowance_charge_global_discount_tax_category_node(self, vals, tax_category):
return {
'_currency': tax_category['currency'],
'cbc:ID': {'_text': tax_category['tax_category_code']},
'cbc:Percent': {'_text': tax_category['percent']},
'cac:TaxScheme': {
'cbc:ID': {'_text': tax_category['scheme_id']},
},
}
def _ubl_get_allowance_charge_early_payment_node(self, vals, early_payment_values):
currency = early_payment_values['currency']
amount = early_payment_values['amount']
is_charge = early_payment_values['is_charge']
return {
'_currency': currency,
'cbc:ChargeIndicator': {'_text': 'true' if is_charge else 'false'},
'cbc:AllowanceChargeReasonCode': {'_text': 'ZZZ' if is_charge else '64'},
'cbc:AllowanceChargeReason': {'_text': _("Conditional cash/payment discount")},
'cbc:Amount': {
'_text': currency.round(abs(amount)),
'currencyID': currency.name,
},
'cac:TaxCategory': [
self._ubl_get_allowance_charge_early_payment_tax_category_node(vals, tax_category)
for tax_category in early_payment_values['tax_categories'].values()
],
}
def _ubl_get_allowance_charge_global_discount_node(self, vals, global_discount_values):
currency = global_discount_values['currency']
amount = global_discount_values['amount']
is_charge = global_discount_values['is_charge']
return {
'_currency': currency,
'cbc:ChargeIndicator': {'_text': 'true' if is_charge else 'false'},
'cbc:AllowanceChargeReasonCode': {'_text': 'ADK' if is_charge else '95'},
'cbc:AllowanceChargeReason': {'_text': _("General upsell") if is_charge else _("General discount")},
'cbc:Amount': {
'_text': FloatFmt(abs(amount), max_dp=currency.decimal_places),
'currencyID': currency.name,
},
'cac:TaxCategory': [
self._ubl_get_allowance_charge_global_discount_tax_category_node(vals, tax_category)
for tax_category in global_discount_values['tax_categories'].values()
],
}
def _ubl_add_allowance_charge_nodes_early_payment_discount(self, vals, in_foreign_currency=True):
AccountTax = self.env['account.tax']
suffix = '_currency' if in_foreign_currency else ''
currency = vals['currency'] if in_foreign_currency else vals['company_currency']
allowance_charge_nodes = vals['document_node']['cac:AllowanceCharge']
base_lines_aggregated_values = AccountTax._aggregate_base_lines_tax_details(
base_lines=vals['base_lines'],
grouping_function=lambda base_line, tax_data: self._ubl_default_allowance_charge_early_payment_grouping_key(
base_line=base_line,
tax_data=tax_data,
vals=vals,
currency=currency,
),
)
values_per_grouping_key = AccountTax._aggregate_base_lines_aggregated_values(base_lines_aggregated_values)
for grouping_key, values in values_per_grouping_key.items():
if not grouping_key:
continue
allowance_charge_nodes.append(self._ubl_get_allowance_charge_early_payment_node(vals, {
'currency': currency,
'amount': values[f'total_excluded{suffix}'],
'is_charge': values[f'total_excluded{suffix}'] > 0.0,
'tax_categories': {
grouping_key: {
**grouping_key,
'base_amount': values[f'base_amount{suffix}'],
'tax_amount': values[f'tax_amount{suffix}'],
},
},
}))
def _ubl_add_allowance_charge_nodes_global_discount(self, vals, in_foreign_currency=True):
AccountTax = self.env['account.tax']
suffix = '_currency' if in_foreign_currency else ''
currency = vals['currency'] if in_foreign_currency else vals['company_currency']
allowance_charge_nodes = vals['document_node']['cac:AllowanceCharge']
base_lines_aggregated_values = AccountTax._aggregate_base_lines_tax_details(
base_lines=vals['base_lines'],
grouping_function=lambda base_line, tax_data: self._ubl_default_allowance_charge_global_discount_grouping_key(
base_line=base_line,
tax_data=tax_data,
vals=vals,
currency=currency,
),
)
values_per_grouping_key = AccountTax._aggregate_base_lines_aggregated_values(base_lines_aggregated_values)
for grouping_key, values in values_per_grouping_key.items():
if not grouping_key:
continue
allowance_charge_nodes.append(
self._ubl_get_allowance_charge_global_discount_node(vals, {
'currency': currency,
'amount': values[f'total_excluded{suffix}'],
'is_charge': values[f'total_excluded{suffix}'] > 0.0,
'tax_categories': {
grouping_key: {
**grouping_key,
'base_amount': values[f'base_amount{suffix}'],
'tax_amount': values[f'tax_amount{suffix}'],
},
},
}),
)
def _ubl_add_allowance_charge_nodes(self, vals):
vals['document_node']['cac:AllowanceCharge'] = []
def _ubl_get_tax_category_node(self, vals, tax_category):
""" Generate the node 'cac:TaxCategory' in 'cac:SubTotal'.
:param vals: Some custom data.
:param tax_category: An entry of vals['_ubl_values'](['tax_totals']|['withholding_tax_totals'])['tax_subtotals']
containing all the necessary data to build the node.
:return: A new node in 'cac:TaxTotal'.
"""
return {
'_currency': tax_category['currency'],
'cbc:ID': {'_text': tax_category['tax_category_code']},
'cbc:Name': {'_text': None},
'cbc:Percent': {'_text': tax_category['percent']},
'cbc:TaxExemptionReasonCode': {'_text': tax_category.get('tax_exemption_reason_code')},
'cbc:TaxExemptionReason': {'_text': tax_category.get('tax_exemption_reason')},
'cac:TaxScheme': {
'cbc:ID': {'_text': tax_category['scheme_id']},
}
}
def _ubl_get_tax_subtotal_node(self, vals, tax_subtotal):
""" Generate the node 'cac:SubTotal' in 'cac:TaxTotal'/'cac:WithholdingTaxTotal'.
Note: 'cac:TaxCategory' is managed by '_ubl_get_tax_category_node'.
:param vals: Some custom data.
:param tax_subtotal: An entry of vals['_ubl_values'](['tax_totals']|['withholding_tax_totals'])['tax_subtotals']
containing all the necessary data to build the node.
:return: A new node in 'cac:TaxTotal'.
"""
currency = tax_subtotal['currency']
return {
'_currency': currency,
'cbc:TaxableAmount': {
'_text': FloatFmt(tax_subtotal['base_amount'], min_dp=currency.decimal_places),
'currencyID': currency.name
},
'cbc:TaxAmount': {
'_text': FloatFmt(tax_subtotal['tax_amount'], min_dp=currency.decimal_places),
'currencyID': currency.name
},
'cbc:Percent': {
'_text': (
tax_subtotal['percent']
if tax_subtotal.get('percent') is not None
else None
),
},
'cac:TaxCategory': [
self._ubl_get_tax_category_node(vals, tax_category)
for tax_category in tax_subtotal['tax_categories'].values()
],
}
def _ubl_get_tax_total_node(self, vals, tax_total):
""" Generate the node 'cac:TaxTotal'.
Note: 'cac:Subtotal' is managed by '_ubl_get_tax_subtotal_node'.
:param vals: Some custom data.
:param tax_total: An entry of vals['_ubl_values']['tax_totals'] containing all the necessary data to build the node.
:return: A new node in 'cac:TaxTotal'.
"""
currency = tax_total['currency']
return {
'_currency': currency,
'cbc:TaxAmount': {
'_text': FloatFmt(tax_total['amount'], min_dp=currency.decimal_places),
'currencyID': currency.name
},
'cac:TaxSubtotal': [
self._ubl_get_tax_subtotal_node(vals, subtotal)
for subtotal in tax_total['subtotals'].values()
],
}
def _ubl_get_withholding_tax_total_node(self, vals, tax_total):
""" Generate the node 'cac:WithholdingTaxTotal'.
Note: 'cac:Subtotal' is managed by '_ubl_get_tax_subtotal_node'.
:param vals: Some custom data.
:param tax_total: An entry of vals['_ubl_values']['withholding_tax_totals'] containing all the necessary data to build the node.
:return: A new node in 'cac:WithholdingTaxTotal'.
"""
return self._ubl_get_tax_total_node(vals, tax_total)
def _ubl_tax_totals_node_grouping_key(self, base_line, tax_data, vals, currency):
tax_category_key = self._ubl_default_tax_category_grouping_key(base_line, tax_data, vals, currency)
if tax_category_key:
tax_subtotal_key = {
'currency': tax_category_key['currency'],
'is_withholding': tax_category_key['is_withholding'],
'tax_category_code': tax_category_key['tax_category_code'],
'scheme_id': tax_category_key['scheme_id'],
'percent': tax_category_key['percent'],
}
else:
tax_subtotal_key = None
if tax_category_key:
tax_total_key = {
'is_withholding': tax_category_key['is_withholding'],
'currency': currency,
}
else:
tax_total_key = None
return {
'tax_category_key': tax_category_key,
'tax_subtotal_key': tax_subtotal_key,
'tax_total_key': tax_total_key
}
def _ubl_add_tax_totals_nodes(self, vals):
AccountTax = self.env['account.tax']
base_lines = vals['base_lines']
company = vals['company']
company_currency = company.currency_id
currency = vals['currency_id']
iter_currency = [(currency, '_currency')]
if currency != company_currency:
iter_currency.append((company_currency, ''))
# Since we have to combine 3 keys given by the same grouping key but used at 3 different places, we compute them in advance.
# /!\ Even without tax, a base line could get a grouping key for a 0% tax without any real record set.
new_base_lines = [
{
**base_line,
'_tax_totals_keys': {
(tax_data['tax'] if tax_data else None, currency): self._ubl_tax_totals_node_grouping_key(base_line, tax_data, vals, currency)
for currency, _suffix in iter_currency
for tax_data in base_line['tax_details']['taxes_data'] or [None]
}
}
for base_line in base_lines
]
collected_tax_totals_values = {
'cac:TaxTotal': {},
'cac:WithholdingTaxTotal': {},
}
for sub_currency, suffix in iter_currency:
# tax_totals / withholding_tax_totals
base_lines_aggregated_values = AccountTax._aggregate_base_lines_tax_details(
base_lines=new_base_lines,
grouping_function=lambda base_line, tax_data: base_line['_tax_totals_keys'][(tax_data or {}).get('tax'), sub_currency]['tax_total_key']
)
values_per_grouping_key = AccountTax._aggregate_base_lines_aggregated_values(base_lines_aggregated_values)
for grouping_key, values in values_per_grouping_key.items():
if not grouping_key:
continue
if grouping_key['is_withholding']:
target_key = 'cac:WithholdingTaxTotal'
sign = -1
else:
target_key = 'cac:TaxTotal'
sign = 1
collected_tax_totals_values[target_key][frozendict(grouping_key)] = {
**grouping_key,
'amount': sign * values[f'tax_amount{suffix}'],
'subtotals': {},
}
# tax_subtotals
base_lines_aggregated_values = AccountTax._aggregate_base_lines_tax_details(
base_lines=new_base_lines,
grouping_function=lambda base_line, tax_data: {
k: v
for k, v in base_line['_tax_totals_keys'][(tax_data or {}).get('tax'), sub_currency].items()
if k in ('tax_total_key', 'tax_subtotal_key')
},
)
values_per_grouping_key = AccountTax._aggregate_base_lines_aggregated_values(base_lines_aggregated_values)
for grouping_key, values in values_per_grouping_key.items():
if not grouping_key:
continue
tax_total_key = grouping_key['tax_total_key']
tax_subtotal_key = grouping_key['tax_subtotal_key']
if not tax_total_key or not tax_subtotal_key:
continue
if tax_total_key['is_withholding']:
target_key = 'cac:WithholdingTaxTotal'
sign = -1
else:
target_key = 'cac:TaxTotal'
sign = 1
tax_total_values = collected_tax_totals_values[target_key][frozendict(tax_total_key)]
tax_total_values['subtotals'][frozendict(tax_subtotal_key)] = {
**tax_subtotal_key,
'base_amount': values[f'base_amount{suffix}'],
'tax_amount': sign * values[f'tax_amount{suffix}'],
'tax_categories': {},
}
# tax_categories
base_lines_aggregated_values = AccountTax._aggregate_base_lines_tax_details(
base_lines=new_base_lines,
grouping_function=lambda base_line, tax_data: base_line['_tax_totals_keys'][(tax_data or {}).get('tax'), sub_currency],
)
values_per_grouping_key = AccountTax._aggregate_base_lines_aggregated_values(base_lines_aggregated_values)
for grouping_key, values in values_per_grouping_key.items():
if not grouping_key:
continue
tax_total_key = grouping_key['tax_total_key']
tax_subtotal_key = grouping_key['tax_subtotal_key']
tax_category_key = grouping_key['tax_category_key']
if not tax_total_key or not tax_subtotal_key or not tax_category_key:
continue
if tax_total_key['is_withholding']:
target_key = 'cac:WithholdingTaxTotal'
sign = -1
else:
target_key = 'cac:TaxTotal'
sign = 1
tax_total_values = collected_tax_totals_values[target_key][frozendict(tax_total_key)]
tax_subtotal_values = tax_total_values['subtotals'][frozendict(tax_subtotal_key)]
tax_subtotal_values['tax_categories'][frozendict(tax_category_key)] = {
**tax_category_key,
'base_amount': values[f'base_amount{suffix}'],
'tax_amount': sign * values[f'tax_amount{suffix}'],
}
for key, tax_totals_values in collected_tax_totals_values.items():
nodes = vals['document_node'][key] = []
for tax_total_values in tax_totals_values.values():
tax_total_node = self._ubl_get_tax_total_node(vals, tax_total_values)
nodes.append(tax_total_node)
def _ubl_add_legal_monetary_total_line_extension_amount_node(self, vals, in_foreign_currency=True):
currency = vals['currency_id'] if in_foreign_currency else vals['company_currency']
line_extension_amount = sum(
line_node['cbc:LineExtensionAmount']['_text']
for line_key in ('cac:InvoiceLine', 'cac:CreditNoteLine', 'cac:DebitNoteLine')
for line_node in vals['document_node'].get(line_key, [])
)
vals['legal_monetary_total_node']['cbc:LineExtensionAmount'] = {
'_text': FloatFmt(line_extension_amount, min_dp=currency.decimal_places),
'currencyID': currency.name,
}
def _ubl_add_legal_monetary_total_tax_exclusive_amount_node(self, vals, in_foreign_currency=True):
""" The total amount of the document without TAX including all line net amounts
minus sum of allowance amount on document level
plus sum of charges on document level.
"""
currency = vals['currency_id'] if in_foreign_currency else vals['company_currency']
node = vals['legal_monetary_total_node']
tax_exlusive_amount = node['cbc:LineExtensionAmount']['_text']
document_node = vals['document_node']
for allowance_charge_node in document_node['cac:AllowanceCharge']:
sign = 1 if allowance_charge_node['cbc:ChargeIndicator']['_text'] == 'true' else -1
tax_exlusive_amount += sign * allowance_charge_node['cbc:Amount']['_text']
node['cbc:TaxExclusiveAmount'] = {
'_text': FloatFmt(
tax_exlusive_amount,
min_dp=currency.decimal_places,
),
'currencyID': currency.name,
}
def _ubl_add_legal_monetary_total_tax_inclusive_amount_node(self, vals, in_foreign_currency=True):
currency = vals['currency_id'] if in_foreign_currency else vals['company_currency']
document_node = vals['document_node']
node = vals['legal_monetary_total_node']
tax_amount = sum(
tax_total_node['cbc:TaxAmount']['_text']
for tax_total_node in document_node['cac:TaxTotal']
if tax_total_node['_currency'] == currency
) + sum(
-tax_total_node['cbc:TaxAmount']['_text']
for tax_total_node in document_node['cac:WithholdingTaxTotal']
if tax_total_node['_currency'] == currency
)
node['cbc:TaxInclusiveAmount'] = {
'_text': FloatFmt(
node['cbc:TaxExclusiveAmount']['_text'] + tax_amount,
min_dp=currency.decimal_places,
),
'currencyID': currency.name,
}
def _ubl_add_legal_monetary_total_allowance_charge_total_amount_node(self, vals, in_foreign_currency=True):
currency = vals['currency_id'] if in_foreign_currency else vals['company_currency']
node = vals['legal_monetary_total_node']
total_allowance = sum(
allowance_node['cbc:Amount']['_text']
for allowance_node in vals['document_node']['cac:AllowanceCharge']
if allowance_node['cbc:ChargeIndicator']['_text'] == 'false'
)
total_charge = sum(
charge_node['cbc:Amount']['_text']
for charge_node in vals['document_node']['cac:AllowanceCharge']
if charge_node['cbc:ChargeIndicator']['_text'] == 'true'
)
node.update({
'cbc:AllowanceTotalAmount': {
'_text': FloatFmt(total_allowance, min_dp=currency.decimal_places),
'currencyID': currency.name,
} if total_allowance else None,
'cbc:ChargeTotalAmount': {
'_text': FloatFmt(total_charge, min_dp=currency.decimal_places),
'currencyID': currency.name,
} if total_charge else None,
})
def _ubl_add_legal_monetary_total_prepaid_payable_amount_node(self, vals, in_foreign_currency=True):
currency = vals['currency_id'] if in_foreign_currency else vals['company_currency']
node = vals['legal_monetary_total_node']
payable_rounding_amount = (node['cbc:PayableRoundingAmount'] or {}).get('_text') or 0.0
node['cbc:PrepaidAmount'] = {
'_text': FloatFmt(0.0, min_dp=currency.decimal_places),
'currencyID': currency.name,
}
node['cbc:PayableAmount'] = {
'_text': FloatFmt(
node['cbc:TaxInclusiveAmount']['_text']
+ payable_rounding_amount,
min_dp=currency.decimal_places,
),
'currencyID': currency.name,
}
if self._is_document(vals, 'invoice', 'credit_note', 'self_invoice', 'self_credit_note'):
invoice = vals['invoice']
if in_foreign_currency:
amount_total = invoice.amount_total
amount_residual = invoice.amount_residual
else:
amount_total = invoice.amount_total_signed * -invoice.direction_sign
amount_residual = invoice.amount_residual_signed * -invoice.direction_sign
node['cbc:PayableAmount']['_text'] = FloatFmt(
amount_residual,
min_dp=currency.decimal_places,
)
node['cbc:PrepaidAmount']['_text'] = FloatFmt(
amount_total - amount_residual,
min_dp=currency.decimal_places,
)
def _ubl_add_legal_monetary_total_payable_rounding_amount_node_from_cash_rounding(self, vals, in_foreign_currency=True):
# DEPRECATED: TO BE REMOVED
pass
def _ubl_add_legal_monetary_total_payable_rounding_amount_node(self, vals):
AccountTax = self.env['account.tax']
base_lines = vals['base_lines']
currency = vals['currency_id']
node = vals['legal_monetary_total_node']
tax_inclusive_amount = node['cbc:TaxInclusiveAmount']['_text']
base_lines_aggregated_values = AccountTax._aggregate_base_lines_tax_details(
base_lines=base_lines,
grouping_function=lambda base_line, tax_data: True,
)
values_per_grouping_key = AccountTax._aggregate_base_lines_aggregated_values(base_lines_aggregated_values)
expected_tax_inclusive_amount = sum(
values['total_excluded_currency'] + values['tax_amount_currency']
for values in values_per_grouping_key.values()
)
payable_rounding_amount = expected_tax_inclusive_amount - tax_inclusive_amount
if currency.is_zero(payable_rounding_amount):
node['cbc:PayableRoundingAmount'] = {
'_text': None,
'currencyID': None,
}
else:
node['cbc:PayableRoundingAmount'] = {
'_text': FloatFmt(payable_rounding_amount, min_dp=currency.decimal_places),
'currencyID': currency.name,
}
def _ubl_add_legal_monetary_total_node(self, vals):
node = vals['document_node']['cac:LegalMonetaryTotal'] = {}
sub_vals = {**vals, 'legal_monetary_total_node': node}
self._ubl_add_legal_monetary_total_line_extension_amount_node(sub_vals)
self._ubl_add_legal_monetary_total_tax_exclusive_amount_node(sub_vals)
self._ubl_add_legal_monetary_total_tax_inclusive_amount_node(sub_vals)
self._ubl_add_legal_monetary_total_allowance_charge_total_amount_node(sub_vals)
self._ubl_add_legal_monetary_total_payable_rounding_amount_node(sub_vals)
self._ubl_add_legal_monetary_total_prepaid_payable_amount_node(sub_vals)
def _ubl_add_requested_monetary_total_node(self, vals):
node = vals['document_node']['cac:RequestedMonetaryTotal'] = {}
sub_vals = {**vals, 'legal_monetary_total_node': node}
self._ubl_add_legal_monetary_total_line_extension_amount_node(sub_vals)
self._ubl_add_legal_monetary_total_tax_exclusive_amount_node(sub_vals)
self._ubl_add_legal_monetary_total_tax_inclusive_amount_node(sub_vals)
self._ubl_add_legal_monetary_total_allowance_charge_total_amount_node(sub_vals)
self._ubl_add_legal_monetary_total_payable_rounding_amount_node(sub_vals)
self._ubl_add_legal_monetary_total_prepaid_payable_amount_node(sub_vals)
def _fill_document_values_invoice(self, vals):
document_node = vals['document_node']
document_node['_template'] = Invoice
document_node['_nsmap'][None] = "urn:oasis:names:specification:ubl:schema:xsd:Invoice-2"
self._ubl_add_version_id_node(vals)
self._ubl_add_customization_id_node(vals)
self._ubl_add_profile_id_node(vals)
self._ubl_add_invoice_period_nodes(vals)
self._ubl_add_id_node(vals)
self._ubl_add_issue_date_node(vals)
self._ubl_add_due_date_node(vals)
self._ubl_add_invoice_type_code_node(vals)
self._ubl_add_notes_nodes(vals)
self._ubl_add_document_currency_code_node(vals)
self._ubl_add_tax_currency_code_node(vals)
self._ubl_add_buyer_reference_node(vals)
self._ubl_add_order_reference_node(vals)
self._ubl_add_accounting_supplier_party_node(vals)
self._ubl_add_accounting_customer_party_node(vals)
self._ubl_add_delivery_nodes(vals)
self._ubl_add_payment_means_nodes(vals)
self._ubl_add_payment_terms_nodes(vals)
self._ubl_add_allowance_charge_nodes(vals)
self._ubl_add_invoice_line_nodes(vals)
self._ubl_add_tax_totals_nodes(vals)
self._ubl_add_legal_monetary_total_node(vals)
def _fill_document_values_credit_note(self, vals):
document_node = vals['document_node']
document_node['_template'] = CreditNote
document_node['_nsmap'][None] = "urn:oasis:names:specification:ubl:schema:xsd:CreditNote-2"
self._ubl_add_version_id_node(vals)
self._ubl_add_customization_id_node(vals)
self._ubl_add_profile_id_node(vals)
self._ubl_add_invoice_period_nodes(vals)
self._ubl_add_id_node(vals)
self._ubl_add_issue_date_node(vals)
self._ubl_add_credit_note_type_code_node(vals)
self._ubl_add_notes_nodes(vals)
self._ubl_add_document_currency_code_node(vals)
self._ubl_add_tax_currency_code_node(vals)
self._ubl_add_buyer_reference_node(vals)
self._ubl_add_order_reference_node(vals)
self._ubl_add_accounting_supplier_party_node(vals)
self._ubl_add_accounting_customer_party_node(vals)
self._ubl_add_delivery_nodes(vals)
self._ubl_add_payment_means_nodes(vals)
self._ubl_add_payment_terms_nodes(vals)
self._ubl_add_allowance_charge_nodes(vals)
self._ubl_add_credit_note_line_nodes(vals)
self._ubl_add_tax_totals_nodes(vals)
self._ubl_add_legal_monetary_total_node(vals)
def _fill_document_values_debit_note(self, vals):
document_node = vals['document_node']
document_node['_template'] = DebitNote
document_node['_nsmap'][None] = "urn:oasis:names:specification:ubl:schema:xsd:DebitNote-2"
self._ubl_add_version_id_node(vals)
self._ubl_add_customization_id_node(vals)
self._ubl_add_profile_id_node(vals)
self._ubl_add_invoice_period_nodes(vals)
self._ubl_add_id_node(vals)
self._ubl_add_issue_date_node(vals)
self._ubl_add_notes_nodes(vals)
self._ubl_add_document_currency_code_node(vals)
self._ubl_add_tax_currency_code_node(vals)
self._ubl_add_buyer_reference_node(vals)
self._ubl_add_order_reference_node(vals)
self._ubl_add_accounting_supplier_party_node(vals)
self._ubl_add_accounting_customer_party_node(vals)
self._ubl_add_delivery_nodes(vals)
self._ubl_add_payment_means_nodes(vals)
self._ubl_add_payment_terms_nodes(vals)
self._ubl_add_allowance_charge_nodes(vals)
self._ubl_add_debit_note_line_nodes(vals)
self._ubl_add_tax_totals_nodes(vals)
self._ubl_add_requested_monetary_total_node(vals)
def _fill_document_values(self, vals):
document_node = vals['document_node']
document_node['_nsmap']['cac'] = "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"
document_node['_nsmap']['cbc'] = "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2"
document_node['_nsmap']['ext'] = "urn:oasis:names:specification:ubl:schema:xsd:CommonExtensionComponents-2"
if self._is_document(vals, 'invoice', 'self_invoice'):
self._fill_document_values_invoice(vals)
elif self._is_document(vals, 'credit_note', 'self_credit_note'):
self._fill_document_values_credit_note(vals)
elif self._is_document(vals, 'debit_note'):
self._fill_document_values_debit_note(vals)
def _export_document_node_constraints(self, vals):
return {}
def _export_document(self, vals):
vals['document_node'] = {
'_nsmap': {},
'_template': Invoice,
}
self._fill_document_values(vals)
vals['constraints'] = {
k: v
for k, v in self._export_document_node_constraints(vals).items()
if v
}
return vals
def _ubl_add_values_document_type(self, vals):
invoice = vals['invoice']
if invoice.move_type == 'out_invoice':
document_type = 'invoice'
elif invoice.move_type == 'out_refund':
document_type = 'credit_note'
elif invoice.move_type == 'in_invoice':
document_type = 'self_invoice'
elif invoice.move_type == 'in_refund':
document_type = 'self_credit_note'
self._define_document_type(vals, document_type)
def _init_invoice_export_values(self, invoice):
vals = {'invoice': invoice.with_context(lang=invoice.partner_id.lang)}
self._ubl_add_values_document_type(vals)
self._ubl_add_values_company(vals, invoice.company_id)
self._ubl_add_values_currency(vals, invoice.currency_id)
if self._is_document(vals, 'invoice', 'credit_note'):
customer = invoice.partner_id
supplier = invoice.company_id.partner_id
delivery = invoice.partner_shipping_id or customer
elif self._is_document(vals, 'self_invoice', 'self_credit_note'):
customer = invoice.company_id.partner_id
supplier = invoice.partner_id
delivery = customer.child_ids.filtered(lambda p: p.type == 'delivery')[:1] or customer
self._ubl_add_values_supplier(vals, supplier)
self._ubl_add_values_customer(vals, customer)
self._ubl_add_values_delivery(vals, delivery)
vals['base_lines'], vals['tax_lines'] = invoice._get_rounded_base_and_tax_lines()
return vals
def _export_invoice(self, invoice):
vals = self._init_invoice_export_values(invoice)
return self._export_document(vals)
# -------------------------------------------------------------------------
# IMPORT: INVOICE
# -------------------------------------------------------------------------
def _import_ubl_invoice_document_sign(self, collected_values):
tree = collected_values['tree']
suffix_invoice_type, document_sign = self._get_import_document_amount_sign(tree)
collected_values['is_refund'] = suffix_invoice_type == 'refund'
collected_values['file_document_sign'] = document_sign
def _import_ubl_invoice_update_move_type(self, collected_values):
invoice = collected_values['invoice']
odoo_document_type = collected_values['odoo_document_type']
is_refund = collected_values['is_refund']
logs = collected_values['logs']
prefix = 'out' if odoo_document_type == 'sale' else 'in'
suffix = 'refund' if is_refund else 'invoice'
move_type = f'{prefix}_{suffix}'
if invoice.move_type != move_type:
invoice.move_type = move_type
if is_refund:
logs.append(_("The invoice has been converted into a credit note and the quantities have been reverted."))
def _import_ubl_invoice_add_customer_values(self, collected_values):
customer_values = collected_values['customer_values'] = {}
odoo_document_type = collected_values['odoo_document_type']
party_tag = "AccountingCustomerParty" if odoo_document_type == 'sale' else "AccountingSupplierParty"
tree = collected_values['tree']
party_node = tree.find(f"./{{*}}{party_tag}/{{*}}Party")
if party_node is None:
return
for key, xpaths in (
('vat', [".//{*}CompanyID"]),
('phone', [".//{*}Telephone"]),
('name', [
".//{*}RegistrationName",
".//{*}Name",
]),
('email', [".//{*}ElectronicMail"]),
('country_code', [".//{*}Country//{*}IdentificationCode"]),
('street', [".//{*}StreetName"]),
('street2', [".//{*}AdditionalStreetName"]),
('city', [".//{*}CityName"]),
('zip', [".//{*}PostalZone"]),
):
customer_values[key] = None
for xpath in xpaths:
if (node := party_node.find(xpath)) is not None:
customer_values[key] = node.text
break
# Peppol EAS/Endpoint.
if (node := party_node.find(".//{*}EndpointID")) is not None and node.text:
customer_values['peppol_endpoint'] = node.text.strip()
if peppol_eas := node.attrib.get('schemeID'):
customer_values['peppol_eas'] = peppol_eas
if not customer_values['vat'] and (country_code := customer_values.get('country_code')):
for scheme_id, field in EAS_MAPPING.get(country_code, {}).items():
if field == 'vat' and (vat := party_node.findtext(f".//{{*}}PartyIdentification/{{*}}ID[@schemeID='{scheme_id}']")):
customer_values['vat'] = vat
break
def _import_ubl_retrieve_customer_search_plan(self, collected_values):
ResPartner = self.env['res.partner']
return [
ResPartner._import_retrieve_customer_from_vat,
ResPartner._import_retrieve_customer_from_eas_endpoint,
ResPartner._import_retrieve_customer_from_bank_account_number,
ResPartner._import_retrieve_customer_from_email,
ResPartner._import_retrieve_customer_from_phone,
ResPartner._import_retrieve_customer_from_name,
]
def _import_ubl_retrieve_customer(self, collected_values):
company = collected_values['company']
customer_values = collected_values['customer_values']
customer_values.update({
'account_numbers': collected_values.get('partner_bank_values', {}).get('account_numbers'),
'company': company,
})
self.env['res.partner']._import_retrieve_customer(
search_plan=self._import_ubl_retrieve_customer_search_plan(collected_values),
company=company,
customer_values_list=[customer_values],
)
if partner := customer_values.get('customer'):
collected_values['to_write']['partner_id'] = partner.id
def _import_ubl_get_country(self, collected_values):
customer_values = collected_values['customer_values']
country_code = customer_values.get('country_code')
if not country_code:
return None
if country_code == 'GB':
# While the code is gb, the xml_id is uk
country_code = 'UK'
return self.env.ref(f'base.{country_code.lower()}', raise_if_not_found=False)
def _import_ubl_prepare_missing_customer_create_values(self, collected_values):
customer_values = collected_values['customer_values']
partner_create_values = {
'is_company': True,
}
for key in ('phone', 'name', 'email', 'street', 'street2', 'zip', 'city'):
if value := customer_values.get(key):
partner_create_values[key] = value
if (peppol_eas := customer_values.get('peppol_eas')) and (peppol_endpoint := customer_values.get('peppol_endpoint')):
partner_create_values['peppol_eas'] = peppol_eas
partner_create_values['peppol_endpoint'] = peppol_endpoint
country = self._import_ubl_get_country(collected_values)
if country:
partner_create_values['country_id'] = country.id
if vat := customer_values.get('vat'):
partner_create_values['vat'], _country_code = self.env['res.partner']._run_vat_checks(country, vat, validation='setnull')
return partner_create_values
def _check_customer_vat_match(self, customer, vat, collected_values):
"""
Compare the VAT from an EDI document against a partner's stored VAT,
with country-specific normalization where needed.
Should stay consistent with `_get_country_specific_vat_variants`.
"""
country = self._import_ubl_get_country(collected_values)
customer_vat = customer.vat.replace(' ', '').upper()
vat_to_compare = vat.replace(' ', '').replace('.', '').upper()
if country.code == 'CH':
customer_vat = re.sub(r"(TVA|IVA|MWST)?$", "", customer_vat.replace('.', '').replace('-', ''))
vat_to_compare = re.sub(r"(TVA|IVA|MWST)?$", "", vat_to_compare.replace('-', ''))
return customer_vat == vat_to_compare
def _import_ubl_create_missing_customer(self, collected_values):
customer_values = collected_values['customer_values']
logs = collected_values['logs']
customer = customer_values.get('customer')
name = customer_values.get('name')
vat = customer_values.get('vat')
if not name or not vat:
return
vat_mismatch = False
if customer:
if not customer.vat:
country = self._import_ubl_get_country(collected_values)
customer.vat, _country_code = self.env['res.partner']._run_vat_checks(country, vat, validation='setnull')
return
if self._check_customer_vat_match(customer, vat, collected_values):
return
vat_mismatch = True
partner_create_values = self._import_ubl_prepare_missing_customer_create_values(collected_values)
customer = self.env['res.partner'].create(partner_create_values)
if vat_mismatch:
logs.append(_("Could not retrieve a partner corresponding to '%s' with the same VAT. A new partner was created.", name))
else:
logs.append(_("Could not retrieve a partner corresponding to '%s'. A new partner was created.", name))
customer_values['customer'] = customer
collected_values['to_write']['partner_id'] = customer.id
def _import_ubl_invoice_add_currency_values(self, collected_values):
currency_values = collected_values['currency_values'] = {}
tree = collected_values['tree']
currency_values['currency_code'] = tree.findtext('.//{*}DocumentCurrencyCode')
def _import_ubl_invoice_add_currency(self, collected_values):
currency_values = collected_values['currency_values']
logs = collected_values['logs']
company = collected_values['company']
currency = company.currency_id
if currency_code := currency_values['currency_code']:
currency = currency.with_context(active_test=False).search([('name', '=', currency_code)], limit=1)
if currency:
if not currency.active:
logs.append(_("The currency '%s' is not active.", currency.name))
else:
logs.append(_(
"Could not retrieve currency: %s. Did you enable the multicurrency option "
"and activate the currency?",
currency_code,
))
currency_values['currency'] = currency
if invoice_date := collected_values['to_write'].get('invoice_date'):
currency_date = invoice_date
else:
currency_date = fields.Date.context_today(self)
currency_values['rate'] = currency._get_conversion_rate(
from_currency=company.currency_id,
to_currency=currency,
company=company,
date=currency_date,
)
collected_values['to_write']['currency_id'] = currency.id
def _import_ubl_invoice_add_issue_date(self, collected_values):
tree = collected_values['tree']
issue_date_str = tree.findtext('./{*}IssueDate')
if issue_date_str:
collected_values['to_write']['invoice_date'] = fields.Date.from_string(issue_date_str)
def _import_ubl_invoice_add_due_date(self, collected_values):
tree = collected_values['tree']
due_date_str = tree.findtext('./{*}DueDate')
if not due_date_str:
due_date_str = tree.findtext('./{*}PaymentDueDate')
if due_date_str:
collected_values['to_write']['invoice_date_due'] = fields.Date.from_string(due_date_str)
def _import_ubl_invoice_add_partner_bank_values(self, collected_values):
partner_bank_values = collected_values['partner_bank_values'] = {}
tree = collected_values['tree']
party_node_list = tree.findall("./{*}PaymentMeans/{*}PayeeFinancialAccount/{*}ID")
partner_bank_values['account_numbers'] = {party_node.text for party_node in party_node_list}
def _import_ubl_retrieve_partner_bank(self, collected_values):
company = collected_values['company']
move_type = collected_values['invoice'].move_type
if move_type in ('out_refund', 'in_invoice'):
partner = collected_values.get('customer_values', {}).get('customer')
elif move_type in ('out_invoice', 'in_refund'):
partner = company.partner_id
else:
return
if not partner:
return
partner_bank_values = collected_values['partner_bank_values']
account_numbers = partner_bank_values['account_numbers']
logs = collected_values['logs']
partner_banks = self.env['res.partner.bank']
for account_number in account_numbers:
try:
partner_banks += self.env['res.partner.bank']._find_or_create_bank_account(
account_number=account_number,
partner=partner,
company=company,
)
except UserError as e:
logs.append(_("The bank account couldn't be fetched: %s", str(e)))
partner_bank_values['partner_banks'] = partner_banks
if partner_banks:
collected_values['to_write']['partner_bank_id'] = partner_banks[:1].id
def _import_ubl_invoice_add_ref(self, collected_values):
tree = collected_values['tree']
odoo_document_type = collected_values['odoo_document_type']
if ref := tree.findtext('./{*}ID'):
if odoo_document_type == 'sale' and collected_values['invoice'].quick_edit_mode:
collected_values['to_write']['name'] = ref
else:
collected_values['to_write']['ref'] = ref
def _import_ubl_invoice_add_invoice_origin(self, collected_values):
tree = collected_values['tree']
invoice_origin = tree.findtext('./{*}OrderReference/{*}ID')
if not invoice_origin and self.module_installed('purchase'):
sequence = self.env['ir.sequence'].search([
('code', '=', 'purchase.order'),
('company_id', 'in', [collected_values['company'].id, False]),
], order='company_id', limit=1)
if sequence:
prefix, suffix = sequence._get_prefix_suffix()
pattern = f'{prefix}\\d{{{sequence.padding}}}{suffix}'
refs = re.findall(pattern, ' '.join([desc.text for desc in tree.findall('.//{*}Item/{*}Description') if desc.text]))
invoice_origin = ' '.join(refs)
if invoice_origin:
collected_values['to_write']['invoice_origin'] = invoice_origin
def _import_ubl_invoice_add_narration(self, collected_values):
if narration := ''.join(f'{note}
' for note in self._get_notes(collected_values)):
collected_values['to_write']['narration'] = narration
@api.model
def _get_notes(self, collected_values):
tree = collected_values["tree"]
nodes = tree.findall("./{*}Note") + tree.findall("./{*}PaymentTerms/{*}Note")
return [html_escape(node.text) for node in nodes if node.text]
def _import_ubl_invoice_add_payment_reference(self, collected_values):
tree = collected_values['tree']
payment_references = []
for node in tree.findall('./{*}PaymentMeans/{*}PaymentID'):
if note := node.text:
payment_references.append(note)
if payment_reference := ','.join(unique(payment_references)):
collected_values['to_write']['payment_reference'] = payment_reference
def _import_ubl_invoice_add_delivery(self, collected_values):
tree = collected_values['tree']
delivery_date_str = tree.findtext('.//{*}Delivery/{*}ActualDeliveryDate')
if delivery_date_str:
collected_values['to_write']['delivery_date'] = fields.Date.from_string(delivery_date_str)
def _import_ubl_invoice_add_incoterm_values(self, collected_values):
incoterm_values = collected_values['incoterm_values'] = {}
codes = incoterm_values['codes'] = set()
tree = collected_values['tree']
for node in tree.findall('./{*}TransportExecutionTerms/{*}DeliveryTerms/{*}ID'):
if text := node.text:
codes.add(text)
def _import_ubl_invoice_add_incoterm(self, collected_values):
incoterm_values = collected_values['incoterm_values']
codes = incoterm_values['codes']
if len(codes) == 1:
code = next(iter(codes))
incoterm = self.env['account.incoterms'].search([('code', '=', code)], limit=1)
else:
incoterm = None
incoterm_values['incoterm'] = incoterm
if incoterm:
collected_values['to_write']['invoice_incoterm_id'] = incoterm.id
def _import_ubl_invoice_add_prepaid_amount(self, collected_values):
file_document_sign = collected_values['file_document_sign']
currency = collected_values['currency_values']['currency']
tree = collected_values['tree']
prepaid_amount_str = tree.findtext('./{*}LegalMonetaryTotal/{*}PrepaidAmount')
prepaid_amount = file_document_sign * float(prepaid_amount_str or 0.0)
if currency.is_zero(prepaid_amount):
return
collected_values['prepaid_amount'] = prepaid_amount
formatted_prepaid_amount = formatLang(self.env, prepaid_amount, currency_obj=currency)
collected_values['logs'].append(_("A payment of %s was detected.", formatted_prepaid_amount))
def _import_ubl_invoice_add_tax_total_values(self, collected_values):
file_document_sign = collected_values['file_document_sign']
odoo_document_type = collected_values['odoo_document_type']
currency = collected_values['currency_values']['currency']
taxes_values = collected_values['tax_total_values'] = {}
tree = collected_values['tree']
for subtotal_elem in tree.findall('./{*}TaxTotal/{*}TaxSubtotal'):
amount_node = subtotal_elem.find('.//{*}TaxAmount')
category_code = subtotal_elem.findtext('.//{*}TaxCategory/{*}ID')
if amount_node is None or category_code is None:
continue
amount = amount_node.text
if amount_node.get('currencyID') and amount_node.get('currencyID') != currency.name:
continue
percentage = subtotal_elem.findtext('.//{*}TaxCategory/{*}Percent')
if percentage is None:
percentage = subtotal_elem.findtext('.//{*}Percent')
if percentage is None:
continue
percentage = float(percentage)
tax_key = frozendict({
'category_code': category_code,
'percentage': percentage,
})
tax_values = taxes_values.setdefault(tax_key, {
'amount_type': 'percent',
'type_tax_use': odoo_document_type,
'amount': percentage,
'ubl_cii_tax_category_code': category_code,
'tax_amount_currency': 0.0,
'related_taxes_values': [],
})
tax_values['tax_amount_currency'] += file_document_sign * float(amount)
def _import_ubl_invoice_add_allowances_charges_values(self, collected_values):
tree = collected_values['tree']
odoo_document_type = collected_values['odoo_document_type']
allowances = collected_values['allowances'] = []
charges = collected_values['charges'] = []
taxes_values = collected_values['taxes_values'] = []
for element in tree.iterfind('./{*}AllowanceCharge'):
reason = element.findtext('./{*}AllowanceChargeReason')
reason_code = element.findtext('./{*}AllowanceChargeReasonCode')
charge_indicator = element.findtext('./{*}ChargeIndicator')
amount_str = element.findtext('./{*}Amount')
base_amount_str = element.findtext('./{*}BaseAmount')
multiplier_factor_numeric_str = element.findtext('./{*}MultiplierFactorNumeric')
percentage_str = element.findtext('./{*}TaxCategory/{*}Percent')
category_code = element.findtext('./{*}TaxCategory/{*}ID')
if amount_str:
amount = float(amount_str)
else:
amount = 0.0
if not percentage_str:
continue
percentage = float(percentage_str)
allowance_charge_values = {
'amount': amount,
'base_amount': float(base_amount_str) if base_amount_str else None,
'reason': reason,
'reason_code': reason_code,
'multiplier_factor_numeric': float(multiplier_factor_numeric_str) if multiplier_factor_numeric_str else None,
'tax_percentage': percentage,
'charge_indicator': charge_indicator,
}
if charge_indicator.lower() == 'true':
charges.append(allowance_charge_values)
else:
allowances.append(allowance_charge_values)
# Try to link the allowance / charge with a percentage tax.
if not category_code:
continue
allowance_charge_values['attempt_tax_values'] = tax_values = {
'amount_type': 'percent',
'type_tax_use': odoo_document_type,
'ubl_cii_tax_category_code': category_code,
'amount': percentage,
}
taxes_values.append(tax_values)
def _import_ubl_invoice_line_add_name(self, collected_values):
"""
In UBL, both Name and Description elements are optional
An item may contain multiple Description elements
"""
line_tree = collected_values['line_tree']
item_ref = line_tree.findtext('.//{*}Item/{*}SellersItemIdentification/{*}ID')
name = line_tree.findtext('.//{*}Item/{*}Name')
if item_ref and name and f'[{item_ref}]' not in name:
name = f"[{item_ref}] {name}"
collected_values['name'] = name
description = ''
for description_elem in line_tree.iterfind('.//{*}Item/{*}Description'):
if description_elem.text:
description += description_elem.text + '\n'
if name and description:
collected_values['to_write']['name'] = f'{name}\n{description.strip()}'
elif name:
collected_values['to_write']['name'] = name
elif description:
collected_values['to_write']['name'] = description.strip()
def _import_ubl_invoice_line_add_allowance_charges_values(self, collected_values):
line_tree = collected_values['line_tree']
allowances = collected_values['allowances'] = []
charges = collected_values['charges'] = []
for allowance_charge_elem in line_tree.iterfind('./{*}AllowanceCharge'):
charge_indicator = allowance_charge_elem.findtext('.//{*}ChargeIndicator')
amount_str = allowance_charge_elem.findtext('.//{*}Amount')
base_amount_str = allowance_charge_elem.findtext('.//{*}BaseAmount')
reason = allowance_charge_elem.findtext('.//{*}AllowanceChargeReason')
reason_code = allowance_charge_elem.findtext('.//{*}AllowanceChargeReasonCode')
if amount_str:
amount = float(amount_str)
else:
continue
allowance_charge_values = {
'amount': amount,
'base_amount': float(base_amount_str) if base_amount_str else None,
'reason': reason,
'reason_code': reason_code,
}
if charge_indicator.lower() == 'true':
charges.append(allowance_charge_values)
else:
allowances.append(allowance_charge_values)
allowance_elem = line_tree.find('./{*}Price/{*}AllowanceCharge')
collected_values['price_allowance_values'] = {}
if allowance_elem is not None:
charge_indicator = allowance_elem.findtext('./{*}ChargeIndicator') or 'false'
amount_str = allowance_elem.findtext('./{*}Amount')
base_amount_str = allowance_elem.findtext('./{*}BaseAmount')
reason = allowance_elem.findtext('./{*}AllowanceChargeReason')
reason_code = allowance_elem.findtext('./{*}AllowanceChargeReasonCode')
if charge_indicator.lower() == 'true':
charge_indicator_sign = 1
else:
charge_indicator_sign = -1
collected_values['price_allowance_values'] = {
'charge_indicator_sign': charge_indicator_sign,
'amount': float(amount_str) if amount_str else None,
'base_amount': float(base_amount_str) if base_amount_str else None,
'reason': reason,
'reason_code': reason_code,
}
def _import_ubl_invoice_line_add_price_unit_quantity_discount(self, collected_values):
file_document_sign = collected_values['file_document_sign']
line_tree = collected_values['line_tree']
currency = collected_values['currency_values']['currency']
line_extension_amount_str = line_tree.findtext('.//{*}LineExtensionAmount')
price_amount_str = line_tree.findtext('.//{*}Price/{*}PriceAmount')
invoiced_quantity_str = (
line_tree.findtext('.//{*}InvoicedQuantity')
or line_tree.findtext('.//{*}CreditedQuantity')
)
base_quantity_str = line_tree.findtext('./{*}Price/{*}BaseQuantity')
line_extension_amount = line_extension_amount_str and float(line_extension_amount_str) * file_document_sign
price_amount = price_amount_str and float(price_amount_str)
invoiced_quantity = invoiced_quantity_str and float(invoiced_quantity_str) * file_document_sign
base_quantity = base_quantity_str and float(base_quantity_str)
total_allowances = sum(allowance['amount'] for allowance in collected_values['allowances'])
total_charges = sum(charge['amount'] for charge in collected_values['charges'])
price_allowance_values = collected_values['price_allowance_values']
price_allowance_base_amount = price_allowance_values.get('base_amount')
price_allowance_amount = price_allowance_values.get('amount')
if price_allowance_amount and (price_allowance_charge_indicator_sign := price_allowance_values.get('charge_indicator_sign')):
price_allowance_amount *= price_allowance_charge_indicator_sign
subtotal = (line_extension_amount or 0.0) + total_allowances - total_charges
# Price level.
# Define at the product level the price for which quantity and how many discount you get
# by buying it
if price_amount:
price_quantity = base_quantity or 1.0
if price_allowance_base_amount:
price_discount_amount = price_allowance_base_amount - price_amount
price_subtotal = price_allowance_base_amount
elif price_allowance_amount:
price_discount_amount = -price_allowance_amount
price_subtotal = price_amount - price_allowance_amount
else:
price_discount_amount = 0.0
price_subtotal = price_amount
elif price_allowance_base_amount:
price_subtotal = price_allowance_base_amount
price_quantity = base_quantity or 1.0
price_discount_amount = -(price_allowance_amount or 0.0)
else:
price_subtotal = 0.0
price_quantity = 0.0
price_discount_amount = 0.0
# Line level.
if (
line_extension_amount is not None
and not invoiced_quantity
):
price_unit = subtotal
quantity = 1.0
discount_amount = total_allowances
# Combine with the price level. Suppose:
# line_extension_amount = 1000.0
# price_subtotal = 1250.0
# price_quantity = 5.0
# price_discount_amount = 250.0
# In that case, we want to compute:
# price_unit = 250.0
# quantity = 5.0
# discount_amount = 250.0
if not currency.is_zero(price_subtotal):
quantity = subtotal * price_quantity / (price_subtotal - price_discount_amount)
if quantity:
price_unit = (subtotal / quantity) + (price_discount_amount / price_quantity)
else:
price_unit = price_amount
discount_amount += price_discount_amount * quantity / price_quantity
elif (
line_extension_amount is not None
and invoiced_quantity
):
quantity = invoiced_quantity
price_unit = subtotal / quantity
discount_amount = total_allowances
# Combine with the price level. Suppose:
# line_extension_amount = 1200.0
# quantity = 6
# price_subtotal = 1250.0
# price_quantity = 5.0
# price_discount_amount = 50.0
# In that case, we want to compute:
# price_unit = 250.0
# quantity = 6.0
# discount_amount = 300.0
if not currency.is_zero(price_subtotal):
price_unit = price_subtotal / price_quantity
discount_amount += price_discount_amount * quantity / price_quantity
else:
quantity = 0.0
price_unit = 0.0
discount_amount = total_allowances
# Combine with the price level.
if not currency.is_zero(price_subtotal):
price_unit = price_subtotal / price_quantity
quantity = price_quantity
discount_amount += price_discount_amount
# Extra charges.
price_unit += total_charges / (quantity or 1.0)
# Turn discount_amount to a percentage
gross_subtotal = price_unit * quantity
discount = (discount_amount * 100 / gross_subtotal) if gross_subtotal else 0.0
to_write = collected_values['to_write']
to_write['quantity'] = quantity
to_write['price_unit'] = price_unit
to_write['discount'] = discount
def _import_ubl_invoice_line_add_vehicle_values(self, collected_values):
if not self.module_installed('account_fleet'):
return
tree = collected_values['tree']
line_tree = collected_values['line_tree']
default_parent_node_path = './{*}Item/{*}AdditionalItemProperty'
default_value_path = './{*}Value'
default_linked_field = 'vin_sn'
def default_condition(parent_node, node, value):
return parent_node.findtext('./{*}Name') == value
paths = [
# {
# 'path_type': 'line' or 'move'
# 'parent_node_path': 'path to the parent node',
# 'condition': lambda parent_node, node, value: 'where parent_node = parent node, node = node containing VIN Number, value = identifier',
# 'value_path': 'path to the node where the information is to be found',
# 'identifier': 'to be used in condition to perform a check, inner tex of a node allowing to identify the node to read',
# 'linked_field': the field to search in DB (vin_sn, license_plate),
# 'pattern': if the value to search is not in a field specific to it (with other words like in a description)
# }
{'path_type': 'line', 'identifier': 'SerialNumber'}, # VIN in AdditionalItemProperty/Value with AdditionalItemProperty/Name == 'SerialNumber'
{'path_type': 'line', 'identifier': 'VIN'}, # VIN in AdditionalItemProperty/Value with AdditionalItemProperty/Name == 'VIN'
{'path_type': 'line', 'identifier': 'PlateNumber', 'linked_field': 'license_plate'}, # LICENSE PLATE in AdditionalItemProperty/Value with AdditionalItemProperty/Name == 'PlateNumber'
{'path_type': 'line', 'identifier': 'LCPL-NO', 'linked_field': 'license_plate'}, # LICENSE PLATE in AdditionalItemProperty/Value with AdditionalItemProperty/Name == 'LCPL-NO'
{
# VIN in Item/Description
'path_type': 'line',
'parent_node_path': './{*}Item',
'condition': lambda parent_node, node, value: True,
'value_path': './{*}Description',
'pattern': r'[A-Za-z0-9]{17}',
},
{
# LICENSE PLATE in Item/Description
'path_type': 'line',
'parent_node_path': './{*}Item',
'condition': lambda parent_node, node, value: True,
'value_path': './{*}Description',
'linked_field': 'license_plate',
'pattern': r'\d-[A-Za-z]{3}-\d{3}', # BE license plate format
},
{
# VIN in AdditionalDocumentReference/ID with schemeID == 'AKG' (1 vin for the whole invoice)
'path_type': 'move',
'parent_node_path': './{*}AdditionalDocumentReference',
'condition': lambda parent_node, node, value: node.get('schemeID') == 'AKG',
'value_path': './{*}ID',
},
]
results = defaultdict(set) # {field (vin_sn|license_plate): {'AZERTYUIOP', 'POIUYTREZA'}}
for path in paths:
local_tree = tree if path['path_type'] == 'move' else line_tree
parent_nodes = local_tree.findall(path.get('parent_node_path', default_parent_node_path))
for parent_node in parent_nodes:
value_node = parent_node.find(path.get('value_path', default_value_path))
if value_node is None or not path.get('condition', default_condition)(parent_node, value_node, path.get('identifier', '')):
continue
value = value_node.text
if value is None:
continue
if path.get('pattern'):
# we need to find the car identifier in the node
if candidates := re.findall(path['pattern'], value or ''):
results[path.get('linked_field', default_linked_field)].update(candidates)
else:
# the car identifier is the full text of the node
results[path.get('linked_field', default_linked_field)].add(value)
if not results or any(len(vals) != 1 for vals in results.values()):
return
collected_values['vehicle_values'] = tuple((field, value.pop()) for field, value in results.items())
def _import_ubl_invoice_line_add_product_values(self, collected_values):
line_tree = collected_values['line_tree']
partner = collected_values.get('customer_values', {}).get('customer')
name = collected_values['to_write'].get('name')
sellers_item_id = line_tree.findtext('.//{*}Item/{*}SellersItemIdentification/{*}ID')
buyers_item_id = line_tree.findtext('.//{*}Item/{*}BuyersItemIdentification/{*}ID')
standard_item_id = line_tree.findtext('.//{*}Item/{*}StandardItemIdentification/{*}ID[@schemeID="0160"]')
product_values = collected_values['product_values'] = {
'barcode': standard_item_id,
'default_code': sellers_item_id or buyers_item_id,
'name': line_tree.findtext('.//{*}Item/{*}Name'),
'sellers_item_id': sellers_item_id,
'buyers_item_id': buyers_item_id,
'standard_item_id': standard_item_id,
'vendor_partner_id': partner.commercial_partner_id.id if partner else None,
'invoice_predictive': {
'invoice': collected_values['invoice'],
'name': name,
'partner': partner or self.env['res.partner'],
},
}
# CommodityClassification
for commodity_tree in line_tree.findall('./{*}Item/{*}CommodityClassification/{*}ItemClassificationCode'):
list_id = commodity_tree.attrib.get('listID')
code = commodity_tree.text
if not list_id or not code:
continue
if list_id == 'HS':
product_values['intrastat_code'] = code
elif list_id == 'TST':
product_values['unspsc_code'] = code
elif list_id == 'STI':
product_values['l10n_ro_cpv_code'] = code
elif list_id == 'CG':
product_values['cg_item_classification_code'] = code
def _import_ubl_invoice_line_add_product_uom_values(self, collected_values):
line_tree = collected_values['line_tree']
product_uom_values = collected_values['product_uom_values'] = {}
quantity_node = line_tree.find('.//{*}InvoicedQuantity')
if quantity_node is None:
quantity_node = line_tree.find('.//{*}CreditedQuantity')
if quantity_node is not None:
if uom_code := quantity_node.attrib.get('unitCode'):
product_uom_values['uom_code'] = uom_code
def _import_ubl_invoice_line_add_account_values(self, collected_values):
account_values = collected_values['account_values'] = {}
partner = collected_values.get('customer_values', {}).get('customer')
name = collected_values['to_write'].get('name')
account_values['invoice_predictive'] = {
'invoice': collected_values['invoice'],
'name': name,
'partner': partner or self.env['res.partner'],
}
def _import_ubl_invoice_line_add_deferred_dates(self, collected_values):
# DEPRECATED
if not self.module_installed('account_accountant'):
return
line_tree = collected_values['line_tree']
start_date_str = line_tree.findtext('./{*}InvoicePeriod/{*}StartDate')
end_date_str = line_tree.findtext('./{*}InvoicePeriod/{*}EndDate')
if start_date_str and end_date_str:
to_write = collected_values['to_write']
to_write['deferred_start_date'] = fields.Date.from_string(start_date_str)
to_write['deferred_end_date'] = fields.Date.from_string(end_date_str)
def _import_ubl_invoice_line_prepare_classified_tax_category_tax_values(self, collected_values, tax_category_tree):
percentage = tax_category_tree.findtext('./{*}Percent')
category_code = tax_category_tree.findtext('./{*}ID')
if percentage is None or category_code is None:
return
percentage = float(percentage)
tax_key = frozendict({
'category_code': category_code,
'percentage': percentage,
})
global_tax_values = collected_values['tax_total_values'].get(tax_key)
if not global_tax_values:
return
tax_values = {
'amount_type': global_tax_values['amount_type'],
'type_tax_use': global_tax_values['type_tax_use'],
'amount': global_tax_values['amount'],
'ubl_cii_tax_category_code': global_tax_values['ubl_cii_tax_category_code'],
'_tax_key': tax_key,
}
partner = collected_values.get('customer_values', {}).get('customer')
if partner and (name := collected_values['to_write'].get('name')):
tax_values['invoice_predictive'] = {
'invoice': collected_values['invoice'],
'name': name,
'partner': partner,
}
return tax_values
def _import_ubl_invoice_line_prepare_charge_tax_values(self, collected_values, charge):
if charge['reason_code'] != 'AEO':
return
odoo_document_type = collected_values['odoo_document_type']
fixed_tax_amount = charge['amount'] / collected_values['to_write']['quantity']
charge['attempt_tax_values'] = tax_values = {
'name': charge['reason'],
'amount_type': 'fixed',
'type_tax_use': odoo_document_type,
'amount': fixed_tax_amount,
'tax_amount_currency': fixed_tax_amount,
}
partner = collected_values.get('customer_values', {}).get('customer')
if partner and (name := collected_values['to_write'].get('name')):
tax_values['invoice_predictive'] = {
'invoice': collected_values['invoice'],
'name': name,
'partner': partner,
}
return tax_values
def _import_ubl_invoice_line_add_taxes_values(self, collected_values):
line_tree = collected_values['line_tree']
taxes_values = collected_values['taxes_values'] = []
tax_total_values = collected_values['tax_total_values']
# Percentage taxes.
for tax_category_tree in line_tree.findall('.//{*}Item/{*}ClassifiedTaxCategory'):
tax_values = self._import_ubl_invoice_line_prepare_classified_tax_category_tax_values(collected_values, tax_category_tree)
if tax_values:
taxes_values.append(tax_values)
global_tax_values = tax_total_values.get(tax_values['_tax_key'])
global_tax_values['related_taxes_values'].append(tax_values)
# Fixed taxes.
for charge in collected_values['charges']:
tax_values = self._import_ubl_invoice_line_prepare_charge_tax_values(collected_values, charge)
if tax_values:
taxes_values.append(tax_values)
def _import_ubl_invoice_line_add_optional_fields(self, collected_values):
line_tree = collected_values['line_tree']
invoice = collected_values['invoice']
if not invoice.is_purchase_document():
return {}
line_fields = {}
if invoice.move_type == 'in_invoice':
line_fields = PEPPOL_INVOICE_OPTIONAL_LINE_FIELDS
elif invoice.move_type == 'in_refund':
line_fields = PEPPOL_CREDIT_NOTE_OPTIONAL_LINE_FIELDS
vals = {}
AccountMoveLine = self.env['account.move.line']
for field_name, config in line_fields.items():
node_value = self._find_value(
'.//' + '/'.join(config.get('path')),
line_tree,
)
field = AccountMoveLine._fields.get(field_name)
if not field or field.type not in config['supported_types'] or not node_value:
continue
vals[field_name] = node_value
return vals
def _import_ubl_invoice_add_invoice_line_values(self, collected_values):
lines_collected_values = collected_values['lines_collected_values'] = []
tree = collected_values['tree']
for xpath in ('./{*}InvoiceLine', './{*}CreditNoteLine', './{*}DebitNoteLine'):
for line_tree in tree.iterfind(xpath):
line_collected_values = {
**collected_values,
'line_tree': line_tree,
'to_write': {},
}
# Extract information about allowance / charges.
self._import_ubl_invoice_line_add_allowance_charges_values(line_collected_values)
# name / quantity / price_unit / discount
self._import_ubl_invoice_line_add_name(line_collected_values)
self._import_ubl_invoice_line_add_price_unit_quantity_discount(line_collected_values)
# vehicle
# product / product_uom / taxes / accounts /vehicles
self._import_ubl_invoice_line_add_product_values(line_collected_values)
self._import_ubl_invoice_line_add_product_uom_values(line_collected_values)
self._import_ubl_invoice_line_add_account_values(line_collected_values)
self._import_ubl_invoice_line_add_vehicle_values(line_collected_values)
self._import_ubl_invoice_line_add_taxes_values(line_collected_values)
lines_collected_values.append(line_collected_values)
def _import_ubl_retrieve_taxes_search_plan(self, collected_values):
AccountTax = self.env['account.tax']
return [
AccountTax._import_retrieve_tax_from_account_default_tax,
AccountTax._import_retrieve_tax_from_invoice_predictive,
AccountTax._import_retrieve_tax_from_price_include_exclude,
]
def _import_ubl_invoice_retrieve_taxes(self, collected_values):
company = collected_values['company']
logs = collected_values['logs']
lines_collected_values = collected_values['lines_collected_values']
tax_values_list = list(collected_values['taxes_values'])
for line_collected_values in lines_collected_values:
if account := line_collected_values['account_values'].get('account'):
for tax_values in line_collected_values['taxes_values']:
tax_values['account'] = account
tax_values_list += line_collected_values['taxes_values']
for charge in line_collected_values['charges']:
if tax_values := charge.get('attempt_tax_values'):
tax_values_list.append(tax_values)
if customer := collected_values.get('customer_values', {}).get('customer'):
fiscal_position = self.env['account.move'].new({
'company_id': collected_values['company'].id,
'move_type': collected_values['invoice'].move_type,
'partner_id': customer.id,
}).fiscal_position_id
for tax_values in tax_values_list:
tax_values['fiscal_position'] = fiscal_position
self.env['account.tax']._import_retrieve_tax(
search_plan=self._import_ubl_retrieve_taxes_search_plan(collected_values),
company=company,
tax_values_list=tax_values_list,
)
# Taxes at the document line level.
for line_collected_values in lines_collected_values:
to_write = line_collected_values['to_write']
tax_ids_commands = to_write['tax_ids'] = [Command.set([])]
for tax_values in line_collected_values['taxes_values']:
if tax := tax_values.get('tax'):
tax_ids_commands[0][2].append(tax.id)
elif reason := tax_values.get('name'):
logs.append(_(
"Could not retrieve the tax: %(tax_percentage)s %% for line '%(line)s'.",
tax_percentage=tax_values['amount'],
line=reason,
))
else:
logs.append(_(
"Could not retrieve the tax: %s for the document level allowance/charge.",
tax_values['amount'],
))
# Taxes at the document level.
for tax_values in collected_values['taxes_values']:
if tax_values.get('tax'):
continue
if reason := tax_values.get('name'):
logs.append(_(
"Could not retrieve the tax: %(tax_percentage)s %% for line '%(line)s'.",
tax_percentage=tax_values['amount'],
line=reason,
))
else:
logs.append(_(
"Could not retrieve the tax: %s for the document level allowance/charge.",
tax_values['amount'],
))
def _import_ubl_invoice_get_default_base_line_kwargs(self, collected_values):
invoice = collected_values['invoice']
taxes = self.env['account.tax']
for tax_values in collected_values['taxes_values']:
if tax := tax_values.get('tax'):
taxes |= tax
base_line_kwargs = {
'sign': invoice.direction_sign,
'is_refund': invoice.move_type in ('out_refund', 'in_refund'),
'currency_id': collected_values['currency_values']['currency'],
'rate': collected_values['currency_values']['rate'],
'special_mode': 'total_excluded',
'_create_values': {},
}
if partner := collected_values['customer_values'].get('partner'):
base_line_kwargs['partner_id'] = partner
return base_line_kwargs
def _import_ubl_invoice_line_get_product_base_line_kwargs(self, collected_values):
to_write = collected_values['to_write']
taxes = self.env['account.tax']
for tax_values in collected_values['taxes_values']:
if tax := tax_values.get('tax'):
taxes |= tax
base_line_kwargs = {
**self._import_ubl_invoice_get_default_base_line_kwargs(collected_values),
'quantity': to_write['quantity'],
'price_unit': to_write['price_unit'],
'discount': to_write['discount'],
'tax_ids': taxes,
}
if product := collected_values['product_values'].get('product'):
base_line_kwargs['product_id'] = product
if uom := collected_values['product_uom_values'].get('uom'):
base_line_kwargs['product_uom_id'] = uom
elif collected_values['product_uom_values'].get('force_empty'):
# Override the product_uom_id compute so the saved line keeps no UoM.
base_line_kwargs['_create_values']['product_uom_id'] = False
if account := collected_values['account_values'].get('account'):
base_line_kwargs['account_id'] = account
if name := to_write.get('name'):
base_line_kwargs['_create_values']['name'] = name
if vehicle_id := to_write.get('vehicle_id'):
base_line_kwargs['_create_values']['vehicle_id'] = vehicle_id
base_line_kwargs['_create_values'] = {
**base_line_kwargs['_create_values'],
**self._import_ubl_invoice_line_add_optional_fields(collected_values),
}
return base_line_kwargs
def _import_ubl_invoice_get_allowance_charge_line_kwargs(self, collected_values):
allowance_charge = collected_values['allowance_charge']
file_document_sign = collected_values['file_document_sign']
tax_values = allowance_charge.get('attempt_tax_values')
multiplier_factor_numeric = allowance_charge['multiplier_factor_numeric']
amount = allowance_charge['amount']
reason = allowance_charge['reason']
charge_indicator = allowance_charge['charge_indicator']
if charge_indicator.lower() == 'true':
charge_indicator_sign = 1
else:
charge_indicator_sign = -1
if base_amount := allowance_charge.get('base_amount'):
price_unit = base_amount * charge_indicator_sign * file_document_sign
quantity = multiplier_factor_numeric / 100
else:
price_unit = amount * charge_indicator_sign * file_document_sign
quantity = 1
base_line_kwargs = {
**self._import_ubl_invoice_get_default_base_line_kwargs(collected_values),
'quantity': quantity,
'price_unit': price_unit,
'tax_ids': tax_values.get('tax'),
}
base_line_kwargs['_create_values']['name'] = reason
return base_line_kwargs
def _import_ubl_retrieve_products_search_plan(self, collected_values):
ProductProduct = self.env['product.product']
search_plan = [method[1] for method in sorted(ProductProduct._get_retrieval_product_search_plan())]
search_plan.append(ProductProduct._import_retrieve_product_from_invoice_predictive)
return search_plan
def _import_ubl_invoice_retrieve_products(self, collected_values):
company = collected_values['company']
lines_collected_values = collected_values['lines_collected_values']
product_values_list = [
line_collected_values['product_values']
for line_collected_values in lines_collected_values
]
self.env['product.product']._import_retrieve_product(
search_plan=self._import_ubl_retrieve_products_search_plan(collected_values),
company=company,
product_values_list=product_values_list,
)
for line_collected_values in lines_collected_values:
to_write = line_collected_values['to_write']
if product := line_collected_values['product_values'].get('product'):
to_write['product_id'] = product.id
else:
to_write['product_id'] = False
def _import_ubl_invoice_retrieve_product_uoms(self, collected_values):
lines_collected_values = collected_values['lines_collected_values']
logs = collected_values['logs']
cache = {}
for line_collected_values in lines_collected_values:
product_uom_values = line_collected_values['product_uom_values']
uom_code = product_uom_values.get('uom_code')
to_write = line_collected_values['to_write']
to_write['product_uom_id'] = False
if uom_code:
matched_uom_xmlid = {v: k for k, v in UOM_TO_UNECE_CODE.items()}.get(uom_code)
if matched_uom_xmlid:
if matched_uom_xmlid in cache:
uom = cache[matched_uom_xmlid]
else:
uom = cache[matched_uom_xmlid] = self.env.ref(matched_uom_xmlid, raise_if_not_found=False)
if uom:
product = line_collected_values['product_values'].get('product')
product_uom = product.product_tmpl_id.uom_id if product else self.env['uom.uom']
if product and not uom._has_common_reference(product_uom):
logs.append(_(
"The Unit of Measure '%(uom)s' (from unit code '%(code)s') was "
"ignored on the line for product '%(product)s' because it is not "
"compatible with the product's Unit of Measure '%(product_uom)s'. "
"The UoM was left empty.",
uom=uom.name,
code=uom_code,
product=product.display_name,
product_uom=product_uom.name,
))
product_uom_values['force_empty'] = True
continue
to_write['product_uom_id'] = uom.id
product_uom_values['uom'] = uom
def _import_ubl_invoice_retrieve_accounts(self, collected_values):
if not self.module_installed('account_accountant'):
# _predict_specific_account is defined in account_accountant
return
accounts_map = {}
lines_collected_values = collected_values['lines_collected_values']
for line_collected_values in lines_collected_values:
account_values = line_collected_values['account_values']
if predictive := account_values.get('invoice_predictive'):
account_params = {'move': predictive['invoice'], 'name': predictive['name'], 'partner': predictive['partner']}
account_key = tuple(account_params.values())
if account_key not in accounts_map:
accounts_map[account_key] = self.env['account.move.line']._predict_specific_account(**account_params)
account_id = accounts_map.get(account_key)
if account_id:
account_values['account'] = self.env['account.account'].browse(account_id)
def _import_ubl_invoice_retrieve_vehicles(self, collected_values):
company = collected_values['company']
cache = {}
for line_collected_values in collected_values['lines_collected_values']:
vehicle_values = line_collected_values.get('vehicle_values', {})
if not vehicle_values:
continue
if vehicle_values in cache:
vehicles = cache[vehicle_values]
else:
vehicles = self.env['fleet.vehicle'].sudo().search([ # noqa: OLS03001
('company_id', '=', company.id),
] + Domain.OR([
[(field, 'in', vals)] for field, vals in vehicle_values
]), limit=2)
cache[vehicle_values] = vehicles
if len(vehicles) == 1:
line_collected_values['to_write']['vehicle_id'] = vehicles.id
def _import_ubl_invoice_add_base_lines(self, collected_values):
AccountTax = self.env['account.tax']
base_lines = collected_values['base_lines'] = []
company = collected_values['company']
lines_collected_values = collected_values['lines_collected_values']
# Allowances / charges lines at document level.
for allowance_charge in collected_values['charges'] + collected_values['allowances']:
base_line_kwargs = self._import_ubl_invoice_get_allowance_charge_line_kwargs({
**collected_values,
'allowance_charge': allowance_charge,
})
base_lines.append(AccountTax._prepare_base_line_for_taxes_computation(
record=None,
**base_line_kwargs,
))
for line_collected_values in lines_collected_values:
to_write = line_collected_values['to_write']
# Extract charges matched with a fixed tax.
for charge in line_collected_values['charges']:
attempt_tax_values = charge.get('attempt_tax_values')
if not attempt_tax_values or not attempt_tax_values.get('tax'):
continue
# Suppose price_unit = 19, quantity = 10, discount = 10%
# for a total of 190 (before discount) and 171 (after discount).
# A charge of 25 is already accounted in 190 and we retrieve a fixed tax of 50 / 10 = 5.
# We need now to extract 25 from 190 as:
# price_subtotal_before = 171
# price_subtotal_after = 171 - 50 = 121
# price_unit = 19 - 5 = 14
# new_price_subtotal_before_discount = 140
# discount = (1 - (121 / 140)) * 100 = 13.5714286%
# That way, 14 * 10 * (1 - 0.135714286) = 121.
# The fix tax is giving an amount of 50.
# 121 + 50 = the original 171 we had at the beginning!
price_subtotal_before = to_write['price_unit'] * to_write['quantity'] * (1.0 - to_write['discount'] / 100.0)
price_subtotal_after = price_subtotal_before - charge['amount']
to_write['price_unit'] -= charge['amount'] / to_write['quantity']
new_price_subtotal_before_discount = to_write['price_unit'] * to_write['quantity']
to_write['discount'] = (1 - (price_subtotal_after / new_price_subtotal_before_discount)) * 100.0
# Product line.
base_line_kwargs = self._import_ubl_invoice_line_get_product_base_line_kwargs(line_collected_values)
base_lines.append(AccountTax._prepare_base_line_for_taxes_computation(
record=None,
**base_line_kwargs,
))
AccountTax._add_tax_details_in_base_lines(base_lines, company)
AccountTax._round_base_lines_tax_details(base_lines, company)
# Fix 'price_unit' if some price-included taxes are involved.
for base_line in base_lines:
for tax_data in base_line['tax_details']['taxes_data']:
if tax_data['tax'].price_include:
discount = base_line['discount'] / 100
raw_tax_amount_currency = tax_data['raw_tax_amount_currency'] / (1 - discount)
base_line['price_unit'] += raw_tax_amount_currency / (base_line['quantity'] if base_line['quantity'] else 1)
# Remove lines having a zero amount except 100% discounts
collected_values['base_lines'] = [
base_line
for base_line in collected_values['base_lines']
if (not base_line['currency_id'].is_zero(base_line['tax_details']['total_included_currency']) or base_line.get('discount'))
]
def _import_ubl_invoice_optional_fields(self, collected_values):
invoice = collected_values.get('invoice')
tree = collected_values.get('tree')
if not invoice.is_purchase_document():
return
invoice_fields = {}
if invoice.move_type == 'in_invoice':
invoice_fields = PEPPOL_INVOICE_OPTIONAL_FIELDS
elif invoice.move_type == 'in_refund':
invoice_fields = PEPPOL_CREDIT_NOTE_OPTIONAL_FIELDS
AccountMove = self.env['account.move']
for field_name, config in invoice_fields.items():
node_value = self._find_value(
'.//' + '/'.join(config.get('path')),
tree,
)
field = AccountMove._fields.get(field_name)
if not field or field.type not in config['supported_types'] or not node_value:
continue
collected_values['to_write'][field_name] = node_value
def _import_ubl_invoice_write_collected_values(self, collected_values):
invoice = collected_values['invoice']
base_lines = collected_values['base_lines']
to_write = collected_values['to_write']
invoice_line_ids = to_write['invoice_line_ids'] = []
for base_line in base_lines:
create_values = {
**base_line['_create_values'],
'quantity': base_line['quantity'],
'price_unit': base_line['price_unit'],
'discount': base_line['discount'],
'tax_ids': [Command.set(base_line['tax_ids'].ids)]
}
# If the values were not initialized, we don't want to prevent compute by explicitly putting them empty
if base_line['product_id']:
create_values['product_id'] = base_line['product_id'].id
if base_line['product_uom_id']:
create_values['product_uom_id'] = base_line['product_uom_id'].id
if base_line['account_id']:
create_values['account_id'] = base_line['account_id'].id
invoice_line_ids.append(Command.create(create_values))
container = {'records': invoice}
with (
invoice._check_balanced(container),
invoice._disable_discount_precision(),
invoice._sync_dynamic_lines(container),
):
invoice.write(to_write)
def _import_ubl_invoice_fix_taxes_amounts(self, collected_values):
AccountTax = self.env['account.tax']
invoice = collected_values['invoice']
tax_total_values = collected_values['tax_total_values']
tolerance = 0.03
total_tax_amount = sum(x['tax_amount_currency'] for x in tax_total_values.values())
currency = collected_values['currency_values']['currency']
tax_to_taxes = {}
taxes_to_tax_amount_currency = {}
is_complete = True
for tax_key, global_tax_values in tax_total_values.items():
taxes = self.env['account.tax']
for related_tax_values in global_tax_values['related_taxes_values']:
tax = related_tax_values.get('tax')
if tax:
taxes |= tax
else:
is_complete = False
break
for tax in taxes:
tax_to_taxes[tax] = taxes
taxes_to_tax_amount_currency[taxes] = global_tax_values['tax_amount_currency']
# If we are too far away from the total retrieved in the xml, don't fix anything: the error is elsewhere.
collected_values['are_taxes_complete'] = is_complete
if (
not is_complete
or currency.compare_amounts(abs(invoice.amount_tax - total_tax_amount) - tolerance, 0.0) > 0
):
return
# Fix the base lines.
def grouping_function(_base_line, tax_data):
return tax_data and tax_to_taxes.get(tax_data['tax'])
base_lines, tax_lines = invoice._get_rounded_base_and_tax_lines()
base_lines_aggregated_values = AccountTax._aggregate_base_lines_tax_details(base_lines, grouping_function)
values_per_grouping_key = AccountTax._aggregate_base_lines_aggregated_values(base_lines_aggregated_values)
for taxes, values in values_per_grouping_key.items():
if not taxes:
continue
target_tax_amount_currency = taxes_to_tax_amount_currency[taxes]
target_factors = [
{
'factor': tax_data['raw_tax_amount_currency'],
'tax_data': tax_data,
}
for _base_line, taxes_data in values['base_line_x_taxes_data']
for tax_data in taxes_data
]
amounts_to_distribute = AccountTax._distribute_delta_amount_smoothly(
precision_digits=currency.decimal_places,
delta_amount=target_tax_amount_currency,
target_factors=target_factors,
)
for target_factor, amount_to_distribute in zip(target_factors, amounts_to_distribute):
tax_data = target_factor['tax_data']
tax_data['tax_amount_currency'] = amount_to_distribute
AccountTax._add_accounting_data_in_base_lines_tax_details(base_lines, invoice.company_id, include_caba_tags=invoice.always_tax_exigible)
tax_results = AccountTax._prepare_tax_lines(base_lines, invoice.company_id, tax_lines=tax_lines)
line_ids_commands = []
for tax_line_vals, grouping_key, to_update in tax_results['tax_lines_to_update']:
line_ids_commands.append(Command.update(tax_line_vals['record'].id, {
'amount_currency': to_update['amount_currency'],
'balance': to_update['balance'],
}))
container = {'records': invoice}
with (
invoice._check_balanced(container),
invoice._disable_discount_precision(),
invoice._sync_dynamic_lines(container),
):
invoice.line_ids = line_ids_commands
def _import_ubl_invoice_fix_untaxed_amount(self, collected_values):
if not collected_values['are_taxes_complete']:
return
tree = collected_values['tree']
file_document_sign = collected_values['file_document_sign']
currency = collected_values['currency_values']['currency']
tax_exclusive_amount_str = tree.findtext('./{*}LegalMonetaryTotal/{*}TaxExclusiveAmount')
if not tax_exclusive_amount_str:
return
payable_rounding_amount_str = tree.findtext('./{*}LegalMonetaryTotal/{*}PayableRoundingAmount')
tax_exclusive_amount = file_document_sign * float(tax_exclusive_amount_str or 0.0)
payable_rounding_amount = file_document_sign * float(payable_rounding_amount_str or 0.0)
expected_untaxed_amount = tax_exclusive_amount + payable_rounding_amount
invoice = collected_values['invoice']
difference = currency.round(expected_untaxed_amount - invoice.amount_untaxed)
for line_collected_values in collected_values['lines_collected_values']:
for charge in line_collected_values['charges']:
attempt_tax_values = charge.get('attempt_tax_values')
if attempt_tax_values and attempt_tax_values.get('tax'):
difference -= charge['amount']
if currency.is_zero(difference):
return
container = {'records': invoice}
with (
invoice._check_balanced(container),
invoice._disable_discount_precision(),
invoice._sync_dynamic_lines(container),
):
invoice.invoice_line_ids = [
Command.create({
'display_type': 'product',
'name': _("Rounding"),
'quantity': 1,
'price_unit': difference,
'tax_ids': [],
}),
]
def _import_attachments(self, invoice, tree):
""" EXTENDS 'account_edi_common': ATTEMPTS to create a PDF attachment when the XML file doesn't provide one."""
IrConfigParam = self.env['ir.config_parameter'].sudo()
disable_pdf_in_xml = IrConfigParam.get_bool("account_edi_ubl_cii.disable_pdf_in_xml")
additional_docs = super()._import_attachments(invoice, tree)
if (
additional_docs or
invoice.message_main_attachment_id or
not invoice.is_purchase_document() or
disable_pdf_in_xml
):
return additional_docs
try:
invoices_by_odoo_xmlid = 'account_edi_ubl_cii.action_report_account_invoices_generated_by_odoo'
if not self.env.ref(invoices_by_odoo_xmlid, raise_if_not_found=False):
_logger.warning("Missing template while generating substitute PDF attachment for invoice %s", invoice.id)
return additional_docs
report_xmlid = invoices_by_odoo_xmlid
pdf_raw, pdf_extension = self.env['ir.actions.report'] \
._render_qweb_pdf(report_xmlid, res_ids=[invoice.id])
if pdf_extension != 'pdf':
return additional_docs
if not self.env.ref('account_edi_ubl_cii.layout_invoices_generated_by_odoo', raise_if_not_found=False):
# TODO: Remove in master
# add a watermark to the generated pdf
with io.BytesIO(pdf_raw) as pdf_stream:
new_pdf_stream = pdf.add_banner(pdf_stream, _('Generated by Odoo'), logo=False)
pdf_raw = new_pdf_stream.getvalue()
new_pdf_stream.close()
invoice_name = invoice.display_name.replace(_('Draft'), '')
pdf_filename = _('%(invoice_name)s - Generated by Odoo', invoice_name=invoice_name)
attachment = self.env['ir.attachment'].create({
'name': pdf_filename + '.pdf',
'res_id': invoice.id,
'res_model': 'account.move',
'raw': pdf_raw,
'type': 'binary',
'mimetype': 'application/pdf',
})
invoice._message_set_main_attachment_id(attachment, force=True, filter_xml=False)
return attachment
except Exception: # noqa: BLE001
_logger.exception("Error while generating substitute PDF attachment for invoice %s", invoice.id)
return additional_docs
def _import_ubl_invoice_post_processing(self, collected_values):
# During the import, fill 'ubl_cii_xml_file' to be retrieved later if necessary.
invoice = collected_values['invoice']
if collected_values['file_data']['attachment'] and invoice.is_purchase_document(include_receipts=True):
collected_values['file_data']['attachment'].write({
'res_field': 'ubl_cii_xml_file',
'res_model': invoice._name,
'res_id': invoice.id,
})
# Collect the embedded documents.
invoice = collected_values['invoice']
source_attachment = collected_values['file_data']['attachment'] or self.env['ir.attachment']
attachments = source_attachment + self._import_attachments(invoice, collected_values['tree'])
# Chatter.
body = Markup("%s") % _(
"Format used to import the invoice: %s",
self.env['ir.model']._get(self._name).name,
)
if logs := dict.fromkeys(collected_values['logs']):
body += Markup("") % Markup().join(Markup("%s") % l for l in logs)
invoice.with_context(no_new_invoice=True).message_post(body=body, attachment_ids=attachments.ids)
def _ubl_import_invoice(self, invoice, file_data, new=False):
tree = file_data['xml_tree']
company = invoice.company_id
collected_values = {
'invoice': invoice,
'company': company,
'odoo_document_type': 'sale' if invoice.journal_id.type == 'sale' else 'purchase',
'tree': tree,
'file_data': file_data,
'logs': [],
'to_write': {},
}
self._import_ubl_invoice_document_sign(collected_values)
self._import_ubl_invoice_update_move_type(collected_values)
# Bank values are required for retrieving the partner via their bank account number
self._import_ubl_invoice_add_partner_bank_values(collected_values)
# partner_id.
self._import_ubl_invoice_add_customer_values(collected_values)
self._import_ubl_retrieve_customer(collected_values)
self._import_ubl_create_missing_customer(collected_values)
# invoice_date / invoice_date_due
self._import_ubl_invoice_add_issue_date(collected_values)
self._import_ubl_invoice_add_due_date(collected_values)
# currency_id.
self._import_ubl_invoice_add_currency_values(collected_values)
self._import_ubl_invoice_add_currency(collected_values)
# partner_bank_id
self._import_ubl_retrieve_partner_bank(collected_values)
# ref / invoice_origin / narration / payment_reference / delivery_date
self._import_ubl_invoice_add_ref(collected_values)
self._import_ubl_invoice_add_invoice_origin(collected_values)
self._import_ubl_invoice_add_narration(collected_values)
self._import_ubl_invoice_add_payment_reference(collected_values)
self._import_ubl_invoice_add_delivery(collected_values)
# optional fields
self._import_ubl_invoice_optional_fields(collected_values)
# invoice_incoterm_id
self._import_ubl_invoice_add_incoterm_values(collected_values)
self._import_ubl_invoice_add_incoterm(collected_values)
# Prepaid / rounding amounts / Tax total values.
self._import_ubl_invoice_add_prepaid_amount(collected_values)
self._import_ubl_invoice_add_tax_total_values(collected_values)
# Extract information about allowance / charges.
self._import_ubl_invoice_add_allowances_charges_values(collected_values)
# Invoice lines values.
self._import_ubl_invoice_add_invoice_line_values(collected_values)
self._import_ubl_invoice_retrieve_products(collected_values)
self._import_ubl_invoice_retrieve_product_uoms(collected_values)
self._import_ubl_invoice_retrieve_accounts(collected_values)
self._import_ubl_invoice_retrieve_taxes(collected_values)
self._import_ubl_invoice_retrieve_vehicles(collected_values)
self._import_ubl_invoice_add_base_lines(collected_values)
# End the invoice.
self._import_ubl_invoice_write_collected_values(collected_values)
self._import_ubl_invoice_fix_taxes_amounts(collected_values)
self._import_ubl_invoice_fix_untaxed_amount(collected_values)
self._import_ubl_invoice_post_processing(collected_values)
return