feat(import): read any line of the file with a line transformation

Some banks name the account once in a header cell above the table, which
skip_lines and start_row throw away. A `line` transformation returns a line
of the file (or one of its columns) as text, so a mapping can read it and
clean it up with regex or replace.

CSV keeps only the requested lines during the existing counting pass; Excel
reads them straight from each sheet. An account picked on the run still
overrides the mapped one.

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)
 

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)

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)