odoo_account_ebics/account_ebics/wizards/ebics_xfer.py

# Copyright 2009-2021 Noviat.
# License LGPL-3 or later (http://www.gnu.org/licenses/lpgl).

"""
import logging
logging.basicConfig(
    level=logging.DEBUG,
    format='[%(asctime)s] %(levelname)s - %(name)s: %(message)s')
"""

import base64
import logging
import os
from sys import exc_info
from traceback import format_exception

from odoo import api, fields, models, _
from odoo.exceptions import UserError

_logger = logging.getLogger(__name__)

try:
    import fintech
    from fintech.ebics import EbicsKeyRing, EbicsBank, EbicsUser, EbicsClient,\
        EbicsFunctionalError, EbicsTechnicalError, EbicsVerificationError
    fintech.cryptolib = 'cryptography'
except ImportError:
    EbicsBank = object
    _logger.warning('Failed to import fintech')


class EbicsBank(EbicsBank):

    def _next_order_id(self, partnerid):
        """
        EBICS protocol version H003 requires generation of the OrderID.
        The OrderID must be a string between 'A000' and 'ZZZZ' and
        unique for each partner id.
        """
        return hasattr(self, '_order_number') and self._order_number or 'A000'


class EbicsXfer(models.TransientModel):
    _name = 'ebics.xfer'
    _description = 'EBICS file transfer'

    ebics_config_id = fields.Many2one(
        comodel_name='ebics.config',
        string='EBICS Configuration',
        domain=[('state', '=', 'confirm')],
        default=lambda self: self._default_ebics_config_id())
    ebics_userid_id = fields.Many2one(
        comodel_name='ebics.userid',
        string='EBICS UserID')
    ebics_passphrase = fields.Char(
        string='EBICS Passphrase')
    date_from = fields.Date()
    date_to = fields.Date()
    upload_data = fields.Binary(string='File to Upload')
    upload_fname = fields.Char(
        string='Upload Filename', default='')
    upload_fname_dummy = fields.Char(
        related='upload_fname', string='Upload Filename', readonly=True)
    format_id = fields.Many2one(
        comodel_name='ebics.file.format',
        string='EBICS File Format',
        help="Select EBICS File Format to upload/download."
             "\nLeave blank to download all available files.")
    allowed_format_ids = fields.Many2many(
        related='ebics_config_id.ebics_file_format_ids',
        string='Allowed EBICS File Formats')
    order_type = fields.Char(
        related='format_id.order_type',
        string='Order Type',
        help="For most banks in France you should use the "
             "format neutral Order Types 'FUL' for upload "
             "and 'FDL' for download.")
    test_mode = fields.Boolean(
        string='Test Mode',
        help="Select this option to test if the syntax of "
             "the upload file is correct."
             "\nThis option is only available for "
             "Order Type 'FUL'.")
    note = fields.Text(string='EBICS file transfer Log', readonly=True)

    @api.model
    def _default_ebics_config_id(self):
        cfg_mod = self.env['ebics.config']
        cfg = cfg_mod.search(
            [('company_ids', 'in', self.env.user.company_ids.ids),
             ('state', '=', 'confirm')])
        if cfg and len(cfg) == 1:
            return cfg
        else:
            return cfg_mod

    @api.onchange('ebics_config_id')
    def _onchange_ebics_config_id(self):
        ebics_userids = self.ebics_config_id.ebics_userid_ids
        if self._context.get('ebics_download'):
            download_formats = self.ebics_config_id.ebics_file_format_ids\
                .filtered(lambda r: r.type == 'down')
            if len(download_formats) == 1:
                self.format_id = download_formats
            if len(ebics_userids) == 1:
                self.ebics_userid_id = ebics_userids
            else:
                transport_users = ebics_userids.filtered(
                    lambda r: r.signature_class == 'T')
                if len(transport_users) == 1:
                    self.ebics_userid_id = transport_users
        else:
            upload_formats = self.ebics_config_id.ebics_file_format_ids\
                .filtered(lambda r: r.type == 'up')
            if len(upload_formats) == 1:
                self.format_id = upload_formats
            if len(ebics_userids) == 1:
                self.ebics_userid_id = ebics_userids

    @api.onchange('upload_data')
    def _onchange_upload_data(self):
        self.upload_fname_dummy = self.upload_fname
        self._detect_upload_format()
        upload_formats = self.format_id \
            or self.ebics_config_id.ebics_file_format_ids.filtered(
                lambda r: r.type == 'up')
        if len(upload_formats) == 1:
            self.format_id = upload_formats

    @api.onchange('format_id')
    def _onchange_format_id(self):
        self.order_type = self.format_id.order_type

    def ebics_upload(self):
        self.ensure_one()
        ctx = self._context.copy()
        ebics_file = self._ebics_upload()
        if ebics_file:
            ctx['ebics_file_id'] = ebics_file.id
        module = __name__.split('addons.')[1].split('.')[0]
        result_view = self.env.ref(
            '%s.ebics_xfer_view_form_result' % module)
        return {
            'name': _('EBICS file transfer result'),
            'res_id': self.id,
            'view_type': 'form',
            'view_mode': 'form',
            'res_model': 'ebics.xfer',
            'view_id': result_view.id,
            'target': 'new',
            'context': ctx,
            'type': 'ir.actions.act_window',
        }

    def ebics_download(self):
        self.ensure_one()
        self.ebics_config_id._check_ebics_files()
        ctx = self._context.copy()
        self.note = ''
        client = self._setup_client()
        if client:
            download_formats = (
                self.format_id
                or self.ebics_config_id.ebics_file_format_ids.filtered(
                    lambda r: r.type == 'down'
                )
            )
            ebics_files = self.env['ebics.file']
            date_from = self.date_from and self.date_from.isoformat() or None
            date_to = self.date_to and self.date_to.isoformat() or None
            for df in download_formats:
                try:
                    success = False
                    if df.order_type == 'FDL':
                        data = client.FDL(df.name, date_from, date_to)
                    else:
                        params = None
                        if date_from and date_to:
                            params = {'DateRange': {
                                'Start': date_from,
                                'End': date_to,
                            }}
                        data = client.download(df.order_type, params=params)
                    ebics_files += self._handle_download_data(data, df)
                    success = True
                except EbicsFunctionalError:
                    e = exc_info()
                    self.note += '\n'
                    self.note += _(
                        "EBICS Functional Error during download of File Format %s (%s):"
                    ) % (df.name, df.order_type)
                    self.note += '\n'
                    self.note += '%s (code: %s)' % (e[1].message, e[1].code)
                except EbicsTechnicalError:
                    e = exc_info()
                    self.note += '\n'
                    self.note += _(
                        "EBICS Technical Error during download of File Format %s (%s):"
                    ) % (df.name, df.order_type)
                    self.note += '\n'
                    self.note += '%s (code: %s)' % (e[1].message, e[1].code)
                except EbicsVerificationError:
                    self.note += '\n'
                    self.note += _(
                        "EBICS Verification Error during download of "
                        "File Format %s (%s):"
                    ) % (df.name, df.order_type)
                    self.note += '\n'
                    self.note += _("The EBICS response could not be verified.")
                except UserError as e:
                    self.note += '\n'
                    self.note += _(
                        "Warning during download of File Format %s (%s):"
                    ) % (df.name, df.order_type)
                    self.note += '\n'
                    self.note += e.name
                except Exception:
                    self.note += '\n'
                    self.note += _(
                        "Unknown Error during download of File Format %s (%s):"
                    ) % (df.name, df.order_type)
                    tb = ''.join(format_exception(*exc_info()))
                    self.note += '\n%s' % tb
                else:
                    # mark received data so that it is not included in further
                    # downloads
                    trans_id = client.last_trans_id
                    client.confirm_download(trans_id=trans_id, success=success)

            ctx['ebics_file_ids'] = ebics_files._ids

            if ebics_files:
                self.note += '\n'
                for f in ebics_files:
                    self.note += _(
                        "EBICS File '%s' is available for further processing."
                    ) % f.name
                    self.note += '\n'

        module = __name__.split('addons.')[1].split('.')[0]
        result_view = self.env.ref(
            '%s.ebics_xfer_view_form_result' % module)
        return {
            'name': _('EBICS file transfer result'),
            'res_id': self.id,
            'view_type': 'form',
            'view_mode': 'form',
            'res_model': 'ebics.xfer',
            'view_id': result_view.id,
            'target': 'new',
            'context': ctx,
            'type': 'ir.actions.act_window',
        }

    def button_close(self):
        self.ensure_one()
        return {'type': 'ir.actions.act_window_close'}

    def view_ebics_file(self):
        self.ensure_one()
        module = __name__.split('addons.')[1].split('.')[0]
        act = self.env['ir.actions.act_window']._for_xml_id(
            '{}.ebics_file_action_download'.format(module))
        act['domain'] = [('id', 'in', self._context['ebics_file_ids'])]
        return act

    def _ebics_upload(self):
        self.ensure_one()
        ebics_file = self.env['ebics.file']
        self.note = ''
        client = self._setup_client()
        if client:
            upload_data = base64.decodestring(self.upload_data)
            ef_format = self.format_id
            OrderID = False
            try:
                order_type = self.order_type
                if order_type == 'FUL':
                    kwargs = {}
                    bank = self.ebics_config_id.journal_ids[0].bank_id
                    cc = bank.country.code
                    if cc:
                        kwargs['country'] = cc
                    if self.test_mode:
                        kwargs['TEST'] = 'TRUE'
                    OrderID = client.FUL(ef_format.name, upload_data, **kwargs)
                else:
                    OrderID = client.upload(order_type, upload_data)
                if OrderID:
                    self.note += '\n'
                    self.note += _(
                        "EBICS File has been uploaded (OrderID %s)."
                    ) % OrderID
                    ef_note = _("EBICS OrderID: %s") % OrderID
                    if self.env.context.get('origin'):
                        ef_note += '\n' + _(
                            "Origin: %s") % self._context['origin']
                    suffix = self.format_id.suffix
                    fn = self.upload_fname
                    if not fn.endswith(suffix):
                        fn = '.'.join[fn, suffix]
                    ef_vals = {
                        'name': self.upload_fname,
                        'data': self.upload_data,
                        'date': fields.Datetime.now(),
                        'format_id': self.format_id.id,
                        'state': 'done',
                        'user_id': self._uid,
                        'ebics_userid_id': self.ebics_userid_id.id,
                        'note': ef_note,
                        "company_ids": [
                            self.env.context.get("force_company", self.env.company.id)
                        ],
                    }
                    self._update_ef_vals(ef_vals)
                    ebics_file = self.env['ebics.file'].create(ef_vals)

            except EbicsFunctionalError:
                e = exc_info()
                self.note += '\n'
                self.note += _("EBICS Functional Error:")
                self.note += '\n'
                self.note += '%s (code: %s)' % (e[1].message, e[1].code)
            except EbicsTechnicalError:
                e = exc_info()
                self.note += '\n'
                self.note += _("EBICS Technical Error:")
                self.note += '\n'
                self.note += '%s (code: %s)' % (e[1].message, e[1].code)
            except EbicsVerificationError:
                self.note += '\n'
                self.note += _("EBICS Verification Error:")
                self.note += '\n'
                self.note += _("The EBICS response could not be verified.")
            except Exception:
                self.note += '\n'
                self.note += _("Unknown Error")
                tb = ''.join(format_exception(*exc_info()))
                self.note += '\n%s' % tb

            if self.ebics_config_id.ebics_version == 'H003' and OrderID:
                self.ebics_config_id._update_order_number(OrderID)

        return ebics_file

    def _setup_client(self):
        self.ebics_config_id._check_ebics_keys()
        passphrase = self._get_passphrase()
        keyring = EbicsKeyRing(
            keys=self.ebics_userid_id.ebics_keys_fn,
            passphrase=passphrase)

        bank = EbicsBank(
            keyring=keyring,
            hostid=self.ebics_config_id.ebics_host,
            url=self.ebics_config_id.ebics_url)
        if self.ebics_config_id.ebics_version == 'H003':
            bank._order_number = self.ebics_config_id._get_order_number()

        user = EbicsUser(
            keyring=keyring,
            partnerid=self.ebics_config_id.ebics_partner,
            userid=self.ebics_userid_id.name)
        signature_class = self.format_id.signature_class \
            or self.ebics_userid_id.signature_class
        if signature_class == 'T':
            user.manual_approval = True

        try:
            client = EbicsClient(
                bank, user, version=self.ebics_config_id.ebics_version)
        except Exception:
            self.note += '\n'
            self.note += _("Unknown Error")
            tb = ''.join(format_exception(*exc_info()))
            self.note += '\n%s' % tb
            client = False

        return client

    def _get_passphrase(self):
        passphrase = self.ebics_userid_id.ebics_passphrase

        if passphrase:
            return passphrase

        module = __name__.split('addons.')[1].split('.')[0]
        passphrase_view = self.env.ref(
            '%s.ebics_xfer_view_form_passphrase' % module)
        return {
            'name': _('EBICS file transfer'),
            'res_id': self.id,
            'view_type': 'form',
            'view_mode': 'form',
            'res_model': 'ebics.xfer',
            'view_id': passphrase_view.id,
            'target': 'new',
            'context': self._context,
            'type': 'ir.actions.act_window',
        }

    def _file_format_methods(self):
        """
        Extend this dictionary in order to add support
        for extra file formats.
        """
        res = {
            'camt.xxx.cfonb120.stm': self._handle_cfonb120,
            'camt.052.001.02.stm': self._handle_camt052,
            'camt.053.001.02.stm': self._handle_camt053,
        }
        return res

    def _update_ef_vals(self, ef_vals):
        """
        Adapt this method to customize the EBICS File values.
        """
        if self.format_id and self.format_id.type == 'up':
            fn = ef_vals['name']
            dups = self._check_duplicate_ebics_file(
                fn, self.format_id)
            if dups:
                n = 1
                fn = '_'.join([fn, str(n)])
                while self._check_duplicate_ebics_file(fn, self.format_id):
                    n += 1
                    fn = '_'.join([fn, str(n)])
                ef_vals['name'] = fn

    def _handle_download_data(self, data, file_format):
        ebics_files = self.env['ebics.file']
        if isinstance(data, dict):
            for doc in data:
                ebics_files += self._create_ebics_file(
                    data[doc], file_format, docname=doc)
        else:
            ebics_files += self._create_ebics_file(data, file_format)
        return ebics_files

    def _create_ebics_file(self, data, file_format, docname=None):
        """
        Write the data as received over the EBICS connection
        to a temporary file so that is is available for
        analysis (e.g. in case formats are received that cannot
        be handled in the current version of this module).

        TODO: add code to clean-up /tmp on a regular basis.

        After saving the data received we call the method to perform
        file format specific processing.
        """
        ebics_files_root = self.ebics_config_id.ebics_files
        tmp_dir = os.path.normpath(ebics_files_root + '/tmp')
        if not os.path.isdir(tmp_dir):
            os.makedirs(tmp_dir, mode=0o700)
        fn_parts = [self.ebics_config_id.ebics_host,
                    self.ebics_config_id.ebics_partner]
        if docname:
            fn_parts.append(docname)
        else:
            fn_date = self.date_to or fields.Date.today()
            fn_parts.append(fn_date.isoformat())
        base_fn = '_'.join(fn_parts)
        n = 1
        full_tmp_fn = os.path.normpath(tmp_dir + '/' + base_fn)
        while os.path.exists(full_tmp_fn):
            n += 1
            tmp_fn = base_fn + '_' + str(n).rjust(3, '0')
            full_tmp_fn = os.path.normpath(tmp_dir + '/' + tmp_fn)

        with open(full_tmp_fn, 'wb') as f:
            f.write(data)

        ff_methods = self._file_format_methods()
        if file_format.name in ff_methods:
            data = ff_methods[file_format.name](data)

        fn = '.'.join([base_fn, file_format.suffix])
        dups = self._check_duplicate_ebics_file(fn, file_format)
        if dups:
            raise UserError(_(
                "EBICS File with name '%s' has already been downloaded."
                "\nPlease check this file and rename in case there is "
                "no risk on duplicate transactions.")
                % fn)
        data = base64.encodestring(data)
        ef_vals = {
            'name': fn,
            'data': data,
            'date': fields.Datetime.now(),
            'date_from': self.date_from,
            'date_to': self.date_to,
            'format_id': file_format.id,
            'user_id': self._uid,
            'ebics_userid_id': self.ebics_userid_id.id,
            'company_ids': self.ebics_config_id.company_ids.ids,
        }
        self._update_ef_vals(ef_vals)
        ebics_file = self.env['ebics.file'].create(ef_vals)
        return ebics_file

    def _check_duplicate_ebics_file(self, fn, file_format):
        dups = self.env['ebics.file'].search(
            [('name', '=', fn),
             ('format_id', '=', file_format.id)])
        return dups

    def _detect_upload_format(self):
        """
        Use this method in order to automatically detect and set the
        EBICS upload file format.
        """
        pass

    def _update_order_number(self, OrderID):
        o_list = list(OrderID)
        for i, c in enumerate(reversed(o_list), start=1):
            if c == '9':
                o_list[-i] = 'A'
                break
            if c == 'Z':
                continue
            else:
                o_list[-i] = chr(ord(c) + 1)
                break
        next_nr = ''.join(o_list)
        if next_nr == 'ZZZZ':
            next_nr = 'A000'
        self.ebics_config_id.order_number = next_nr

    def _insert_line_terminator(self, data_in, line_len):
        data_in = data_in.replace(b'\n', b'').replace(b'\r', b'')
        data_out = b''
        max_len = len(data_in)
        i = 0
        while i + line_len <= max_len:
            data_out += data_in[i:i + line_len] + b'\n'
            i += line_len
        return data_out

    def _handle_cfonb120(self, data_in):
        return self._insert_line_terminator(data_in, 120)

    def _handle_cfonb240(self, data_in):
        return self._insert_line_terminator(data_in, 240)

    def _handle_camt052(self, data_in):
        """
        Use this method if you need to fix camt files received
        from your bank before passing them to the
        Odoo Community CAMT parser.
        Remark: Odoo Enterprise doesn't support camt.052.
        """
        return data_in

    def _handle_camt053(self, data_in):
        """
        Use this method if you need to fix camt files received
        from your bank before passing them to the
        Odoo Enterprise or Community CAMT parser.
        """
        return data_in