# -*- coding: utf-8 -*- # Part of Odoo. See LICENSE file for full copyright and licensing details. import io import zipfile from werkzeug.urls import url_encode from odoo import api, fields, models, _ from odoo.exceptions import UserError class AccountMove(models.Model): _inherit = 'account.move' edi_document_ids = fields.One2many( comodel_name='account.edi.document', inverse_name='move_id') edi_state = fields.Selection( selection=[('to_send', 'To Send'), ('sent', 'Sent'), ('to_cancel', 'To Cancel'), ('cancelled', 'Cancelled')], string="Electronic invoicing", store=True, compute='_compute_edi_state', help='The aggregated state of all the EDIs with web-service of this move') edi_error_count = fields.Integer( compute='_compute_edi_error_count', help='How many EDIs are in error for this move?') edi_blocking_level = fields.Selection( selection=[('info', 'Info'), ('warning', 'Warning'), ('error', 'Error')], compute='_compute_edi_error_message') edi_error_message = fields.Html( compute='_compute_edi_error_message') # Technical field to display the documents that will be processed by the CRON edi_web_services_to_process = fields.Text( compute='_compute_edi_web_services_to_process') edi_show_cancel_button = fields.Boolean( compute='_compute_edi_show_cancel_button') edi_show_abandon_cancel_button = fields.Boolean( compute='_compute_edi_show_abandon_cancel_button') edi_show_force_cancel_button = fields.Boolean( compute='_compute_edi_show_force_cancel_button') @api.depends('edi_document_ids.state') def _compute_edi_state(self): for move in self: all_states = set(move.edi_document_ids.filtered(lambda d: d.edi_format_id._needs_web_services()).mapped('state')) if all_states == {'sent'}: move.edi_state = 'sent' elif all_states == {'cancelled'}: move.edi_state = 'cancelled' elif 'to_send' in all_states: move.edi_state = 'to_send' elif 'to_cancel' in all_states: move.edi_state = 'to_cancel' else: move.edi_state = False @api.depends('edi_document_ids.state') def _compute_edi_show_force_cancel_button(self): for move in self: move.edi_show_force_cancel_button = move._can_force_cancel() @api.depends('edi_document_ids.error') def _compute_edi_error_count(self): for move in self: move.edi_error_count = len(move.edi_document_ids.filtered(lambda d: d.error)) @api.depends('edi_error_count', 'edi_document_ids.error', 'edi_document_ids.blocking_level') def _compute_edi_error_message(self): for move in self: if move.edi_error_count == 0: move.edi_error_message = None move.edi_blocking_level = None elif move.edi_error_count == 1: error_doc = move.edi_document_ids.filtered(lambda d: d.error) move.edi_error_message = error_doc.error move.edi_blocking_level = error_doc.blocking_level else: error_levels = set([doc.blocking_level for doc in move.edi_document_ids]) count = str(move.edi_error_count) if 'error' in error_levels: move.edi_error_message = _("%(count)s Electronic invoicing error(s)", count=count) move.edi_blocking_level = 'error' elif 'warning' in error_levels: move.edi_error_message = _("%(count)s Electronic invoicing warning(s)", count=count) move.edi_blocking_level = 'warning' else: move.edi_error_message = _("%(count)s Electronic invoicing info(s)", count=count) move.edi_blocking_level = 'info' @api.depends( 'edi_document_ids', 'edi_document_ids.state', 'edi_document_ids.blocking_level', 'edi_document_ids.edi_format_id', 'edi_document_ids.edi_format_id.name') def _compute_edi_web_services_to_process(self): for move in self: to_process = move.edi_document_ids.filtered(lambda d: d.state in ['to_send', 'to_cancel'] and d.blocking_level != 'error') format_web_services = to_process.edi_format_id.filtered(lambda f: f._needs_web_services()) move.edi_web_services_to_process = ', '.join(f.name for f in format_web_services) def _check_edi_documents_for_reset_to_draft(self): self.ensure_one() for doc in self.edi_document_ids: move_applicability = doc.edi_format_id._get_move_applicability(self) if doc.edi_format_id._needs_web_services() \ and doc.state in ('sent', 'to_cancel') \ and move_applicability \ and move_applicability.get('cancel'): return False return True @api.depends('edi_document_ids.state') def _compute_show_reset_to_draft_button(self): # OVERRIDE super()._compute_show_reset_to_draft_button() for move in self: if not move._check_edi_documents_for_reset_to_draft(): move.show_reset_to_draft_button = False @api.depends('edi_document_ids.state') def _compute_edi_show_cancel_button(self): for move in self: if move.state != 'posted': move.edi_show_cancel_button = False continue move.edi_show_cancel_button = False for doc in move.edi_document_ids: move_applicability = doc.edi_format_id._get_move_applicability(move) if doc.edi_format_id._needs_web_services() \ and doc.state == 'sent' \ and move_applicability \ and move_applicability.get('cancel'): move.edi_show_cancel_button = True break @api.depends('edi_document_ids.state') def _compute_edi_show_abandon_cancel_button(self): for move in self: move.edi_show_abandon_cancel_button = False for doc in move.sudo().edi_document_ids: move_applicability = doc.edi_format_id._get_move_applicability(move) if doc.edi_format_id._needs_web_services() \ and doc.state == 'to_cancel' \ and move_applicability \ and move_applicability.get('cancel'): move.edi_show_abandon_cancel_button = True break #################################################### # Export Electronic Document #################################################### def _prepare_edi_tax_details(self, filter_to_apply=None, filter_invl_to_apply=None, grouping_key_generator=None): ''' Compute amounts related to taxes for the current invoice. :param filter_to_apply: Optional filter to exclude some tax values from the final results. The filter is defined as a method getting a dictionary as parameter representing the tax values for a single repartition line. This dictionary contains: 'base_line_id': An account.move.line record. 'tax_id': An account.tax record. 'tax_repartition_line_id': An account.tax.repartition.line record. 'base_amount': The tax base amount expressed in company currency. 'tax_amount': The tax amount expressed in company currency. 'base_amount_currency': The tax base amount expressed in foreign currency. 'tax_amount_currency': The tax amount expressed in foreign currency. If the filter is returning False, it means the current tax values will be ignored when computing the final results. :param filter_invl_to_apply: Optional filter to exclude some invoice lines. :param grouping_key_generator: Optional method used to group tax values together. By default, the tax values are grouped by tax. This parameter is a method getting a dictionary as parameter (same signature as 'filter_to_apply'). This method must returns a dictionary where values will be used to create the grouping_key to aggregate tax values together. The returned dictionary is added to each tax details in order to retrieve the full grouping_key later. :return: The full tax details for the current invoice and for each invoice line separately. The returned dictionary is the following: 'base_amount': The total tax base amount in company currency for the whole invoice. 'tax_amount': The total tax amount in company currency for the whole invoice. 'base_amount_currency': The total tax base amount in foreign currency for the whole invoice. 'tax_amount_currency': The total tax amount in foreign currency for the whole invoice. 'tax_details': A mapping of each grouping key (see 'grouping_key_generator') to a dictionary containing: 'base_amount': The tax base amount in company currency for the current group. 'tax_amount': The tax amount in company currency for the current group. 'base_amount_currency': The tax base amount in foreign currency for the current group. 'tax_amount_currency': The tax amount in foreign currency for the current group. 'group_tax_details': The list of all tax values aggregated into this group. 'tax_details_per_record': A mapping of each invoice line to a dictionary containing: 'base_amount': The total tax base amount in company currency for the whole invoice line. 'tax_amount': The total tax amount in company currency for the whole invoice line. 'base_amount_currency': The total tax base amount in foreign currency for the whole invoice line. 'tax_amount_currency': The total tax amount in foreign currency for the whole invoice line. 'tax_details': A mapping of each grouping key (see 'grouping_key_generator') to a dictionary containing: 'base_amount': The tax base amount in company currency for the current group. 'tax_amount': The tax amount in company currency for the current group. 'base_amount_currency': The tax base amount in foreign currency for the current group. 'tax_amount_currency': The tax amount in foreign currency for the current group. 'group_tax_details': The list of all tax values aggregated into this group. ''' return self._prepare_invoice_aggregated_taxes( filter_invl_to_apply=filter_invl_to_apply, filter_tax_values_to_apply=filter_to_apply, grouping_key_generator=grouping_key_generator, ) def _is_ready_to_be_sent(self): # OVERRIDE # Prevent a mail to be sent to the customer if the EDI document is not sent. res = super()._is_ready_to_be_sent() if not res: return False edi_documents_to_send = self.edi_document_ids.filtered(lambda x: x.state == 'to_send') return not bool(edi_documents_to_send) def _post(self, soft=True): # OVERRIDE # Set the electronic document to be posted and post immediately for synchronous formats. posted = super()._post(soft=soft) edi_document_vals_list = [] for move in posted: for edi_format in move.journal_id.edi_format_ids: move_applicability = edi_format._get_move_applicability(move) if move_applicability: errors = edi_format._check_move_configuration(move) if errors: raise UserError(_("Invalid invoice configuration:\n\n%s", '\n'.join(errors))) existing_edi_document = move.edi_document_ids.filtered(lambda x: x.edi_format_id == edi_format) if existing_edi_document: existing_edi_document.sudo().write({ 'state': 'to_send', 'attachment_id': False, }) else: edi_document_vals_list.append({ 'edi_format_id': edi_format.id, 'move_id': move.id, 'state': 'to_send', }) self.env['account.edi.document'].create(edi_document_vals_list) posted.edi_document_ids._process_documents_no_web_services() if not self.env.context.get('skip_account_edi_cron_trigger'): self.env.ref('account_edi.ir_cron_edi_network')._trigger() return posted def button_force_cancel(self): """ Cancel the invoice without waiting for the cancellation request to succeed. """ for move in self: to_cancel_edi_documents = move.edi_document_ids.filtered(lambda doc: doc.state == 'to_cancel') move.message_post(body=_("This invoice was canceled while the EDIs %s still had a pending cancellation request.", ", ".join(to_cancel_edi_documents.mapped('edi_format_id.name')))) self.button_cancel() def button_cancel(self): # OVERRIDE # Set the electronic document to be canceled and cancel immediately for synchronous formats. res = super().button_cancel() self.edi_document_ids.filtered(lambda doc: doc.state != 'sent').write({'state': 'cancelled', 'error': False, 'blocking_level': False}) self.edi_document_ids.filtered(lambda doc: doc.state == 'sent').write({'state': 'to_cancel', 'error': False, 'blocking_level': False}) self.edi_document_ids._process_documents_no_web_services() self.env.ref('account_edi.ir_cron_edi_network')._trigger() return res def _edi_allow_button_draft(self): self.ensure_one() return not self.edi_show_cancel_button def button_draft(self): # OVERRIDE for move in self: if not move._edi_allow_button_draft(): raise UserError(_( "You can't edit the following journal entry %s because an electronic document has already been " "sent. Please use the 'Request EDI Cancellation' button instead.", move.display_name)) res = super().button_draft() self.edi_document_ids.write({'error': False, 'blocking_level': False}) self.edi_document_ids.filtered(lambda doc: doc.state == 'to_send').unlink() return res def button_cancel_posted_moves(self): '''Mark the edi.document related to this move to be canceled. ''' to_cancel_documents = self.env['account.edi.document'] for move in self: move._check_fiscal_lock_dates() is_move_marked = False for doc in move.edi_document_ids: move_applicability = doc.edi_format_id._get_move_applicability(move) if doc.edi_format_id._needs_web_services() \ and doc.state == 'sent' \ and move_applicability \ and move_applicability.get('cancel'): to_cancel_documents |= doc is_move_marked = True if is_move_marked: move.message_post(body=_("A cancellation of the EDI has been requested.")) to_cancel_documents.write({'state': 'to_cancel', 'error': False, 'blocking_level': False}) def button_abandon_cancel_posted_posted_moves(self): '''Cancel the request for cancellation of the EDI. ''' documents = self.env['account.edi.document'] for move in self: is_move_marked = False for doc in move.edi_document_ids: move_applicability = doc.edi_format_id._get_move_applicability(move) if doc.state == 'to_cancel' and move_applicability and move_applicability.get('cancel'): documents |= doc is_move_marked = True if is_move_marked: move.message_post(body=_("A request for cancellation of the EDI has been called off.")) documents.write({'state': 'sent', 'error': False, 'blocking_level': False}) def _get_edi_document(self, edi_format): return self.edi_document_ids.filtered(lambda d: d.edi_format_id == edi_format) def _get_edi_attachment(self, edi_format): return self._get_edi_document(edi_format).sudo().attachment_id # this override is to make sure that the main attachment is not the edi xml otherwise the attachment viewer will not work correctly def _message_set_main_attachment_id(self, attachments, force=False, filter_xml=True): if not force and len(attachments) > 1 and self.message_main_attachment_id in self.edi_document_ids.attachment_id: force = True super()._message_set_main_attachment_id(attachments, force=force, filter_xml=filter_xml) #################################################### # Business operations #################################################### def button_process_edi_web_services(self): self.ensure_one() self.action_process_edi_web_services(with_commit=False) def action_process_edi_web_services(self, with_commit=True): docs = self.edi_document_ids.filtered(lambda d: d.state in ('to_send', 'to_cancel') and d.blocking_level != 'error') docs._process_documents_web_services(with_commit=with_commit) def _retry_edi_documents_error_hook(self): ''' Hook called when edi_documents are retried. For example, when it's needed to clean a field. TO OVERRIDE ''' return def action_retry_edi_documents_error(self): self._retry_edi_documents_error_hook() self.edi_document_ids.write({'error': False, 'blocking_level': False}) self.action_process_edi_web_services() #################################################### # Mailing #################################################### def _process_attachments_for_template_post(self, mail_template): """ Add Edi attachments to templates. """ result = super()._process_attachments_for_template_post(mail_template) for move in self.filtered('edi_document_ids'): move_result = result.setdefault(move.id, {}) for edi_doc in move.edi_document_ids: edi_attachments = edi_doc._filter_edi_attachments_for_mailing() move_result.setdefault('attachment_ids', []).extend(edi_attachments.get('attachment_ids', [])) move_result.setdefault('attachments', []).extend(edi_attachments.get('attachments', [])) return result