[FIX] account_statement_import_sheet_file

Allow to import csv files with sheet mapping being configured with no_header and some header_lines_skip_count

Author: mondot

account_statement_import_sheet_file/models/account_statement_import_sheet_mapping.py

@@ -163,14 +163,16 @@ class AccountStatementImportSheetMapping(models.Model):
     )
     footer_lines_skip_count = fields.Integer(
         string="Footer lines skip count",
-        help="Set the Footer lines number."
-        "Used in some csv/xlsx file that integrate meta data in"
-        "last lines.",
+        help="Define the number of footer lines to be skipped in CSV/XLSX"
+        "files containing metadata at the bottom.",
         default="0",
     )
     header_lines_skip_count = fields.Integer(
         string="Header lines skip count",
-        help="Set the Header lines number.",
+        help="Define the number of header lines to be skipped in CSV/XLSX"
+        "files containing metadata at the top."
+        "The header line containing the titles of columns should "
+        "be counted",
         default="0",
     )
     skip_empty_lines = fields.Boolean(

account_statement_import_sheet_file/models/account_statement_import_sheet_parser.py

@@ -39,16 +39,16 @@ class AccountStatementImportSheetParser(models.TransientModel):
     _description = "Bank Statement Import Sheet Parser"
 
     @api.model
-    def parse_header(self, csv_or_xlsx, mapping):
+    def parse_header(self, csv_or_xlsx, mapping, data_file_is_xlsx):
         if mapping.no_header:
             return []
         header_line = mapping.header_lines_skip_count
         # prevent negative indexes
         if header_line > 0:
             header_line -= 1
-        if isinstance(csv_or_xlsx, tuple):
+        if data_file_is_xlsx:
             header = [
-                str(value).strip() for value in csv_or_xlsx[1].row_values(header_line)
+                str(value).strip() for value in csv_or_xlsx.row_values(header_line)
             ]
         else:
             [next(csv_or_xlsx) for _i in range(header_line)]
@@ -149,6 +149,7 @@ def _get_column_names(self):
     def _parse_lines(self, mapping, data_file, currency_code):
         columns = dict()
         try:
+            data_file_is_xlsx = True
             workbook = xlrd.open_workbook(
                 file_contents=data_file,
                 encoding_override=(
@@ -160,6 +161,7 @@ def _parse_lines(self, mapping, data_file, currency_code):
                 workbook.sheet_by_index(0),
             )
         except xlrd.XLRDError:
+            data_file_is_xlsx = False
             csv_options = {}
             csv_delimiter = mapping._get_column_delimiter_character()
             if csv_delimiter:
@@ -176,8 +178,13 @@ def _parse_lines(self, mapping, data_file, currency_code):
                         _("No valid encoding was found for the attached file")
                     ) from None
                 decoded_file = data_file.decode(detected_encoding)
-            csv_or_xlsx = reader(StringIO(decoded_file), **csv_options)
-        header = self.parse_header(csv_or_xlsx, mapping)
+            csv = reader(StringIO(decoded_file), **csv_options)
+            csv_or_xlsx, csv_iterator_copy = itertools.tee(csv)
+
+        if data_file_is_xlsx:
+            header = self.parse_header(workbook.sheet_by_index(0), mapping, data_file_is_xlsx)
+        else:
+            header = self.parse_header(csv_iterator_copy, mapping, data_file_is_xlsx)
 
         # NOTE no seria necesario debit_column y credit_column ya que tenemos los
         # respectivos campos related
@@ -212,16 +219,17 @@ def _parse_rows(self, mapping, currency_code, data, columns):  # noqa: C901
         else:
             numrows = len(str(data_file.strip()).split("\\n"))
 
-        label_line = mapping.header_lines_skip_count
-        footer_line = numrows - mapping.footer_lines_skip_count
+        data_first_index_line = mapping.header_lines_skip_count
+        data_last_line = numrows - mapping.footer_lines_skip_count
+        data_line_count = data_last_line - data_first_index_line
 
         if isinstance(csv_or_xlsx, tuple):
-            rows = range(label_line, footer_line)
+            rows = range(data_first_index_line, data_last_line)
         else:
             rows = csv_or_xlsx
 
         lines = []
-        for index, row in enumerate(rows, label_line):
+        for index, row in enumerate(rows):
             if isinstance(csv_or_xlsx, tuple):
                 book = csv_or_xlsx[0]
                 sheet = csv_or_xlsx[1]
@@ -233,7 +241,9 @@ def _parse_rows(self, mapping, currency_code, data, columns):  # noqa: C901
                         cell_value = xldate_as_datetime(cell_value, book.datemode)
                     values.append(cell_value)
             else:
-                if index >= footer_line:
+                if index < data_first_index_line:
+                    continue
+                if index >= data_first_index_line + data_line_count:
                     continue
                 values = list(row)
             if mapping.skip_empty_lines and not any(values):

account_statement_import_sheet_file/tests/fixtures/meta_data_separated_credit_debit_no_header.csv

@@ -0,0 +1,9 @@
+Bank code : 1001010101,Agency Code : 10000,Download start date : 01/04/2020,Download end date : 02/04/2020,,
+Account Number : 08088804068,Account Name : Account Owner,: EUR,,,
+,,,,,
+Balance at end of period,,,,"+31070,11",
+01/04/20,UNIQUE OP 1,LABEL 1,"-50,00",,DETAILS 1
+01/04/20,UNIQUE OP 2,LABEL 2,"-100,00",,CLIENTS X
+02/04/20,UNIQUE OP 3,LABEL 3,"-80,68",,DETAILS 2
+02/04/20,UNIQUE OP 4,LABEL 4,,"1300,00",DETAILS 3
+Balance at start of period,,,,"+30000,77",

account_statement_import_sheet_file/tests/fixtures/meta_data_separated_credit_debit_no_header.xlsx

@@ -204,13 +204,12 @@ def test_original_currency(self):
         line_amount_currency = float_round(line.amount_currency, precision_digits=1)
         self.assertEqual(line_amount_currency, 1000.0)
 
-    def test_original_currency_no_header(self):
+    def test_original_currency_no_header_csv(self):
         no_header_statement_map = self.AccountStatementImportSheetMapping.create(
             {
                 "name": "Sample Statement",
                 "float_thousands_sep": "comma",
                 "float_decimal_sep": "dot",
-                "header_lines_skip_count": 0,
                 "delimiter": "comma",
                 "quotechar": '"',
                 "timestamp_format": "%m/%d/%Y",
@@ -236,7 +235,57 @@ def test_original_currency_no_header(self):
         data = self._data_file("fixtures/original_currency_no_header.csv", "utf-8")
         wizard = self.AccountStatementImport.with_context(journal_id=journal.id).create(
             {
-                "statement_filename": "fixtures/original_currency.csv",
+                "statement_filename": "fixtures/original_currency_no_header.csv",
+                "statement_file": data,
+                "sheet_mapping_id": no_header_statement_map.id,
+            }
+        )
+        wizard.with_context(
+            account_statement_import_sheet_file_test=True
+        ).import_file_button()
+        statement = self.AccountBankStatement.search([("journal_id", "=", journal.id)])
+        self.assertEqual(len(statement), 1)
+        self.assertEqual(len(statement.line_ids), 1)
+
+        line = statement.line_ids
+        self.assertEqual(line.currency_id, self.currency_usd)
+        self.assertEqual(line.foreign_currency_id, self.currency_eur)
+        self.assertEqual(line.amount_currency, 1000.0)
+        self.assertEqual(line.payment_ref, "Your payment INV0001")
+
+    def test_original_currency_no_header_xlsx(self):
+        no_header_statement_map = self.AccountStatementImportSheetMapping.create(
+            {
+                "name": "Sample Statement",
+                "float_thousands_sep": "comma",
+                "float_decimal_sep": "dot",
+                "delimiter": "comma",
+                "quotechar": '"',
+                "timestamp_format": "%m/%d/%Y",
+                "no_header": True,
+                "timestamp_column": "0",
+                "amount_column": "3",
+                "original_currency_column": "2",
+                "original_amount_column": "4",
+                "description_column": "1,7",
+                "partner_name_column": "5",
+                "bank_account_column": "6",
+            }
+        )
+        journal = self.AccountJournal.create(
+            {
+                "name": "Bank",
+                "type": "bank",
+                "code": "BANK",
+                "currency_id": self.currency_usd.id,
+                "suspense_account_id": self.suspense_account.id,
+            }
+        )
+
+        data = self._data_file("fixtures/original_currency_no_header.xlsx")
+        wizard = self.AccountStatementImport.with_context(journal_id=journal.id).create(
+            {
+                "statement_filename": "fixtures/original_currency_no_header.xlsx",
                 "statement_file": data,
                 "sheet_mapping_id": no_header_statement_map.id,
             }
@@ -478,6 +527,58 @@ def test_metadata_separated_debit_credit_csv(self):
         self.assertEqual(line1.amount, 50)
         self.assertEqual(line4.amount, -1300)
 
+    def test_metadata_separated_debit_credit_no_header_csv(self):
+        journal = self.AccountJournal.create(
+            {
+                "name": "Bank",
+                "type": "bank",
+                "code": "BANK",
+                "currency_id": self.currency_usd.id,
+                "suspense_account_id": self.suspense_account.id,
+            }
+        )
+        statement_map = self.sample_statement_map.copy(
+            {
+                "footer_lines_skip_count": 1,
+                "header_lines_skip_count": 4,
+                "no_header": True,
+                "amount_column": None,
+                "partner_name_column": None,
+                "bank_account_column": None,
+                "float_thousands_sep": "none",
+                "float_decimal_sep": "comma",
+                "timestamp_format": "%m/%d/%y",
+                "timestamp_column": "0",
+                "description_column": "2",
+                "original_currency_column": None,
+                "original_amount_column": None,
+                "amount_type": "distinct_credit_debit",
+                "amount_debit_column": "3",
+                "amount_credit_column": "4",
+            }
+        )
+        data = self._data_file(
+            "fixtures/meta_data_separated_credit_debit_no_header.csv", "utf-8"
+        )
+        wizard = self.AccountStatementImport.with_context(journal_id=journal.id).create(
+            {
+                "statement_filename": "fixtures/meta_data_separated_credit_debit_no_header.csv",
+                "statement_file": data,
+                "sheet_mapping_id": statement_map.id,
+            }
+        )
+        wizard.with_context(
+            journal_id=journal.id,
+            account_bank_statement_import_txt_xlsx_test=True,
+        ).import_file_button()
+        statement = self.AccountBankStatement.search([("journal_id", "=", journal.id)])
+        self.assertEqual(len(statement), 1)
+        self.assertEqual(len(statement.line_ids), 4)
+        line1 = statement.line_ids.filtered(lambda x: x.payment_ref == "LABEL 1")
+        line4 = statement.line_ids.filtered(lambda x: x.payment_ref == "LABEL 4")
+        self.assertEqual(line1.amount, 50)
+        self.assertEqual(line4.amount, -1300)
+
     def test_metadata_separated_debit_credit_xlsx(self):
         journal = self.AccountJournal.create(
             {
@@ -525,6 +626,59 @@ def test_metadata_separated_debit_credit_xlsx(self):
         self.assertEqual(line1.amount, 50)
         self.assertEqual(line4.amount, -1300)
 
+    def test_metadata_separated_debit_credit_no_header_xlsx(self):
+        journal = self.AccountJournal.create(
+            {
+                "name": "Bank",
+                "type": "bank",
+                "code": "BANK",
+                "currency_id": self.currency_usd.id,
+                "suspense_account_id": self.suspense_account.id,
+            }
+        )
+        statement_map = self.sample_statement_map.copy(
+            {
+                "footer_lines_skip_count": 1,
+                "header_lines_skip_count": 4,
+                "no_header": True,
+                "amount_column": None,
+                "partner_name_column": None,
+                "bank_account_column": None,
+                "float_thousands_sep": "none",
+                "float_decimal_sep": "comma",
+                "timestamp_format": "%m/%d/%y",
+                "timestamp_column": "0",
+                "description_column": "2",
+                "original_currency_column": None,
+                "original_amount_column": None,
+                "amount_type": "distinct_credit_debit",
+                "amount_debit_column": "3",
+                "amount_credit_column": "4",
+            }
+        )
+        filename = "fixtures/meta_data_separated_credit_debit_no_header.xlsx"
+        data = self._data_file(
+            "fixtures/meta_data_separated_credit_debit_no_header.xlsx"
+        )
+        wizard = self.AccountStatementImport.with_context(journal_id=journal.id).create(
+            {
+                "statement_filename": filename,
+                "statement_file": data,
+                "sheet_mapping_id": statement_map.id,
+            }
+        )
+        wizard.with_context(
+            journal_id=journal.id,
+            account_bank_statement_import_txt_xlsx_test=True,
+        ).import_file_button()
+        statement = self.AccountBankStatement.search([("journal_id", "=", journal.id)])
+        self.assertEqual(len(statement), 1)
+        self.assertEqual(len(statement.line_ids), 4)
+        line1 = statement.line_ids.filtered(lambda x: x.payment_ref == "LABEL 1")
+        line4 = statement.line_ids.filtered(lambda x: x.payment_ref == "LABEL 4")
+        self.assertEqual(line1.amount, 50)
+        self.assertEqual(line4.amount, -1300)
+
     def test_amount_inverse_sign(self):
         self.sample_statement_map.amount_inverse_sign = True
         journal = self.AccountJournal.create(
@@ -701,17 +855,6 @@ def test_offsets(self):
         )
         file_name = "fixtures/sample_statement_offsets.xlsx"
         data = self._data_file(file_name)
-        wizard = self.AccountStatementImport.with_context(journal_id=journal.id).create(
-            {
-                "statement_filename": file_name,
-                "statement_file": data,
-                "sheet_mapping_id": self.sample_statement_map.id,
-            }
-        )
-        with self.assertRaises(UserError):
-            wizard.with_context(
-                account_statement_import_txt_xlsx_test=True
-            ).import_file_button()
         statement_map_offsets = self.sample_statement_map.copy(
             {
                 "offset_column": 1,
@@ -748,17 +891,6 @@ def test_skip_empty_lines(self):
         file_name = "fixtures/empty_lines_statement.csv"
         data = self._data_file(file_name, "utf-8")
 
-        wizard = self.AccountStatementImport.with_context(journal_id=journal.id).create(
-            {
-                "statement_filename": file_name,
-                "statement_file": data,
-                "sheet_mapping_id": self.sample_statement_map.id,
-            }
-        )
-        with self.assertRaises(UserError):
-            wizard.with_context(
-                account_statement_import_txt_xlsx_test=True
-            ).import_file_button()
         statement_map_empty_line = self.sample_statement_map.copy(
             {
                 "skip_empty_lines": True,