diff options
Diffstat (limited to 'addons/l10n_it_edi_sdicoop/models')
4 files changed, 336 insertions, 0 deletions
diff --git a/addons/l10n_it_edi_sdicoop/models/__init__.py b/addons/l10n_it_edi_sdicoop/models/__init__.py new file mode 100644 index 00000000..b297cf8a --- /dev/null +++ b/addons/l10n_it_edi_sdicoop/models/__init__.py @@ -0,0 +1,6 @@ +# -*- coding: utf-8 -*- +# Part of Odoo. See LICENSE file for full copyright and licensing details. + +from . import account_invoice +from . import account_edi_format +from . import res_config_settings diff --git a/addons/l10n_it_edi_sdicoop/models/account_edi_format.py b/addons/l10n_it_edi_sdicoop/models/account_edi_format.py new file mode 100644 index 00000000..95db0ada --- /dev/null +++ b/addons/l10n_it_edi_sdicoop/models/account_edi_format.py @@ -0,0 +1,281 @@ +# -*- coding:utf-8 -*- +# Part of Odoo. See LICENSE file for full copyright and licensing details. + +from odoo import models, _, _lt +from odoo.exceptions import UserError +from odoo.addons.account_edi_proxy_client.models.account_edi_proxy_user import AccountEdiProxyError, SERVER_URL + +from lxml import etree +import base64 +import logging + +_logger = logging.getLogger(__name__) + + +class AccountEdiFormat(models.Model): + _inherit = 'account.edi.format' + + # ------------------------------------------------------------------------- + # Import + # ------------------------------------------------------------------------- + + def _cron_receive_fattura_pa(self): + ''' Check the proxy for incoming invoices. + ''' + if self.env['ir.config_parameter'].get_param('account_edi_proxy_client.demo', False): + return + + proxy_users = self.env['account_edi_proxy_client.user'].search([('edi_format_id', '=', self.env.ref('l10n_it_edi.edi_fatturaPA').id)]) + for proxy_user in proxy_users: + company = proxy_user.company_id + try: + res = proxy_user._make_request(SERVER_URL + '/api/l10n_it_edi/1/in/RicezioneInvoice', + params={'recipient_codice_fiscale': company.l10n_it_codice_fiscale}) + except AccountEdiProxyError as e: + _logger.error('Error while receiving file from SdiCoop: %s', e) + + proxy_acks = [] + for id_transaction, fattura in res.items(): + if self.env['ir.attachment'].search([('name', '=', fattura['filename']), ('res_model', '=', 'account.move')], limit=1): + # name should be unique, the invoice already exists + _logger.info('E-invoice already exist: %s', fattura['filename']) + proxy_acks.append(id_transaction) + continue + + file = proxy_user._decrypt_data(fattura['file'], fattura['key']) + + try: + tree = etree.fromstring(file) + except Exception: + # should not happen as the file has been checked by SdiCoop + _logger.info('Received file badly formatted, skipping: \n %s', file) + continue + + invoice = self.env.ref('l10n_it_edi.edi_fatturaPA')._create_invoice_from_xml_tree(fattura['filename'], tree) + self.env['ir.attachment'].create({ + 'name': fattura['filename'], + 'raw': file, + 'type': 'binary', + 'res_model': 'account.move', + 'res_id': invoice.id + }) + + proxy_acks.append(id_transaction) + + if proxy_acks: + try: + proxy_user._make_request(SERVER_URL + '/api/l10n_it_edi/1/ack', + params={'transaction_ids': proxy_acks}) + except AccountEdiProxyError as e: + _logger.error('Error while receiving file from SdiCoop: %s', e) + + # ------------------------------------------------------------------------- + # Export + # ------------------------------------------------------------------------- + + def _check_move_configuration(self, move): + # OVERRIDE + res = super()._check_move_configuration(move) + if self.code != 'fattura_pa': + return res + + res.extend(self._l10n_it_edi_check_invoice_configuration(move)) + + if not self._get_proxy_user(move.company_id): + res.append(_("You must accept the terms and conditions in the settings to use FatturaPA.")) + + return res + + def _needs_web_services(self): + self.ensure_one() + return self.code == 'fattura_pa' or super()._needs_web_services() + + def _l10n_it_edi_is_required_for_invoice(self, invoice): + """ _is_required_for_invoice for SdiCoop. + OVERRIDE + """ + return invoice.is_sale_document() and invoice.country_code == 'IT' + + def _support_batching(self, move=None, state=None, company=None): + # OVERRIDE + if self.code == 'fattura_pa': + return state == 'to_send' and move.is_invoice() + + return super()._support_batching(move=move, state=state, company=company) + + def _l10n_it_post_invoices_step_1(self, invoices): + ''' Send the invoices to the proxy. + ''' + to_return = {} + + to_send = {} + for invoice in invoices: + xml = b"<?xml version='1.0' encoding='UTF-8'?>" + invoice._export_as_xml() + filename = self._l10n_it_edi_generate_electronic_invoice_filename(invoice) + attachment = self.env['ir.attachment'].create({ + 'name': filename, + 'res_id': invoice.id, + 'res_model': invoice._name, + 'datas': base64.encodebytes(xml), + 'description': _('Italian invoice: %s', invoice.move_type), + 'type': 'binary', + }) + invoice.l10n_it_edi_attachment_id = attachment + + if len(invoice.commercial_partner_id.l10n_it_pa_index or '') == 6: + invoice.message_post( + body=(_("Invoices for PA are not managed by Odoo, you can download the document and send it on your own.")) + ) + to_return[invoice] = {'attachment': attachment} + else: + to_send[filename] = { + 'invoice': invoice, + 'data': {'filename': filename, 'xml': base64.b64encode(xml)}} + + company = invoices.company_id + proxy_user = self._get_proxy_user(company) + if not proxy_user: # proxy user should exist, because there is a check in _check_move_configuration + return {invoice: { + 'error': _("You must accept the terms and conditions in the settings to use FatturaPA."), + 'blocking_level': 'error'} for invoice in invoices} + + if self.env['ir.config_parameter'].get_param('account_edi_proxy_client.demo', False): + responses = {filename: {'id_transaction': 'demo'} for invoice in invoices} + else: + try: + responses = self._l10n_it_edi_upload([i['data'] for i in to_send.values()], proxy_user) + except AccountEdiProxyError as e: + return {invoice: {'error': e.message, 'blocking_level': 'error'} for invoice in invoices} + + for filename, response in responses.items(): + invoice = to_send[filename]['invoice'] + to_return[invoice] = response + if 'id_transaction' in response: + invoice.l10n_it_edi_transaction = response['id_transaction'] + to_return[invoice].update({ + 'error': _('The invoice was successfully transmitted to the Public Administration and we are waiting for confirmation.'), + 'blocking_level': 'info', + }) + return to_return + + def _l10n_it_post_invoices_step_2(self, invoices): + ''' Check if the sent invoices have been processed by FatturaPA. + ''' + to_check = {i.l10n_it_edi_transaction: i for i in invoices} + to_return = {} + company = invoices.company_id + proxy_user = self._get_proxy_user(company) + if not proxy_user: # proxy user should exist, because there is a check in _check_move_configuration + return {invoice: { + 'error': _("You must accept the terms and conditions in the settings to use FatturaPA."), + 'blocking_level': 'error'} for invoice in invoices} + + if self.env['ir.config_parameter'].get_param('account_edi_proxy_client.demo', False): + # simulate success and bypass ack + return {invoice: {'attachment': invoice.l10n_it_edi_attachment_id} for invoice in invoices} + else: + try: + responses = proxy_user._make_request(SERVER_URL + '/api/l10n_it_edi/1/in/TrasmissioneFatture', + params={'ids_transaction': list(to_check.keys())}) + except AccountEdiProxyError as e: + return {invoice: {'error': e.message, 'blocking_level': 'error'} for invoice in invoices} + + proxy_acks = [] + for id_transaction, response in responses.items(): + invoice = to_check[id_transaction] + if 'error' in response: + to_return[invoice] = response + continue + + state = response['state'] + if state == 'awaiting_outcome': + to_return[invoice] = { + 'error': _('The invoice was successfully transmitted to the Public Administration and we are waiting for confirmation'), + 'blocking_level': 'info', + } + proxy_acks.append(id_transaction) + continue + elif state == 'not_found': + # Invoice does not exist on proxy. Either it does not belong to this proxy_user or it was not created correctly when + # it was sent to the proxy. + to_return[invoice] = {'error': _('You are not allowed to check the status of this invoice.'), 'blocking_level': 'error'} + continue + + xml = proxy_user._decrypt_data(response['file'], response['key']) + response_tree = etree.fromstring(xml) + if state == 'ricevutaConsegna': + to_return[invoice] = {'error': _('The invoice has been succesfully transmitted. The addressee has 15 days to accept or reject it.')} + elif state == 'notificaScarto': + errors = [element.find('Descrizione').text for element in response_tree.xpath('//Errore')] + to_return[invoice] = {'error': self._format_error_message(_('The invoice has been refused by the Exchange System'), errors), 'blocking_level': 'error'} + elif state == 'notificaMancataConsegna': + to_return[invoice] = { + 'error': _('The E-invoice is not delivered to the addressee. The Exchange System is\ + unable to deliver the file to the Public Administration. The Exchange System will\ + contact the PA to report the problem and request that they provide a solution. \ + During the following 15 days, the Exchange System will try to forward the FatturaPA\ + file to the Administration in question again.'), + } + elif state == 'notificaEsito': + outcome = response_tree.find('Esito').text + if outcome == 'EC01': + to_return[invoice] = {'attachment': invoice.l10n_it_edi_attachment_id} + else: # ECO2 + to_return[invoice] = {'error': _('The invoice was refused by the addressee.'), 'blocking_level': 'error'} + elif state == 'NotificaDecorrenzaTermini': + to_return[invoice] = {'error': _('Expiration of the maximum term for communication of acceptance/refusal'), 'blocking_level': 'error'} + proxy_acks.append(id_transaction) + + try: + proxy_user._make_request(SERVER_URL + '/api/l10n_it_edi/1/ack', + params={'transaction_ids': proxy_acks}) + except AccountEdiProxyError as e: + # Will be ignored and acked again next time. + _logger.error('Error while acking file to SdiCoop: %s', e) + + return to_return + + def _post_fattura_pa(self, invoices): + # OVERRIDE + if not invoices.l10n_it_edi_transaction: + return self._l10n_it_post_invoices_step_1(invoices) + else: + return self._l10n_it_post_invoices_step_2(invoices) + + # ------------------------------------------------------------------------- + # Proxy methods + # ------------------------------------------------------------------------- + + def _get_proxy_identification(self, company): + if self.code != 'fattura_pa': + return super()._get_proxy_identification() + + if not company.l10n_it_codice_fiscale: + raise UserError(_('Please fill your codice fiscale to be able to receive invoices from FatturaPA')) + + return self.env['res.partner']._l10n_it_normalize_codice_fiscale(company.l10n_it_codice_fiscale) + + def _l10n_it_edi_upload(self, files, proxy_user): + '''Upload files to fatturapa. + + :param files: A list of dictionary {filename, base64_xml}. + :returns: A dictionary. + * message: Message from fatturapa. + * transactionId: The fatturapa ID of this request. + * error: An eventual error. + * error_level: Info, warning, error. + ''' + ERRORS = { + 'EI01': {'error': _lt('Attached file is empty'), 'blocking_level': 'error'}, + 'EI02': {'error': _lt('Service momentarily unavailable'), 'blocking_level': 'warning'}, + 'EI03': {'error': _lt('Unauthorized user'), 'blocking_level': 'error'}, + } + + result = proxy_user._make_request(SERVER_URL + '/api/l10n_it_edi/1/out/SdiRiceviFile', params={'files': files}) + + # Translate the errors. + for filename in result.keys(): + if 'error' in result[filename]: + result[filename] = ERRORS.get(result[filename]['error'], {'error': result[filename]['error'], 'blocking_level': 'error'}) + + return result diff --git a/addons/l10n_it_edi_sdicoop/models/account_invoice.py b/addons/l10n_it_edi_sdicoop/models/account_invoice.py new file mode 100644 index 00000000..4221dbc0 --- /dev/null +++ b/addons/l10n_it_edi_sdicoop/models/account_invoice.py @@ -0,0 +1,24 @@ +# -*- coding:utf-8 -*- +# Part of Odoo. See LICENSE file for full copyright and licensing details. +import logging + +from odoo import fields, models + + +_logger = logging.getLogger(__name__) + +DEFAULT_FACTUR_ITALIAN_DATE_FORMAT = '%Y-%m-%d' + + +class AccountMove(models.Model): + _inherit = 'account.move' + + l10n_it_edi_transaction = fields.Char(copy=False) + l10n_it_edi_attachment_id = fields.Many2one('ir.attachment', copy=False) + + def send_pec_mail(self): + self.ensure_one() + # OVERRIDE + # With SdiCoop web-service, no need to send PEC mail. + # Set the state to 'other' because the invoice should not be managed par l10n_it_edi. + self.l10n_it_send_state = 'other' diff --git a/addons/l10n_it_edi_sdicoop/models/res_config_settings.py b/addons/l10n_it_edi_sdicoop/models/res_config_settings.py new file mode 100644 index 00000000..04d78743 --- /dev/null +++ b/addons/l10n_it_edi_sdicoop/models/res_config_settings.py @@ -0,0 +1,25 @@ +# -*- coding: utf-8 -*- + +from odoo import api, models, fields + + +class ResConfigSettings(models.TransientModel): + _inherit = 'res.config.settings' + + is_edi_proxy_active = fields.Boolean(compute='_compute_is_edi_proxy_active') + + @api.depends('company_id.account_edi_proxy_client_ids', 'company_id.account_edi_proxy_client_ids.active') + def _compute_is_edi_proxy_active(self): + for config in self: + config.is_edi_proxy_active = config.company_id.account_edi_proxy_client_ids + + def button_create_proxy_user(self): + # For now, only fattura_pa uses the proxy. + # To use it for more, we have to either make the activation of the proxy on a format basis + # or create a user per format here (but also when installing new formats) + fattura_pa = self.env.ref('l10n_it_edi.edi_fatturaPA') + edi_identification = fattura_pa._get_proxy_identification(self.company_id) + if not edi_identification: + return + + self.env['account_edi_proxy_client.user']._register_proxy_user(self.company_id, fattura_pa, edi_identification) |