<commit_before># -*- coding: UTF-8 -*-
# Copyright 2014-2015 <NAME>
# This file is part of Lino Cosi.
#
# Lino Cosi is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# Lino Cosi is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public
# License along with Lino Cosi.  If not, see
# <http://www.gnu.org/licenses/>.


"""
Database models for `lino_cosi.lib.sepa`.

"""

from __future__ import unicode_literals
import logging
from pprint import pformat
import glob
import os
from django.db import models
from django.core.exceptions import MultipleObjectsReturned
from lino.api import dd, _, rt
from lino.core.utils import ChangeWatcher
from .camt import CamtParser
from .fields import IBANField, BICField
from .utils import belgian_nban_to_iban_bic, iban2bic
import time

logger = logging.getLogger(__name__)


class ImportStatements(dd.Action):
    """Import the .xml files found in the directory specified at
    :attr:`import_statements_path
    <lino_cosi.lib.sepa.Plugin.import_statements_path>`.

    End-users invoke this via the menu command :menuselection:`SEPA
    --> Import SEPA`.

    When a file has been successfully imported, Lino deletes it.

    It might happen that an .xml file accidentally gets downloaded a
    second time. Lino does not create these statements again.

    """
    label = _("Import SEPA")
    http_method = 'POST'
    select_rows = False

    def get_view_permission(self, profile):
        """Make it invisible when :attr:`import_statements_path
        <lino_cosi.lib.sepa.Plugin.import_statements_path>` is empty.

        """
        if not dd.plugins.sepa.import_statements_path:
            return False
        return super(ImportStatements, self).get_view_permission(profile)

    def run_from_ui(self, ar):
        pth = dd.plugins.sepa.import_statements_path
        wc = os.path.join(pth, '*.xml')
        dd.logger.info("Importing SEPA statements from %s...", wc)
        count = 0
        if pth:
            for filename in glob.iglob(wc):
                self.import_file(ar, filename)
                count += 1
            msg = "{0} xml files would have been imported.".format(count)
            dd.logger.info(msg)
            return ar.success(msg, alert=_("Success"))
        msg = "No import_statements_path configured."
        return ar.error(msg, alert=_("Error"))

    def import_file(self, ar, filename):
        Account = rt.modules.sepa.Account

        # Parse a CAMT053 XML file.
        parser = CamtParser()
        data_file = open(filename, 'rb').read()
        num = 0
        failed_statements = {}
        try:
            dd.logger.info("Parsing %s with camt.", filename)
            res = parser.parse(data_file)
            if res is None:
                raise Exception("res is None")
            for _statement in res:
                iban = _statement['account_number']
                if iban is None:
                    # msg = "Statement without account number : {0}"
                    failed_statements[num] = "IBAN Not found"
                    continue
                    # raise Exception(msg.format(pformat(_statement)))
                try:
                    account = Account.objects.get(iban=iban)
                except Account.DoesNotExist:
                    account = Account(iban=iban)
                    account.full_clean()
                    account.save()
                except MultipleObjectsReturned:
                    msg = "Found more than one account with IBAN {0}"
                    raise Exception(msg.format(iban))
                s = Statement(account=account,
                              date=_statement['date'].strftime("%Y-%m-%d"),
                              date_done=time.strftime("%Y-%m-%d"),
                              statement_number=_statement['name'],
                              balance_end=_statement['balance_end'],
                              balance_start=_statement['balance_start'],
                              balance_end_real=_statement['balance_end_real'],
                              currency_code=_statement['currency_code'])
                s.save()
                for _movement in _statement['transactions']:
                    # TODO :check if the movement is already imported.
                    if not Movement.objects.filter(
                            unique_import_id=_movement['unique_import_id']).exists():
                        _ref = _movement.get('ref', '') or ''
                        m = Movement(statement=s,
                                     unique_import_id=_movement['unique_import_id'],
                                     movement_date=_movement['date'],
                                     amount=_movement['amount'],
                                     partner_name=_movement.remote_owner,
                                     ref=_ref,
                                     bank_account=_movement.remote_account)
                        m.save()
                num += 1

        except ValueError:
            dd.logger.info("Statement file was not a camt file.")

        if len(failed_statements) == 0:
            msg = "Imported {0} statemenmts from file {1}.".format(num, filename)
            dd.logger.info(msg)
            ar.info(msg)
        else:
            count_statements_imported = num - len(failed_statements)
            msg = "Imported {0} statemenmts from file {1} with {2} statements failed.".format(count_statements_imported,
                                                                                              filename,
                                                                                              len(failed_statements))
            dd.logger.info(msg)
            for i, _statement in enumerate(failed_statements):
                msg_error = "Statement number {0} failed to be imported : {1}.".format(i + 1, _statement)<commit_msg>Fixing bug in import_file function. #505<commit_after># -*- coding: UTF-8 -*-
# Copyright 2014-2015 <NAME>
# This file is part of Lino Cosi.
#
# Lino Cosi is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# Lino Cosi is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public
# License along with Lino Cosi.  If not, see
# <http://www.gnu.org/licenses/>.


"""
Database models for `lino_cosi.lib.sepa`.

"""

from __future__ import unicode_literals
import logging
from pprint import pformat
import glob
import os
from django.db import models
from django.core.exceptions import MultipleObjectsReturned
from lino.api import dd, _, rt
from lino.core.utils import ChangeWatcher
from .camt import CamtParser
from .fields import IBANField, BICField
from .utils import belgian_nban_to_iban_bic, iban2bic
import time

logger = logging.getLogger(__name__)


class ImportStatements(dd.Action):
    """Import the .xml files found in the directory specified at
    :attr:`import_statements_path
    <lino_cosi.lib.sepa.Plugin.import_statements_path>`.

    End-users invoke this via the menu command :menuselection:`SEPA
    --> Import SEPA`.

    When a file has been successfully imported, Lino deletes it.

    It might happen that an .xml file accidentally gets downloaded a
    second time. Lino does not create these statements again.

    """
    label = _("Import SEPA")
    http_method = 'POST'
    select_rows = False

    def get_view_permission(self, profile):
        """Make it invisible when :attr:`import_statements_path
        <lino_cosi.lib.sepa.Plugin.import_statements_path>` is empty.

        """
        if not dd.plugins.sepa.import_statements_path:
            return False
        return super(ImportStatements, self).get_view_permission(profile)

    def run_from_ui(self, ar):
        pth = dd.plugins.sepa.import_statements_path
        wc = os.path.join(pth, '*.xml')
        dd.logger.info("Importing SEPA statements from %s...", wc)
        count = 0
        if pth:
            for filename in glob.iglob(wc):
                self.import_file(ar, filename)
                count += 1
            msg = "{0} xml files would have been imported.".format(count)
            dd.logger.info(msg)
            return ar.success(msg, alert=_("Success"))
        msg = "No import_statements_path configured."
        return ar.error(msg, alert=_("Error"))

    def import_file(self, ar, filename):
        Account = rt.modules.sepa.Account

        # Parse a CAMT053 XML file.
        parser = CamtParser()
        data_file = open(filename, 'rb').read()
        num = 0
        failed_statements = {}
        try:
            dd.logger.info("Parsing %s with camt.", filename)
            res = parser.parse(data_file)
            if res is None:
                raise Exception("res is None")
            for _statement in res:
                iban = _statement['account_number']
                if iban is None:
                    # msg = "Statement without account number : {0}"
                    failed_statements[num] = "IBAN Not found"
                    continue
                    # raise Exception(msg.format(pformat(_statement)))
                try:
                    account = Account.objects.get(iban=iban)
                except Account.DoesNotExist:
                    account = Account(iban=iban)
                    account.full_clean()
                    account.save()
                except MultipleObjectsReturned:
                    msg = "Found more than one account with IBAN {0}"
                    raise Exception(msg.format(iban))
                s = Statement(account=account,
                              date=_statement['date'].strftime("%Y-%m-%d"),
                              date_done=time.strftime("%Y-%m-%d"),
                              statement_number=_statement['name'],
                              balance_end=_statement['balance_end'],
                              balance_start=_statement['balance_start'],
                              balance_end_real=_statement['balance_end_real'],
                              currency_code=_statement['currency_code'])
                s.save()
                for _movement in _statement['transactions']:
                    # TODO :check if the movement is already imported.
                    if not Movement.objects.filter(
                            unique_import_id=_movement['unique_import_id']).exists():
                        _ref = _movement.get('ref', '') or ''
                        if _movement.remote_account:
                            try:
                                _bank_account = Account.objects.get(iban=_movement.remote_account)
                            except Account.DoesNotExist:
                                _bank_account = Account(iban=_movement.remote_account)
                                _bank_account.full_clean()
                                _bank_account.save()
                            m = Movement(statement=s,
                                         unique_import_id=_movement['unique_import_id'],
                                         movement_date=_movement['date'],
                                         amount=_movement['amount'],
                                         partner_name=_movement.remote_owner,
                                         ref=_ref,
                                         bank_account=_bank_account)
                            m.save()
                num += 1

        except ValueError:
            dd.logger.info("Statement file was not a camt file.")

        if len(failed_statements) == 0:
            msg = "Imported {0} statemenmts from file {1}.".format(num, filename)
            dd.logger.info(msg)
            ar.info(msg)
        else:
            count_statements_imported = num - len(failed_statements)
            msg = "Imported {0} statemenmts from file {1} with {2} statements failed.".format(count_statements_imported,
                                                                                              filename,
                                                                                              len(failed_statements))
            dd.logger.info(msg)
            for i, _statement in enumerate(failed_statements):
                msg_error = "Statement number {0} failed to be imported : {1}.".format(i + 1, failed_statements[_statement])
{'fraction_non_alphanumeric': 0.08499315797476054, 'fraction_numerical': 0.004789417667629618, 'mean_word_length': 1.720785935884178, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 8}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 36, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6932267', 'n_tokens_mistral': 3422, 'n_tokens_neox': 3230, 'n_words': 953}