ofxstatement-n26: Add implementation of N26 ofx parser

This supports only the Italian N26 reports yet, but categories can easily
be added for other languages by updating the mapping dictionary

.gitignore

@@ -0,0 +1,12 @@
+*.pyc
+*.swp
+develop-eggs
+dist
+src/*.egg-info
+tags
+build
+eggs
+.venv
+.project
+.pydevproject
+*.sublime*

MANIFEST.in

@@ -0,0 +1 @@
+include README.md

Makefile

@@ -0,0 +1,17 @@
+all: test mypy black
+
+PHONY: test
+test:
+	pytest
+
+PHONY: coverage
+coverage: bin/pytest
+	pytest --cov src/ofxstatement
+
+.PHONY: black
+black:
+	black setup.py src tests
+
+.PHONY: mypy
+mypy:
+	mypy src tests

README.md

@@ -0,0 +1,25 @@
+# N26 Plugin for [ofxstatement](https://github.com/kedder/ofxstatement/)
+
+Parses N26 csv statement files to be used with GNU Cash or HomeBank.
+
+It only supports categories the Italian statements yet, but all languages can be
+easily supported by adding localized strings to the mapping dictionary.
+
+So, contributions are welcome!
+
+## Installation
+
+You can install the plugin as usual from pip or directly from the downloaded git
+
+### `pip`
+
+    pip3 install --user ofxstatement-n26
+
+### `setup.py`
+
+    python3 setup.py install --user
+
+## Usage
+Download your transactions file from the official bank's site and then run
+
+    ofxstatement convert -t n26 n26-csv-transactions.csv n26.ofx

setup.cfg

@@ -0,0 +1,2 @@
+[mypy]
+namespace_packages=True

setup.py

@@ -0,0 +1,44 @@
+#!/usr/bin/python3
+"""Setup
+"""
+from setuptools import find_packages
+from distutils.core import setup
+
+version = "0.1"
+
+with open("README.md") as f:
+    long_description = f.read()
+
+setup(
+    name="ofxstatement-n26",
+    version=version,
+    author="Marco Trevisan",
+    author_email="mail@3v1n0.net",
+    url="https://github.com/3v1n0/ofxstatement-n26",
+    description=("N26 plugin for ofxstatement"),
+    long_description=long_description,
+    long_description_content_type="text/markdown",
+    license="GPLv3",
+    keywords=["ofx", "banking", "statement", "n26", "che-banca"],
+    classifiers=[
+        "Development Status :: 3 - Alpha",
+        "Programming Language :: Python :: 3",
+        "Natural Language :: English",
+        "Topic :: Office/Business :: Financial :: Accounting",
+        "Topic :: Utilities",
+        "Environment :: Console",
+        "Operating System :: OS Independent",
+        "License :: OSI Approved :: GNU General Public License v3 (GPLv3)",
+    ],
+    packages=find_packages("src"),
+    package_dir={"": "src"},
+    namespace_packages=["ofxstatement", "ofxstatement.plugins"],
+    entry_points={
+        "ofxstatement": [
+            "n26 = ofxstatement.plugins.n26:N26Plugin",
+        ],
+    },
+    install_requires=["ofxstatement"],
+    include_package_data=True,
+    zip_safe=True,
+)

src/ofxstatement/__init__.py

@@ -0,0 +1 @@
+__import__("pkg_resources").declare_namespace(__name__)

src/ofxstatement/plugins/__init__.py

@@ -0,0 +1 @@
+__import__("pkg_resources").declare_namespace(__name__)

src/ofxstatement/plugins/n26.py

@@ -0,0 +1,113 @@
+import logging
+
+from decimal import Decimal
+from enum import Enum
+from typing import Any, Iterable, List, Optional
+
+from ofxstatement.plugin import Plugin
+from ofxstatement.parser import CsvStatementParser
+from ofxstatement.statement import (
+    BankAccount,
+    Currency,
+    Statement,
+    StatementLine,
+    generate_transaction_id,
+    recalculate_balance,
+)
+
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger("n26")
+
+TYPE_MAPPING = {
+    "Entrata": "XFER",
+    "Pagamento MasterCard": "POS",
+    "Trasferimento in uscita": "XFER",
+    "N26 sponsorizzazione": "CREDIT",
+    "MoneyBeam": "XFER",
+
+    # Add translations for other statements
+}
+
+
+class N26Parser(CsvStatementParser):
+    date_format = "%Y-%m-%d"
+
+    mappings = {
+        "date": 0,
+        "payee": 1,
+        "account_number": 2,
+        "trntype": 3,
+        "memo": 4,
+        "amount": 5,
+        "orig_amount": 6,
+        "orig_currency": 7,
+        "exchange_rate":8,
+    }
+
+    def parse(self):
+        statement = super().parse()
+        recalculate_balance(statement)
+        return statement
+
+    def split_records(self):
+        return [r for r in super().split_records()][1:]
+
+    def strip_spaces(self, string: str) -> str:
+        return " ".join(string.strip().split())
+
+    def parse_value(self, value: Optional[str], field: str) -> Any:
+        if field == "trntype":
+            native_type = value.split(" - ", 1)[0].strip()
+            trntype = TYPE_MAPPING.get(native_type)
+
+            if not trntype:
+                logger.warning(f"Mapping not found for {value}")
+                return "OTHER"
+
+            return trntype
+
+        elif field == "orig_currency":
+            return Currency(symbol=value)
+
+        elif field == "memo":
+            return self.strip_spaces(value)
+
+        elif field == "payee":
+            return self.strip_spaces(value)
+
+        return super().parse_value(value, field)
+
+    def parse_record(self, line: List[str]) -> Optional[StatementLine]:
+        stmt_line = super().parse_record(line)
+
+        if stmt_line.payee == 'N26 Bank' and 'mark-up fee' in stmt_line.memo:
+            stmt_line.trntype = "FEE"
+        elif stmt_line.payee == 'N26' and 'N26' in stmt_line.memo:
+            stmt_line.trntype = "FEE"
+
+        if not stmt_line.memo or stmt_line.memo == "-":
+            stmt_line.memo = stmt_line.payee
+
+        account_number = line[self.mappings["account_number"]]
+        if account_number:
+            if stmt_line.payee:
+                stmt_line.payee += f' ({account_number})'
+            else:
+                stmt_line.payee = account_number
+
+        if stmt_line.orig_currency:
+            exchange_rate = line[self.mappings["exchange_rate"]]
+            if exchange_rate:
+                stmt_line.orig_currency.rate = self.parse_float(exchange_rate)
+
+        stmt_line.id = generate_transaction_id(stmt_line)
+        logger.debug(stmt_line)
+        return stmt_line
+
+
+class N26Plugin(Plugin):
+    """N26 parser"""
+
+    def get_parser(self, filename: str) -> N26Parser:
+        file = open(filename, "r", encoding='utf-8')
+        return N26Parser(file)

tests/test_chebanca.py

@@ -0,0 +1,16 @@
+import os
+
+from ofxstatement.ui import UI
+
+from ofxstatement.plugins.n26 import N26Plugin
+
+
+def test_n26() -> None:
+    plugin = N26Plugin(UI(), {})
+    here = os.path.dirname(__file__)
+    sample_filename = os.path.join(here, "sample-statement.csv")
+
+    parser = plugin.get_parser(sample_filename)
+    statement = parser.parse()
+
+    assert statement is not None