import base64 import re from collections import defaultdict from copy import deepcopy from hashlib import sha1 from lxml import etree from markupsafe import Markup from odoo import Command, _, api, fields, models from odoo.exceptions import UserError from odoo.tools import float_round, float_repr, float_compare, date_utils, SQL from odoo.tools.xml_utils import cleanup_xml_node, find_xml_value from odoo.tools.sql import column_exists, create_column from odoo.addons.l10n_es_edi_facturae.xml_utils import ( NS_MAP, _canonicalize_node, _reference_digests, ) PHONE_CLEAN_TABLE = str.maketrans({" ": None, "-": None, "(": None, ")": None, "+": None}) COUNTRY_CODE_MAP = { "BD": "BGD", "BE": "BEL", "BF": "BFA", "BG": "BGR", "BA": "BIH", "BB": "BRB", "WF": "WLF", "BL": "BLM", "BM": "BMU", "BN": "BRN", "BO": "BOL", "BH": "BHR", "BI": "BDI", "BJ": "BEN", "BT": "BTN", "JM": "JAM", "BV": "BVT", "BW": "BWA", "WS": "WSM", "BQ": "BES", "BR": "BRA", "BS": "BHS", "JE": "JEY", "BY": "BLR", "BZ": "BLZ", "RU": "RUS", "RW": "RWA", "RS": "SRB", "TL": "TLS", "RE": "REU", "TM": "TKM", "TJ": "TJK", "RO": "ROU", "TK": "TKL", "GW": "GNB", "GU": "GUM", "GT": "GTM", "GS": "SGS", "GR": "GRC", "GQ": "GNQ", "GP": "GLP", "JP": "JPN", "GY": "GUY", "GG": "GGY", "GF": "GUF", "GE": "GEO", "GD": "GRD", "GB": "GBR", "GA": "GAB", "SV": "SLV", "GN": "GIN", "GM": "GMB", "GL": "GRL", "GI": "GIB", "GH": "GHA", "OM": "OMN", "TN": "TUN", "JO": "JOR", "HR": "HRV", "HT": "HTI", "HU": "HUN", "HK": "HKG", "HN": "HND", "HM": "HMD", "VE": "VEN", "PR": "PRI", "PS": "PSE", "PW": "PLW", "PT": "PRT", "SJ": "SJM", "PY": "PRY", "IQ": "IRQ", "PA": "PAN", "PF": "PYF", "PG": "PNG", "PE": "PER", "PK": "PAK", "PH": "PHL", "PN": "PCN", "PL": "POL", "PM": "SPM", "ZM": "ZMB", "EH": "ESH", "EE": "EST", "EG": "EGY", "ZA": "ZAF", "EC": "ECU", "IT": "ITA", "VN": "VNM", "SB": "SLB", "ET": "ETH", "SO": "SOM", "ZW": "ZWE", "SA": "SAU", "ES": "ESP", "ER": "ERI", "ME": "MNE", "MD": "MDA", "MG": "MDG", "MF": "MAF", "MA": "MAR", "MC": "MCO", "UZ": "UZB", "MM": "MMR", "ML": "MLI", "MO": "MAC", "MN": "MNG", "MH": "MHL", "MK": "MKD", "MU": "MUS", "MT": "MLT", "MW": "MWI", "MV": "MDV", "MQ": "MTQ", "MP": "MNP", "MS": "MSR", "MR": "MRT", "IM": "IMN", "UG": "UGA", "TZ": "TZA", "MY": "MYS", "MX": "MEX", "IL": "ISR", "FR": "FRA", "IO": "IOT", "SH": "SHN", "FI": "FIN", "FJ": "FJI", "FK": "FLK", "FM": "FSM", "FO": "FRO", "NI": "NIC", "NL": "NLD", "NO": "NOR", "NA": "NAM", "VU": "VUT", "NC": "NCL", "NE": "NER", "NF": "NFK", "NG": "NGA", "NZ": "NZL", "NP": "NPL", "NR": "NRU", "NU": "NIU", "CK": "COK", "XK": "XKX", "CI": "CIV", "CH": "CHE", "CO": "COL", "CN": "CHN", "CM": "CMR", "CL": "CHL", "CC": "CCK", "CA": "CAN", "CG": "COG", "CF": "CAF", "CD": "COD", "CZ": "CZE", "CY": "CYP", "CX": "CXR", "CR": "CRI", "CW": "CUW", "CV": "CPV", "CU": "CUB", "SZ": "SWZ", "SY": "SYR", "SX": "SXM", "KG": "KGZ", "KE": "KEN", "SS": "SSD", "SR": "SUR", "KI": "KIR", "KH": "KHM", "KN": "KNA", "KM": "COM", "ST": "STP", "SK": "SVK", "KR": "KOR", "SI": "SVN", "KP": "PRK", "KW": "KWT", "SN": "SEN", "SM": "SMR", "SL": "SLE", "SC": "SYC", "KZ": "KAZ", "KY": "CYM", "SG": "SGP", "SE": "SWE", "SD": "SDN", "DO": "DOM", "DM": "DMA", "DJ": "DJI", "DK": "DNK", "VG": "VGB", "DE": "DEU", "YE": "YEM", "DZ": "DZA", "US": "USA", "UY": "URY", "YT": "MYT", "UM": "UMI", "LB": "LBN", "LC": "LCA", "LA": "LAO", "TV": "TUV", "TW": "TWN", "TT": "TTO", "TR": "TUR", "LK": "LKA", "LI": "LIE", "LV": "LVA", "TO": "TON", "LT": "LTU", "LU": "LUX", "LR": "LBR", "LS": "LSO", "TH": "THA", "TF": "ATF", "TG": "TGO", "TD": "TCD", "TC": "TCA", "LY": "LBY", "VA": "VAT", "VC": "VCT", "AE": "ARE", "AD": "AND", "AG": "ATG", "AF": "AFG", "AI": "AIA", "VI": "VIR", "IS": "ISL", "IR": "IRN", "AM": "ARM", "AL": "ALB", "AO": "AGO", "AQ": "ATA", "AS": "ASM", "AR": "ARG", "AU": "AUS", "AT": "AUT", "AW": "ABW", "IN": "IND", "AX": "ALA", "AZ": "AZE", "IE": "IRL", "ID": "IDN", "UA": "UKR", "QA": "QAT", "MZ": "MOZ" } REVERSED_COUNTRY_CODE = {v: k for k, v in COUNTRY_CODE_MAP.items()} # The reason type should be exactly the fields given below. # We cannot rely on the translated Selection since a single character difference would render the XML incorrect. SPANISH_CREDIT_REASON_TYPE = { '01': 'Número de la factura', '02': 'Serie de la factura', '03': 'Fecha expedición', '04': 'Nombre y apellidos/Razón Social-Emisor', '05': 'Nombre y apellidos/Razón Social-Receptor', '06': 'Identificación fiscal Emisor/obligado', '07': 'Identificación fiscal Receptor', '08': 'Domicilio Emisor/Obligado', '09': 'Domicilio Receptor', '10': 'Detalle Operación', '11': 'Porcentaje impositivo a aplicar', '12': 'Cuota tributaria a aplicar', '13': 'Fecha/Periodo a aplicar', '14': 'Clase de factura', '15': 'Literales legales', '16': 'Base imponible', '80': 'Cálculo de cuotas repercutidas', '81': 'Cálculo de cuotas retenidas', '82': 'Base imponible modificada por devolución de envases / embalajes', '83': 'Base imponible modificada por descuentos y bonificaciones', '84': 'Base imponible modificada por resolución firme, judicial o administrativa', '85': 'Base imponible modificada cuotas repercutidas no satisfechas. Auto de declaración de concurso', } class AccountMove(models.Model): _inherit = 'account.move' l10n_es_edi_facturae_xml_id = fields.Many2one( comodel_name='ir.attachment', string="Facturae Attachment", compute=lambda self: self._compute_linked_attachment_id('l10n_es_edi_facturae_xml_id', 'l10n_es_edi_facturae_xml_file'), depends=['l10n_es_edi_facturae_xml_file'] ) l10n_es_edi_facturae_xml_file = fields.Binary( attachment=True, string="Facturae File", copy=False, ) l10n_es_edi_facturae_reason_code = fields.Selection( selection=[ ('01', "Invoice number"), ('02', "Invoice serial number"), ('03', "Issue date"), ('04', "Name and surnames/Corporate name - Issuer (Sender)"), ('05', "Name and surnames/Corporate name - Receiver"), ('06', "Issuer's Tax Identification Number"), ('07', "Receiver's Tax Identification Number"), ('08', "Issuer's address"), ('09', "Receiver's address"), ('10', "Item line"), ('11', "Applicable Tax Rate"), ('12', "Applicable Tax Amount"), ('13', "Applicable Date/Period"), ('14', "Invoice Class"), ('15', "Legal literals"), ('16', "Taxable Base"), ('80', "Calculation of tax outputs"), ('81', "Calculation of tax inputs"), ('82', "Taxable Base modified due to return of packages and packaging materials"), ('83', "Taxable Base modified due to discounts and rebates"), ('84', "Taxable Base modified due to firm court ruling or administrative decision"), ('85', "Taxable Base modified due to unpaid outputs where there is a judgement opening insolvency proceedings"), ], string='Spanish Facturae EDI Reason Code', compute='_compute_l10n_es_edi_facturae_reason_code', store=True, readonly=False, ) l10n_es_invoicing_period_start_date = fields.Date(string="Invoice Period Start Date") l10n_es_invoicing_period_end_date = fields.Date(string="Invoice Period End Date") l10n_es_payment_means = fields.Selection( selection=[ ('01', "In cash"), ('02', "Direct debit"), ('03', "Receipt"), ('04', "Credit transfer"), ('05', "Accepted bill of exchange"), ('06', "Documentary credit"), ('07', "Contract award"), ('08', "Bill of exchange"), ('09', "Transferable promissory note"), ('10', "Non transferable promissory note"), ('11', "Cheque"), ('12', "Open account reimbursement"), ('13', "Special payment"), ('14', "Set-off by reciprocal credits"), ('15', "Payment by postgiro"), ('16', "Certified cheque"), ('17', "Banker’s draft"), ('18', "Cash on delivery"), ('19', "Payment by card"), ], string="Payment Means", compute='_compute_l10n_es_payment_means', store=True, readonly=False, ) def _auto_init(self): # Create compute stored field l10n_es_edi_facturae_reason_code and # l10n_es_payment_means here to avoid timeout error on large databases. if not column_exists(self.env.cr, 'account_move', 'l10n_es_edi_facturae_reason_code'): create_column(self.env.cr, 'account_move', 'l10n_es_edi_facturae_reason_code', 'varchar') if not column_exists(self.env.cr, 'account_move', 'l10n_es_payment_means'): create_column(self.env.cr, 'account_move', 'l10n_es_payment_means', 'varchar') return super()._auto_init() @api.depends('country_code') def _compute_l10n_es_edi_facturae_reason_code(self): for move in self.filtered(lambda move: move.country_code == 'ES'): move.l10n_es_edi_facturae_reason_code = move.l10n_es_edi_facturae_reason_code or '10' @api.depends('country_code') def _compute_l10n_es_payment_means(self): for move in self.filtered(lambda move: move.country_code == 'ES'): move.l10n_es_payment_means = move.l10n_es_payment_means or '04' def _get_fields_to_detach(self): # EXTENDS account fields_list = super()._get_fields_to_detach() fields_list.append("l10n_es_edi_facturae_xml_file") return fields_list def _l10n_es_edi_facturae_export_data_check(self): """ This function checks the Settings, Company, Partners involved in the sending activity and returns an errors dictionary ready for the actionable_errors widget to display. """ return { **self.mapped("company_id")._l10n_es_edi_facturae_export_check(), **self.mapped("partner_id")._l10n_es_edi_facturae_export_check(), } def _l10n_es_edi_facturae_get_default_enable(self): self.ensure_one() return not self.invoice_pdf_report_id \ and not self.l10n_es_edi_facturae_xml_id \ and not self.l10n_es_is_simplified \ and self.is_invoice(include_receipts=True) \ and self.country_code == 'ES' \ and self.company_id.sudo().l10n_es_edi_facturae_certificate_ids # We only enable Facturae if a certificate is valid or has been valid (which will raise an error) def _l10n_es_edi_facturae_get_filename(self): self.ensure_one() return f'{self.name.replace("/", "_")}_facturae_signed.xml' def _l10n_es_edi_facturae_get_tax_period(self): self.ensure_one() if self.env['res.company'].fields_get(['account_return_periodicity']): period_start, period_end = self.env.ref('l10n_es_reports.es_mod303_tax_return_type')._get_period_boundaries(self.company_id, self.date) else: period_start = date_utils.start_of(self.date, 'month') period_end = date_utils.end_of(self.date, 'month') return {'start':period_start, 'end':period_end} def _l10n_es_edi_facturae_get_refunded_invoices(self): self.env['account.partial.reconcile'].flush_model() invoices_refunded_mapping = {invoice.id: invoice.reversed_entry_id.id for invoice in self} stored_ids = tuple(self.ids) queries = [] for source_field, counterpart_field in ( ('debit_move_id', 'credit_move_id'), ('credit_move_id', 'debit_move_id'), ): queries.append(SQL(''' SELECT source_line.move_id AS source_move_id, counterpart_line.move_id AS counterpart_move_id FROM account_partial_reconcile part JOIN account_move_line source_line ON source_line.id = part.%s JOIN account_move_line counterpart_line ON counterpart_line.id = part.%s WHERE source_line.move_id IN %s AND counterpart_line.move_id != source_line.move_id GROUP BY source_move_id, counterpart_move_id ''', SQL.identifier(source_field), SQL.identifier(counterpart_field), stored_ids)) payment_data = defaultdict(list) for row in self.env.execute_query_dict(SQL(" UNION ALL ").join(queries)): payment_data[row['source_move_id']].append(row) for invoice in self: if not invoice.move_type.endswith('refund'): # We only want to map refunds continue for move_id in (record_dict['counterpart_move_id'] for record_dict in payment_data.get(invoice.id, [])): invoices_refunded_mapping[invoice.id] = move_id return invoices_refunded_mapping def _l10n_es_edi_facturae_get_corrective_data(self): self.ensure_one() if self.move_type.endswith('refund'): if not self.reversed_entry_id: raise UserError(_("The credit note/refund appears to have been issued manually. For the purpose of " "generating a Facturae document, it's necessary that the credit note/refund is created " "directly from the associated invoice/bill.")) refunded_invoice = self.env['account.move'].browse(self._l10n_es_edi_facturae_get_refunded_invoices()[self.id]) tax_period = refunded_invoice._l10n_es_edi_facturae_get_tax_period() reason_code = self.l10n_es_edi_facturae_reason_code or '10' reason_description = SPANISH_CREDIT_REASON_TYPE[reason_code] return { 'refunded_invoice_record': refunded_invoice, 'ReasonCode': reason_code, 'Reason': reason_description, 'TaxPeriod': { 'StartDate': tax_period.get('start'), 'EndDate': tax_period.get('end'), } } return {} def _l10n_es_edi_facturae_get_administrative_centers(self, partner): self.ensure_one() administrative_centers = [] for ac in partner.child_ids.filtered(lambda p: p.type == 'facturae_ac'): ac_template = { 'center_code': ac.l10n_es_edi_facturae_ac_center_code, 'name': ac.name, 'partner': ac, 'partner_country_code': COUNTRY_CODE_MAP[ac.country_code], 'partner_phone': ac.phone.translate(PHONE_CLEAN_TABLE) if ac.phone else False, 'physical_gln': ac.l10n_es_edi_facturae_ac_physical_gln, 'logical_operational_point': ac.l10n_es_edi_facturae_ac_logical_operational_point, } # An administrative center can have multiple roles, each of which should be reported separately. for role in ac.l10n_es_edi_facturae_ac_role_type_ids or [self.env['l10n_es_edi_facturae.ac_role_type']]: administrative_centers.append({ **ac_template, 'role_type_code': role.code, }) return administrative_centers def _l10n_es_edi_facturae_get_tax_node_from_tax_data(self, values, round=False): self.ensure_one() tax = values['grouping_key'] prefix = '' if round else 'raw_' tax_sign = -1 if tax.amount < 0.0 else 1 return { 'tax_record': tax, 'TaxRate': f'{abs(tax.amount):.3f}', 'TaxableBase': { 'TotalAmount': self.currency_id.round(values[f'{prefix}base_amount_currency']), 'EquivalentInEuros': self.company_currency_id.round(values[f'{prefix}base_amount']), }, 'TaxAmount': { 'TotalAmount': self.currency_id.round(tax_sign * values[f'{prefix}tax_amount_currency']), 'EquivalentInEuros': self.company_currency_id.round(tax_sign * values[f'{prefix}tax_amount']), }, } def _l10n_es_edi_facturae_convert_payment_terms_to_installments(self): """ Convert the payments terms to a list of elements to be used in the node of the Facturae XML generation. """ self.ensure_one() installments = [] if self.is_inbound() and self.partner_bank_id: for payment_term in self.line_ids.filtered(lambda l: l.display_type == 'payment_term').sorted('date_maturity'): installments.append({ 'InstallmentDueDate': payment_term.date_maturity, 'InstallmentAmount': payment_term.amount_residual_currency, 'PaymentMeans': self.l10n_es_payment_means or '04', 'AccountToBeCredited': { 'IBAN': self.partner_bank_id.sanitized_acc_number, 'BIC': self.partner_bank_id.bank_bic, }, }) return installments def _l10n_es_edi_facturae_prepare_inv_line(self, base_line, aggregated_values): """ Convert the invoice lines to a list of items required for the Facturae xml generation :return: A tuple containing the Face items, the taxes and the invoice totals data. """ self.ensure_one() invoice_ref = self.ref and self.ref[:20] line = base_line['record'] tax_details = base_line['tax_details'] receiver_transaction_reference = ( line.sale_line_ids.order_id.client_order_ref[:20] if 'sale_line_ids' in line._fields and line.sale_line_ids.order_id.client_order_ref else invoice_ref ) xml_values = { 'ReceiverTransactionReference': receiver_transaction_reference, 'FileReference': invoice_ref, 'ReceiverContractReference': invoice_ref, 'FileDate': fields.Date.context_today(self), 'ItemDescription': line.name, 'Quantity': line.quantity, 'UnitOfMeasure': line.product_uom_id.l10n_es_edi_facturae_uom_code, 'DiscountsAndRebates': [], 'Charges': [], 'GrossAmount': float_round(tax_details['raw_total_excluded_currency'], precision_digits=8), } if line.discount == 100.0: raw_total_cost = line.price_unit * line.quantity else: raw_total_cost = tax_details['raw_total_excluded_currency'] / (1 - (line.discount / 100.0)) xml_values['TotalCost'] = float_round(raw_total_cost, precision_digits=8) if line.quantity: xml_values['UnitPriceWithoutTax'] = float_round(raw_total_cost / line.quantity, precision_digits=8) else: xml_values['UnitPriceWithoutTax'] = 0.0 discount_amount = xml_values['TotalCost'] - xml_values['GrossAmount'] if float_compare(discount_amount, 0.0, precision_digits=8) > 0: xml_values['DiscountsAndRebates'].append({ 'DiscountReason': '/', 'DiscountRate': f'{line.discount:.2f}', 'DiscountAmount': discount_amount, }) if float_compare(discount_amount, 0.0, precision_digits=8) < 0: xml_values['Charges'].append({ 'ChargeReason': '/', 'ChargeRate': f'{-line.discount:.2f}', 'ChargeAmount': -discount_amount, }) xml_values['TaxesOutputs'] = [ self._l10n_es_edi_facturae_get_tax_node_from_tax_data(values) for values in aggregated_values.values() if values['grouping_key'] and values['grouping_key'].amount >= 0.0 ] xml_values['TaxesWithheld'] = [ self._l10n_es_edi_facturae_get_tax_node_from_tax_data(values) for values in aggregated_values.values() if values['grouping_key'] and values['grouping_key'].amount < 0.0 ] return xml_values def _l10n_es_edi_facturae_export_facturae(self): """ Produce the Facturae XML data for the invoice. :return: (data needed to render the full template, data needed to render the signature template) """ def extract_party_name(party): name = {'firstname': 'UNKNOWN', 'surname': 'UNKNOWN', 'surname2': ''} if not party.is_company: name_split = [part for part in party.name.replace(', ', ' ').split(' ') if part] if len(name_split) > 2: name['firstname'] = ' '.join(name_split[:-2]) name['surname'], name['surname2'] = name_split[-2:] elif len(name_split) == 2: name['firstname'] = ' '.join(name_split[:-1]) name['surname'] = name_split[-1] return name self.ensure_one() company = self.company_id partner = self.commercial_partner_id if not company.vat: raise UserError(_('The company needs a set tax identification number or VAT number')) if not partner.vat: raise UserError(_('The partner needs a set tax identification number or VAT number')) if not partner.country_id: raise UserError(_("The partner needs a set country")) if self.move_type == "entry": return False operation_date = None if self.delivery_date and self.delivery_date != self.invoice_date: operation_date = self.delivery_date.isoformat() # Multi-currencies. eur_curr = self.env['res.currency'].search([('name', '=', 'EUR')]) inv_curr = self.currency_id conversion_needed = inv_curr != eur_curr # Invoice xml values. invoice_ref = self.ref and self.ref[:20] legal_literals = self.narration and self.narration.striptags() legal_literals = legal_literals.split(";") if legal_literals else False invoice_values = { 'invoice_record': self, 'invoice_currency': inv_curr, 'InvoiceDocumentType': 'FA' if self.l10n_es_is_simplified else 'FC', 'InvoiceClass': 'OR' if self.move_type in ['out_refund', 'in_refund'] else 'OO', 'Corrective': self._l10n_es_edi_facturae_get_corrective_data(), 'InvoiceIssueData': { 'OperationDate': operation_date, 'ExchangeRateDetails': conversion_needed, 'ExchangeRate': f"{round(self.invoice_currency_rate, 4):.4f}", 'LanguageName': self.env.context.get('lang', 'en_US').split('_')[0], 'InvoicingPeriod': None, 'ReceiverTransactionReference': invoice_ref, 'FileReference': invoice_ref, 'ReceiverContractReference': invoice_ref, }, 'TaxOutputs': [], 'TaxesWithheld': [], 'TotalGrossAmount': 0.0, 'TotalGeneralDiscounts': 0.0, 'TotalGeneralSurcharges': 0.0, 'TotalGrossAmountBeforeTaxes': 0.0, 'TotalTaxOutputs': 0.0, 'TotalTaxesWithheld': 0.0, 'PaymentsOnAccount': [], 'TotalOutstandingAmount': abs(self.amount_total_in_currency_signed), 'InvoiceTotal': abs(self.amount_total_in_currency_signed), 'TotalPaymentsOnAccount': 0.0, 'AmountsWithheld': None, 'TotalExecutableAmount': abs(self.amount_total_in_currency_signed), 'Items': [], 'PaymentDetails': self._l10n_es_edi_facturae_convert_payment_terms_to_installments(), 'LegalLiterals': legal_literals, } # Taxes. AccountTax = self.env['account.tax'] base_lines, _tax_lines = self._get_rounded_base_and_tax_lines() def grouping_function_per_tax(base_line, tax_data): return tax_data['tax'] if tax_data else None base_lines_aggregated_values = AccountTax._aggregate_base_lines_tax_details(base_lines, grouping_function_per_tax) for base_line, aggregated_values in base_lines_aggregated_values: invoice_line_values = self._l10n_es_edi_facturae_prepare_inv_line(base_line, aggregated_values) invoice_values['TotalGrossAmount'] += invoice_line_values['GrossAmount'] invoice_values['Items'].append(invoice_line_values) def grouping_function_per_base_line_tax(base_line, tax_data): if not tax_data: return return { 'tax_es_type': tax_data['tax'].l10n_es_edi_facturae_tax_type, 'tax_rate': tax_data['tax'].amount, 'tax_amount_type': tax_data['tax'].amount_type, } base_lines_aggregated_values = AccountTax._aggregate_base_lines_tax_details(base_lines, grouping_function_per_base_line_tax) 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_record = values['base_line_x_taxes_data'][0][1][0]['tax'] if not tax_record: continue is_withholding = values['grouping_key']['tax_rate'] < 0.0 tax_data = self._l10n_es_edi_facturae_get_tax_node_from_tax_data({**values, 'grouping_key': tax_record}, round=True) if is_withholding: invoice_values['TaxesWithheld'].append(tax_data) invoice_values['TotalTaxesWithheld'] -= values['tax_amount_currency'] else: invoice_values['TaxOutputs'].append(tax_data) invoice_values['TotalTaxOutputs'] += values['tax_amount_currency'] invoice_values['TotalTaxesWithheld'] = abs(invoice_values['TotalTaxesWithheld']) invoice_values['TotalGrossAmountBeforeTaxes'] = ( invoice_values['TotalGrossAmount'] - invoice_values['TotalGeneralDiscounts'] + invoice_values['TotalGeneralSurcharges'] ) refund_multiplier = -1 if self.move_type in ('out_refund', 'in_refund') else 1 template_values = { 'self_party': company.partner_id, 'self_party_country_code': COUNTRY_CODE_MAP[company.country_id.code], 'self_party_name': extract_party_name(company.partner_id), 'self_party_administrative_centers': self._l10n_es_edi_facturae_get_administrative_centers(company.partner_id), 'other_party': partner, 'other_party_country_code': COUNTRY_CODE_MAP[partner.country_id.code], 'other_party_phone': partner.phone.translate(PHONE_CLEAN_TABLE) if partner.phone else False, 'other_party_name': extract_party_name(partner), 'other_party_administrative_centers': self._l10n_es_edi_facturae_get_administrative_centers(partner), 'is_outstanding': self.move_type.startswith('out_'), 'float_repr': float_repr, 'file_currency': inv_curr, 'eur': eur_curr, 'conversion_needed': conversion_needed, 'refund_multiplier': refund_multiplier, 'Modality': 'I', 'BatchIdentifier': self.name, 'InvoicesCount': 1, 'TotalInvoicesAmount': { 'TotalAmount': abs(self.amount_total_in_currency_signed), 'EquivalentInEuros': abs(self.amount_total_signed), }, 'TotalOutstandingAmount': { 'TotalAmount': abs(self.amount_total_in_currency_signed), 'EquivalentInEuros': abs(self.amount_total_signed), }, 'TotalExecutableAmount': { 'TotalAmount': abs(self.amount_total_in_currency_signed), 'EquivalentInEuros': abs(self.amount_total_signed), }, 'InvoiceCurrencyCode': inv_curr.name, 'Invoices': [invoice_values], } if self.l10n_es_invoicing_period_start_date and self.l10n_es_invoicing_period_end_date: template_values['Invoices'][0]['InvoiceIssueData']['InvoicingPeriod'] = { 'StartDate': self.l10n_es_invoicing_period_start_date, 'EndDate': self.l10n_es_invoicing_period_end_date, } invoice_issuer_signature_type = 'supplier' if self.move_type == 'out_invoice' else 'customer' signature_values = {'SigningTime': '', 'SignerRole': invoice_issuer_signature_type} return template_values, signature_values def _l10n_es_edi_facturae_render_facturae(self): """ Produce the Facturae XML file for the invoice. :return: rendered xml file string. :rtype: str """ self.ensure_one() company = self.company_id template_values, signature_values = self._l10n_es_edi_facturae_export_facturae() xml_content = cleanup_xml_node(self.env['ir.qweb']._render('l10n_es_edi_facturae.account_invoice_facturae_export', template_values)) errors = [] try: xml_content = self._l10n_es_facturae_sign_xml(xml_content, signature_values) except ValueError: errors.append(_('No valid certificate found for this company, Facturae EDI file will not be signed.\n')) return xml_content, errors # ------------------------------------------------------------------------- # IMPORT # ------------------------------------------------------------------------- def _get_import_file_type(self, file_data): """ Identify Factura-E files. """ # EXTENDS 'account' def is_facturae(tree): return tree.tag in [ '{http://www.facturae.es/Facturae/2014/v3.2.1/Facturae}Facturae', '{http://www.facturae.gob.es/formato/Versiones/Facturaev3_2_2.xml}Facturae', ] if file_data['xml_tree'] is not None and is_facturae(file_data['xml_tree']): return 'l10n_es.facturae' return super()._get_import_file_type(file_data) def _unwrap_attachment(self, file_data, recurse=True): """ Divide a Facturae file into constituent invoices and create a new attachment for each invoice after the first. """ # EXTENDS 'account' if file_data['import_file_type'] != 'l10n_es.facturae': return super()._unwrap_attachment(file_data, recurse) embedded = self._split_xml_into_new_attachments(file_data, tag='Invoice') if embedded and recurse: embedded.extend(self._unwrap_attachments(embedded, recurse=True)) return embedded def _get_edi_decoder(self, file_data, new=False): # EXTENDS 'account' if file_data['import_file_type'] == 'l10n_es.facturae': return { 'priority': 20, 'decoder': self._import_invoice_facturae, } return super()._get_edi_decoder(file_data, new) def _import_invoice_facturae(self, invoice, file_data, new=False): tree = file_data['xml_tree'] is_bill = invoice.move_type.startswith('in_') partner = self._import_get_partner(tree, is_bill) # Only decode the first invoice of the Factura-e file. tree = tree.xpath('//Invoice')[0] self._import_invoice_facturae_invoice(invoice, partner, tree) def _import_get_partner(self, tree, is_bill): # If we're dealing with a vendor bill, then the partner is the seller party, if an invoice then it's the buyer. party = tree.xpath('//SellerParty') if is_bill else tree.xpath('//BuyerParty') if party: partner_vals = self._import_extract_partner_values(party[0]) return self._import_create_or_retrieve_partner(partner_vals) return None def _import_extract_partner_values(self, party_node): name = find_xml_value('.//CorporateName|.//Name', party_node) first_surname = find_xml_value('.//FirstSurname', party_node) second_surname = find_xml_value('.//SecondSurname', party_node) phone = find_xml_value('.//Telephone', party_node) mail = find_xml_value('.//ElectronicMail', party_node) country_code = find_xml_value('.//CountryCode', party_node) vat = find_xml_value('.//TaxIdentificationNumber', party_node) full_name = ' '.join(part for part in [name, first_surname, second_surname] if part) return {'name': full_name, 'vat': vat, 'phone': phone, 'email': mail, 'country_code': country_code} def _import_create_or_retrieve_partner(self, partner_vals): name = partner_vals['name'] vat = partner_vals['vat'] phone = partner_vals['phone'] email = partner_vals['email'] country_code = partner_vals['country_code'] partner = self.env['res.partner']._retrieve_partner(name=name, vat=vat, phone=phone, email=email) if not partner and name: partner_vals = {'name': name, 'email': email, 'phone': phone} country_code = REVERSED_COUNTRY_CODE.get(country_code) country = self.env['res.country'].search([('code', '=', country_code)]) if country_code else False if country: partner_vals['country_id'] = country.id partner = self.env['res.partner'].create(partner_vals) partner.vat, _country_code = self.env['res.partner']._run_vat_checks(country, vat, validation='setnull') return partner def _import_invoice_facturae_invoice(self, invoice, partner, tree): logs = [] # ==== move_type ==== invoice_total = find_xml_value('.//InvoiceTotal', tree) is_refund = float(invoice_total) < 0 if invoice_total else False if is_refund: invoice.move_type = "in_refund" if invoice.move_type.startswith("in_") else "out_refund" ref_multiplier = -1.0 if is_refund else 1.0 # ==== partner_id ==== if partner: invoice.partner_id = partner else: logs.append(_("Customer/Vendor could not be found and could not be created due to missing data in the XML.")) # ==== currency_id ==== invoice_currency_code = find_xml_value('.//InvoiceCurrencyCode', tree) if invoice_currency_code: currency = self.env['res.currency'].search([('name', '=', invoice_currency_code)], limit=1) if currency: invoice.currency_id = currency else: logs.append(_("Could not retrieve currency: %s. Did you enable the multicurrency option " "and activate the currency?", invoice_currency_code)) # ==== invoice date ==== if issue_date := find_xml_value('.//IssueDate', tree): invoice.invoice_date = issue_date # ==== invoice_date_due ==== if end_date := find_xml_value('.//InstallmentDueDate', tree): invoice.invoice_date_due = end_date # ==== ref ==== if invoice_number := find_xml_value('.//InvoiceNumber', tree): invoice.ref = invoice_number # ==== narration ==== invoice.narration = "\n".join( ref.text for ref in tree.xpath('.//LegalReference') if ref.text ) # === invoice_line_ids === logs += self._import_invoice_fill_lines(invoice, tree, ref_multiplier) body = Markup("%s") % _("Invoice imported from Factura-E XML file.") if logs: body += Markup("") \ % Markup().join(Markup("
  • %s
    • ") % log for log in logs) invoice.message_post(body=body) return logs def _import_invoice_fill_lines(self, invoice, tree, ref_multiplier): lines = tree.xpath('.//InvoiceLine') logs = [] vals_list = [] for line in lines: line_vals = {'move_id': invoice.id} # ==== name ==== if item_description := find_xml_value('.//ItemDescription', line): product = self._search_product_for_import(item_description) if product: line_vals['product_id'] = product.id else: logs.append(_("The product '%s' could not be found.", item_description)) line_vals['name'] = item_description # ==== quantity ==== line_vals['quantity'] = find_xml_value('.//Quantity', line) or 1 # ==== price_unit ==== price_unit = find_xml_value('.//UnitPriceWithoutTax', line) line_vals['price_unit'] = ref_multiplier * float(price_unit) if price_unit else 1.0 # ==== discount ==== discounts = line.xpath('.//DiscountRate') discount_rate = 0.0 for discount in discounts: discount_rate += float(discount.text) charges = line.xpath('.//ChargeRate') charge_rate = 0.0 for charge in charges: charge_rate += float(charge.text) discount_rate -= charge_rate line_vals['discount'] = discount_rate # ==== tax_ids ==== taxes_withheld_nodes = line.xpath('.//TaxesWithheld/Tax') taxes_outputs_nodes = line.xpath('.//TaxesOutputs/Tax') is_purchase = invoice.move_type.startswith('in') tax_ids = [] logs += self._import_fill_invoice_line_taxes(invoice, line_vals, tax_ids, taxes_outputs_nodes, False, is_purchase) logs += self._import_fill_invoice_line_taxes(invoice, line_vals, tax_ids, taxes_withheld_nodes, True, is_purchase) line_vals['tax_ids'] = [Command.set(tax_ids)] vals_list.append(line_vals) invoice.invoice_line_ids = self.env['account.move.line'].create(vals_list) return logs def _import_fill_invoice_line_taxes(self, invoice, line_vals, tax_ids, tax_nodes, is_withheld, is_purchase): logs = [] for tax_node in tax_nodes: tax_rate = find_xml_value('.//TaxRate', tax_node) if tax_rate: # Since the 'TaxRate' node isn't guaranteed to be a percentage, we can find out by # applying the tax rate on the taxable base, and if it's equal to the tax amount # then we can say this is a percentage, otherwise a fixed amount. taxable_base = find_xml_value('.//TaxableBase/TotalAmount', tax_node) tax_amount = find_xml_value('.//TaxAmount/TotalAmount', tax_node) is_fixed = False if taxable_base and tax_amount and invoice.currency_id.compare_amounts(float(taxable_base) * (float(tax_rate) / 100), float(tax_amount)) != 0: is_fixed = True tax_excl = self._search_tax_for_import(invoice.company_id, float(tax_rate), is_fixed, is_withheld, is_purchase, price_included=False) if tax_excl: tax_ids.append(tax_excl.id) elif tax_incl := self._search_tax_for_import(invoice.company_id, float(tax_rate), is_fixed, is_withheld, is_purchase, price_included=True): tax_ids.append(tax_incl) line_vals['price_unit'] *= (1.0 + float(tax_rate) / 100.0) else: logs.append(_("Could not retrieve the tax: %(tax_rate)s %% for line '%(line)s'.", tax_rate=tax_rate, line=line_vals.get('name', ""))) return logs def _search_tax_for_import(self, company, amount, is_fixed, is_withheld, is_purchase, price_included): taxes = self.env['account.tax'].search([ ('company_id', '=', company.id), ('amount', '=', -1.0 * amount if is_withheld else amount), ('amount_type', '=', 'fixed' if is_fixed else 'percent'), ('type_tax_use', '=', 'purchase' if is_purchase else 'sale'), ('price_include', '=', price_included), ], limit=1) return taxes def _search_product_for_import(self, item_description): # Exported Odoo XML will have item_description = "[default_code] name". # We can check if it follows the same format and search for the product with the default code and the name. code_and_name = re.match(r"(\[(?P.*?)\]\s)?(?P.*)", item_description).groupdict() product = self.env['product.product']._retrieve_product(**code_and_name) return product # ------------------------------------------------------------------------- # ACTION METHODS # ------------------------------------------------------------------------- def action_invoice_download_facturae(self): if invoices_with_facturae := self.filtered('l10n_es_edi_facturae_xml_id'): return { 'type': 'ir.actions.act_url', 'url': f'/account/download_invoice_documents/{",".join(map(str, invoices_with_facturae.ids))}/facturae', 'target': 'download', } return False # ------------------------------------------------------------------------- # BUSINESS METHODS # # ------------------------------------------------------------------------- def _l10n_es_facturae_sign_xml(self, edi_data, signature_data): """ Signs the given XML data with the certificate and private key. :param etree._Element edi_data: The XML data to sign. :param dict signature_data: The signature data to use. :return: The signed XML data string. :rtype: str """ self.ensure_one() certificates_sudo = self.company_id.sudo().l10n_es_edi_facturae_certificate_ids.filtered("is_valid") if not certificates_sudo: raise UserError(_('No valid certificate found')) certificate_sudo = certificates_sudo[0] root = deepcopy(edi_data) e, n = certificate_sudo._get_public_key_numbers_bytes() issuer = certificate_sudo._l10n_es_edi_facturae_get_issuer() # Identifiers document_id = f"Document-{sha1(etree.tostring(edi_data)).hexdigest()}" signature_id = f"Signature-{document_id}" keyinfo_id = f"KeyInfo-{document_id}" sigproperties_id = f"SignatureProperties-{document_id}" signature_data.update({ 'document_id': document_id, 'x509_certificate': base64.encodebytes(base64.b64decode(certificate_sudo._get_der_certificate_bytes())).decode(), 'public_modulus': n.decode(), 'public_exponent': e.decode(), 'iso_now': fields.Datetime.now().isoformat(), 'keyinfo_id': keyinfo_id, 'signature_id': signature_id, 'sigproperties_id': sigproperties_id, 'reference_uri': f"Reference-{document_id}", 'sigpolicy_url': "http://www.facturae.es/politica_de_firma_formato_facturae/politica_de_firma_formato_facturae_v3_1.pdf", 'sigpolicy_description': "Política de firma electrónica para facturación electrónica con formato Facturae", 'sigcertif_digest': certificate_sudo._get_fingerprint_bytes(formatting='base64').decode(), 'x509_issuer_description': issuer, 'x509_serial_number': int(certificate_sudo.serial_number), }) signature = self.env['ir.qweb']._render('l10n_es_edi_facturae.template_xades_signature', signature_data) signature = cleanup_xml_node(signature, remove_blank_nodes=False) root.append(signature) _reference_digests(signature.find("ds:SignedInfo", namespaces=NS_MAP)) signed_info_xml = signature.find("ds:SignedInfo", namespaces=NS_MAP) signature.find("ds:SignatureValue", namespaces=NS_MAP).text = certificate_sudo._sign(_canonicalize_node(signed_info_xml)).decode() return etree.tostring(root, xml_declaration=True, encoding='UTF-8', standalone=True) def _get_invoice_legal_documents(self, filetype, allow_fallback=False): # EXTENDS 'account' self.ensure_one() if filetype == 'facturae': if facturae_attachment := self.l10n_es_edi_facturae_xml_id: return { 'filename': facturae_attachment.name, 'filetype': 'xml', 'content': facturae_attachment.raw, } return super()._get_invoice_legal_documents(filetype, allow_fallback=allow_fallback) def get_extra_print_items(self): # EXTENDS 'account' - add possibility to download Factura-e XML files print_items = super().get_extra_print_items() if self.filtered('l10n_es_edi_facturae_xml_id'): print_items.append({ 'key': 'download_xml_facturae', 'description': _('Factura-e XML'), **self.action_invoice_download_facturae(), }) return print_items