diff --git a/app/apps/import_app/schemas/v1.py b/app/apps/import_app/schemas/v1.py index ef79c197..9d716c46 100644 --- a/app/apps/import_app/schemas/v1.py +++ b/app/apps/import_app/schemas/v1.py @@ -75,6 +75,20 @@ class SubtractTransformationRule(BaseModel): ) +class LineTransformationRule(BaseModel): + type: Literal["line"] + line: int = Field( + ..., + ge=1, + description="Line of the file to read, counted from 1 at the top of the file, including skipped lines", + ) + column: int | None = Field( + default=None, + ge=1, + description="Column of that line to read, counted from 1. Empty to read the whole line (Excel cells are joined with commas).", + ) + + class CSVImportSettings(BaseModel): skip_errors: bool = Field( default=False, @@ -133,6 +147,7 @@ class ColumnMapping(BaseModel): | SplitTransformationRule | AddTransformationRule | SubtractTransformationRule + | LineTransformationRule ] ] = Field(default_factory=list) diff --git a/app/apps/import_app/services/v1.py b/app/apps/import_app/services/v1.py index ba2a6d48..38cdb571 100644 --- a/app/apps/import_app/services/v1.py +++ b/app/apps/import_app/services/v1.py @@ -53,6 +53,15 @@ class ImportService: ) self.mapping: Dict[str, version_1.ColumnMapping] = self.config.mapping + # Lines read by `line` transformations, captured as (text, cells) while the file is read + self._wanted_lines = { + transform.line + for mapping in self.mapping.values() + for transform in mapping.transformations or [] + if transform.type == "line" + } + self._file_lines: Dict[int, tuple[str, list[str]]] = {} + # Ensure temp directory exists os.makedirs(settings.IMPORT_TEMP_ROOT, exist_ok=True) @@ -189,6 +198,15 @@ class ImportService: transform.pattern, transform.replacement, transformed ) + elif transform.type == "line": + text, cells = self._file_lines.get(transform.line, ("", [])) + if transform.column is None: + transformed = text + elif transform.column <= len(cells): + transformed = cells[transform.column - 1] + else: + transformed = "" + elif transform.type == "date_format": transformed = datetime.strptime( transformed, transform.original_format @@ -749,14 +767,31 @@ class ImportService: logger.error(f"Fatal error processing row {row_number}", exc_info=e) + def _capture_csv_lines(self, csv_file): + """Pass the file's lines through, keeping the ones `line` transformations read.""" + for number, line in enumerate(csv_file, start=1): + if number in self._wanted_lines: + text = line.rstrip("\r\n") + cells = next(csv.reader([text], delimiter=self.settings.delimiter), []) + self._file_lines[number] = (text, cells) + yield line + + def _capture_excel_lines(self, read_row): + """Keep the sheet rows `line` transformations read. read_row(n) returns row n's values.""" + self._file_lines = {} + for number in self._wanted_lines: + cells = ["" if value is None else str(value) for value in read_row(number)] + self._file_lines[number] = (",".join(cells), cells) + def _process_csv(self, file_path): - # First pass: count rows + # First pass: count rows, and capture the lines `line` transformations read with open(file_path, "r", encoding=self.settings.encoding) as csv_file: + lines = self._capture_csv_lines(csv_file) # Skip specified number of rows for _ in range(self.settings.skip_lines): - next(csv_file) + next(lines) - reader = csv.DictReader(csv_file, delimiter=self.settings.delimiter) + reader = csv.DictReader(lines, delimiter=self.settings.delimiter) self._update_totals("total", value=sum(1 for _ in reader)) with open(file_path, "r", encoding=self.settings.encoding) as csv_file: @@ -808,6 +843,14 @@ class ImportService: sheet = workbook[sheet_name] self._log("info", f"Processing sheet: {sheet_name}") + self._capture_excel_lines( + lambda number: next( + sheet.iter_rows( + min_row=number, max_row=number, values_only=True + ), + (), + ) + ) headers = [ str(cell.value or "") for cell in sheet[self.settings.start_row] ] @@ -868,6 +911,13 @@ class ImportService: continue sheet = workbook.sheet_by_name(sheet_name) self._log("info", f"Processing sheet: {sheet_name}") + self._capture_excel_lines( + lambda number: ( + sheet.row_values(number - 1) + if number <= sheet.nrows + else [] + ) + ) headers = [ str(sheet.cell_value(self.settings.start_row - 1, col) or "") for col in range(sheet.ncols) diff --git a/app/apps/import_app/tests/test_line_transformation.py b/app/apps/import_app/tests/test_line_transformation.py new file mode 100644 index 00000000..6a5d22c5 --- /dev/null +++ b/app/apps/import_app/tests/test_line_transformation.py @@ -0,0 +1,167 @@ +"""Tests for the `line` transformation, which reads a line outside the table, like a header cell.""" + +import shutil +import tempfile +from pathlib import Path + +import openpyxl +from django.contrib.auth import get_user_model +from django.test import TestCase, override_settings + +from apps.accounts.models import Account +from apps.common.middleware.thread_local import delete_current_user, write_current_user +from apps.currencies.models import Currency +from apps.import_app.models import ImportProfile, ImportRun +from apps.import_app.services.v1 import ImportService +from apps.transactions.models import Transaction + +MAPPING = """ +mapping: + date: + target: date + source: date + format: "%Y-%m-%d" + amount: + target: amount + source: amount + type: + target: type + source: amount + detection_method: sign + description: + target: description + source: description + account: + target: account + transformations: +{account_transformations} +""" + +# The account name sits in the header, above the table +CSV_CONTENT = """Bank statement +Account:;Savings +Account number:;Savings 1234 + +date;amount;description +2026-01-05;-50.00;Groceries +2026-01-06;1000.00;Salary +""" + + +class LineTransformationTests(TestCase): + def setUp(self): + self.temp_root = Path(tempfile.mkdtemp()).resolve() + self.addCleanup(shutil.rmtree, self.temp_root, ignore_errors=True) + temp_root_override = override_settings(IMPORT_TEMP_ROOT=self.temp_root) + temp_root_override.enable() + self.addCleanup(temp_root_override.disable) + + self.user = get_user_model().objects.create_user( + email="test@example.com", password="password" + ) + write_current_user(self.user) + self.addCleanup(delete_current_user) + + currency = Currency.objects.create( + code="EUR", name="Euro", decimal_places=2, prefix="€ " + ) + Account.objects.create(name="Checking", currency=currency, owner=self.user) + self.savings = Account.objects.create( + name="Savings", currency=currency, owner=self.user + ) + + def _import(self, settings, account_transformations, file_name, write): + yaml_config = settings + MAPPING.format( + account_transformations=account_transformations + ) + profile = ImportProfile.objects.create( + name="Profile", yaml_config=yaml_config, owner=self.user + ) + run = ImportRun.objects.create( + profile=profile, file_name=file_name, owner=self.user + ) + file_path = self.temp_root / file_name + write(file_path) + + ImportService(run).process_file(str(file_path)) + + run.refresh_from_db() + return run + + def _import_csv(self, account_transformations): + settings = """ +settings: + file_type: csv + importing: transactions + delimiter: ";" + skip_lines: 4 + trigger_transaction_rules: false +""" + return self._import( + settings, + account_transformations, + "statement.csv", + lambda path: path.write_text(CSV_CONTENT, encoding="utf-8"), + ) + + def assert_imported_into_savings(self, run): + self.assertEqual(run.status, ImportRun.Status.FINISHED, run.logs) + self.assertEqual(run.successful_rows, 2) + self.assertEqual(Transaction.objects.filter(account=self.savings).count(), 2) + + def test_csv_reads_a_cell_from_a_skipped_line(self): + run = self._import_csv( + """ + - type: line + line: 2 + column: 2 +""" + ) + + self.assert_imported_into_savings(run) + + def test_csv_whole_line_cleaned_with_regex(self): + run = self._import_csv( + """ + - type: line + line: 3 + - type: regex + pattern: '^Account number:;(\\w+) \\d+$' + replacement: '\\1' +""" + ) + + self.assert_imported_into_savings(run) + + def test_xlsx_reads_a_cell_above_the_header(self): + def write(path): + workbook = openpyxl.Workbook() + sheet = workbook.active + sheet["A1"] = "Bank statement" + sheet["A4"] = "Account:" + sheet["B4"] = "Savings" + sheet.append([]) + sheet.append(["date", "amount", "description"]) + sheet.append(["2026-01-05", "-50.00", "Groceries"]) + sheet.append(["2026-01-06", "1000.00", "Salary"]) + workbook.save(path) + + settings = """ +settings: + file_type: xlsx + importing: transactions + start_row: 6 + trigger_transaction_rules: false +""" + run = self._import( + settings, + """ + - type: line + line: 4 + column: 2 +""", + "statement.xlsx", + write, + ) + + self.assert_imported_into_savings(run)