384 lines
19 KiB
Python
384 lines
19 KiB
Python
# -*- 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)
|
|
|
|
@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:
|
|
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 in ('sent', 'to_cancel') \
|
|
and move_applicability \
|
|
and move_applicability.get('cancel'):
|
|
move.show_reset_to_draft_button = False
|
|
break
|
|
|
|
@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 button_draft(self):
|
|
# OVERRIDE
|
|
for move in self:
|
|
if move.edi_show_cancel_button:
|
|
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
|