gaugendre/ofx-processor
1
0
Fork 1
Code Issues Pull requests Releases Activity

Add LCL processor and extract common OFX logic

Browse source
This commit is contained in:
Gabriel Augendre 2020-03-31 18:32:54 +02:00
parent b8a3519f10
commit 33cd7bf8f6
17 changed files with 416 additions and 63 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
ofx_processor/processors/bpvf.py
View file

@ -1,23 +1,11 @@
import re
import sys
import click
from ofxtools import OFXTree
from ofxtools.header import OFXHeaderError
from ofx_processor.utils.processor import Processor, Line
from ofx_processor.utils.base_ofx import OfxBaseLine, OfxBaseProcessor
class BpvfLine(Line):
def __init__(self, data=None):
super(BpvfLine, self).__init__(data)
def get_date(self):
return self.data.dtposted.isoformat().split("T")[0]
def get_amount(self):
return int(self.data.trnamt * 1000)
class BpvfLine(OfxBaseLine):
def get_memo(self):
return self._process_name_and_memo(self.data.name, self.data.memo)[1]
@ -41,21 +29,10 @@ class BpvfLine(Line):
return name, memo
class BpvfProcessor(Processor):
class BpvfProcessor(OfxBaseProcessor):
line_class = BpvfLine
account_name = "bpvf"
def parse_file(self):
parser = OFXTree()
try:
parser.parse(self.filename)
except (FileNotFoundError, OFXHeaderError):
click.secho("Couldn't open or parse ofx file", fg="red")
sys.exit(1)
ofx = parser.convert()
return ofx.statements[0].transactions
@staticmethod
@click.command("bpvf")
@click.argument("ofx_filename")

12
ofx_processor/processors/ce.py
View file

@ -2,10 +2,16 @@ import re
import click
from ofx_processor.processors.bpvf import BpvfProcessor, BpvfLine
from ofx_processor.utils.base_ofx import OfxBaseProcessor, OfxBaseLine
class CeLine(BpvfLine):
class CeLine(OfxBaseLine):
def get_memo(self):
return self._process_name_and_memo(self.data.name, self.data.memo)[1]
def get_payee(self):
return self._process_name_and_memo(self.data.name, self.data.memo)[0]
@staticmethod
def _process_name_and_memo(name: str, memo: str):
name = name.strip()
@ -20,7 +26,7 @@ class CeLine(BpvfLine):
return res_name, res_memo
class CeProcessor(BpvfProcessor):
class CeProcessor(OfxBaseProcessor):
account_name = "ce"
line_class = CeLine

43
ofx_processor/processors/lcl.py Normal file
View file

@ -0,0 +1,43 @@
import sys
import click
from ofx_processor.utils.base_ofx import OfxBaseLine, OfxBaseProcessor
class LclLine(OfxBaseLine):
pass
class LclProcessor(OfxBaseProcessor):
line_class = LclLine
account_name = "lcl"
def parse_file(self):
# The first line of this file needs to be removed.
# It contains something that is not part of the header of an OFX file.
try:
with open(self.filename, "r") as user_file:
data = user_file.read().splitlines(True)
except FileNotFoundError:
click.secho("Couldn't find ofx file", fg="red")
sys.exit(1)
if "Content-Type:" in data[0]:
with open(self.filename, "w") as temp_file:
temp_file.writelines(data[1:])
transactions = super(LclProcessor, self).parse_file()
if "Content-Type:" in data[0]:
with open(self.filename, "w") as temp_file:
temp_file.writelines(data)
return transactions
@staticmethod
@click.command("lcl")
@click.argument("ofx_filename")
def main(ofx_filename):
"""Import LCL bank statement (OFX file)."""
LclProcessor(ofx_filename).push_to_ynab()

9
ofx_processor/processors/revolut.py
View file

@ -4,7 +4,7 @@ import sys
import click
import dateparser
from ofx_processor.utils.processor import Processor, Line
from ofx_processor.utils.base_processor import BaseProcessor, BaseLine
def _amount_str_to_float(amount):
@ -13,10 +13,7 @@ def _amount_str_to_float(amount):
return ""
class RevolutLine(Line):
def __init__(self, data: dict = None):
super(RevolutLine, self).__init__(data)
class RevolutLine(BaseLine):
def _process_inflow(self):
return _amount_str_to_float(self.data.get("Paid In (EUR)"))
@ -48,7 +45,7 @@ class RevolutLine(Line):
return self.data.get("Reference")
class RevolutProcessor(Processor):
class RevolutProcessor(BaseProcessor):
line_class = RevolutLine
account_name = "revolut"

36
ofx_processor/utils/base_ofx.py Normal file
View file

@ -0,0 +1,36 @@
import sys
import click
from ofxtools import OFXTree
from ofxtools.header import OFXHeaderError
from ofx_processor.utils.base_processor import BaseLine, BaseProcessor
class OfxBaseLine(BaseLine):
def get_date(self):
return self.data.dtposted.isoformat().split("T")[0]
def get_amount(self):
return int(self.data.trnamt * 1000)
def get_memo(self):
return self.data.memo
def get_payee(self):
return self.data.name
class OfxBaseProcessor(BaseProcessor):
line_class = OfxBaseLine
def parse_file(self):
parser = OFXTree()
try:
parser.parse(self.filename)
except (FileNotFoundError, OFXHeaderError):
click.secho("Couldn't open or parse ofx file", fg="red")
sys.exit(1)
ofx = parser.convert()
return ofx.statements[0].transactions

6
ofx_processor/utils/processor.py → ofx_processor/utils/base_processor.py
View file

@ -5,7 +5,7 @@ import click
from ofx_processor.utils import ynab
class Line:
class BaseLine:
data = None
def __init__(self, data=None):
@ -42,8 +42,8 @@ class Line:
return ynab_transaction
class Processor:
line_class = Line
class BaseProcessor:
line_class = BaseLine
account_name = None
def __init__(self, filename):

6
ofx_processor/utils/utils.py
View file

@ -23,7 +23,11 @@ def discover_processors(cli: click.Group):
for module in pkgutil.iter_modules(processors.__path__, prefix):
module = importlib.import_module(module.name)
for item in dir(module):
if item.endswith("Processor") and item != "Processor":
if (
item.endswith("Processor")
and item != "Processor"
and "Base" not in item
):
cls = getattr(module, item)
cli.add_command(cls.main)

3
tests/samples/config.ini
View file

@ -10,3 +10,6 @@ account = <YOUR REVOLUT ACCOUNT ID>
[ce]
account = <YOUR CE ACCOUNT ID>
[lcl]
account = <YOUR CE ACCOUNT ID>

60
tests/samples/lcl.ofx Normal file
View file

@ -0,0 +1,60 @@
OFXHEADER:100
DATA:OFXSGML
VERSION:102
SECURITY:NONE
ENCODING:USASCII
CHARSET:1252
COMPRESSION:NONE
OLDFILEUID:NONE
NEWFILEUID:NONE
<OFX>
<SIGNONMSGSRSV1>
<SONRS>
<STATUS>
<CODE>0
<SEVERITY>INFO
</STATUS>
<DTSERVER>20200331170000
<LANGUAGE>FRA
<DTPROFUP>20200331170000
<DTACCTUP>20200331170000
</SONRS>
</SIGNONMSGSRSV1>
<BANKMSGSRSV1>
<STMTTRNRS>
<TRNUID>00
<STATUS>
<CODE>0
<SEVERITY>INFO
</STATUS>
<STMTRS>
<CURDEF>EUR
<BANKACCTFROM>
<BANKID>30002
<BRANCHID>01047
<ACCTID>059235W
<ACCTTYPE>CHECKING
</BANKACCTFROM>
<BANKTRANLIST>
<DTSTART>20200214120000
<DTEND>20200330120000
<STMTTRN>
<TRNTYPE>SRVCHG
<DTPOSTED>20200312
<TRNAMT>+1000.00
<FITID>348 120320 100000
<NAME>VIR INST M PAYEE 1
</STMTTRN>
</BANKTRANLIST>
<LEDGERBAL>
<BALAMT>+1000.00
<DTASOF>20200330120000
</LEDGERBAL>
<AVAILBAL>
<BALAMT>+1000.00
<DTASOF>20200330120000
</AVAILBAL>
</STMTRS>
</STMTTRNRS>
</BANKMSGSRSV1>
</OFX>

61
tests/samples/lcl_as_downloaded.ofx Normal file
View file

@ -0,0 +1,61 @@
Content-Type: application/x-ofx
OFXHEADER:100
DATA:OFXSGML
VERSION:102
SECURITY:NONE
ENCODING:USASCII
CHARSET:1252
COMPRESSION:NONE
OLDFILEUID:NONE
NEWFILEUID:NONE
<OFX>
<SIGNONMSGSRSV1>
<SONRS>
<STATUS>
<CODE>0
<SEVERITY>INFO
</STATUS>
<DTSERVER>20200331170000
<LANGUAGE>FRA
<DTPROFUP>20200331170000
<DTACCTUP>20200331170000
</SONRS>
</SIGNONMSGSRSV1>
<BANKMSGSRSV1>
<STMTTRNRS>
<TRNUID>00
<STATUS>
<CODE>0
<SEVERITY>INFO
</STATUS>
<STMTRS>
<CURDEF>EUR
<BANKACCTFROM>
<BANKID>30002
<BRANCHID>01047
<ACCTID>059235W
<ACCTTYPE>CHECKING
</BANKACCTFROM>
<BANKTRANLIST>
<DTSTART>20200214120000
<DTEND>20200330120000
<STMTTRN>
<TRNTYPE>SRVCHG
<DTPOSTED>20200312
<TRNAMT>+1000.00
<FITID>348 120320 100000
<NAME>VIR INST M PAYEE 1
</STMTTRN>
</BANKTRANLIST>
<LEDGERBAL>
<BALAMT>+1000.00
<DTASOF>20200330120000
</LEDGERBAL>
<AVAILBAL>
<BALAMT>+1000.00
<DTASOF>20200330120000
</AVAILBAL>
</STMTRS>
</STMTTRNRS>
</BANKMSGSRSV1>
</OFX>

9
tests/samples/lcl_expected.json Normal file
View file

@ -0,0 +1,9 @@
[
{
"date": "2020-03-12",
"amount": 1000000,
"payee_name": "VIR INST M PAYEE 1",
"memo": null,
"import_id": "YNAB:1000000:2020-03-12:1"
}
]

52
tests/samples/lcl_malformed.ofx Normal file
View file

@ -0,0 +1,52 @@
OFXHEADER:100
DATA:OFXSGML
VERSION:102
SECURITY:NONE
ENCODING:USASCII
CHARSET:1252
<SEVERITY>INFO
</STATUS>
<DTSERVER>20200331170000
<LANGUAGE>FRA
<DTPROFUP>20200331170000
<DTACCTUP>20200331170000
</SONRS>
</SIGNONMSGSRSV1>
<BANKMSGSRSV1>
<STMTTRNRS>
<TRNUID>00
<STATUS>
<CODE>0
<SEVERITY>INFO
</STATUS>
<STMTRS>
<CURDEF>EUR
<BANKACCTFROM>
<BANKID>30002
<BRANCHID>01047
<ACCTID>059235W
<ACCTTYPE>CHECKING
</BANKACCTFROM>
<BANKTRANLIST>
<DTSTART>20200214120000
<DTEND>20200330120000
<STMTTRN>
<TRNTYPE>SRVCHG
<DTPOSTED>20200312
<TRNAMT>+1000.00
<FITID>348 120320 100000
<NAME>VIR INST M PAYEE 1
</STMTTRN>
</BANKTRANLIST>
<LEDGERBAL>
<BALAMT>+1000.00
<DTASOF>20200330120000
</LEDGERBAL>
<AVAILBAL>
<BALAMT>+1000.00
<DTASOF>20200330120000
</AVAILBAL>
</STMTRS>
</STMTTRNRS>
</BANKMSGSRSV1>
</OFX>

12
tests/samples/lcl_transactions.json Normal file
View file

@ -0,0 +1,12 @@
{
"transactions": [
{
"date": "2020-03-12",
"amount": 1000000,
"payee_name": "VIR INST M PAYEE 1",
"memo": null,
"import_id": "YNAB:1000000:2020-03-12:1",
"account_id": "<YOUR CE ACCOUNT ID>"
}
]
}

31
tests/test_bpvf_processor.py
View file

@ -3,31 +3,14 @@ import json
import unittest
from ofx_processor.processors.bpvf import BpvfLine, BpvfProcessor
class BpvfTransaction:
"""
Mimick what is retrieved via ofxtools when parsing the file
"""
def __init__(
self,
name: str = "",
memo: str = "",
dtposted: datetime.datetime = None,
trnamt: float = 0,
):
self.dtposted = dtposted
self.memo = memo
self.trnamt = trnamt
self.name = name
from tests.utils import OfxTransaction
class BpvfLineTestCase(unittest.TestCase):
def test_process_name_and_memo_no_change(self):
name = "business"
memo = "2020-01-17"
transaction = BpvfTransaction(name=name, memo=memo)
transaction = OfxTransaction(name=name, memo=memo)
line = BpvfLine(transaction)
result_name = line.get_payee()
@ -38,7 +21,7 @@ class BpvfLineTestCase(unittest.TestCase):
def test_process_name_and_memo_change_required_with_conversion(self):
name = "150120 CB****5874"
memo = "GUY AND SONS FR LYON 0,90EUR 1 EURO = 1,000000"
transaction = BpvfTransaction(name=name, memo=memo)
transaction = OfxTransaction(name=name, memo=memo)
expected_name = "GUY AND SONS FR LYON"
expected_memo = "150120 CB****5874 0,90EUR 1 EURO = 1,000000"
@ -52,7 +35,7 @@ class BpvfLineTestCase(unittest.TestCase):
def test_process_name_and_memo_change_required_no_conversion(self):
name = "150120 CB****5874"
memo = "Dott 75PARIS"
transaction = BpvfTransaction(name=name, memo=memo)
transaction = OfxTransaction(name=name, memo=memo)
expected_name = "Dott 75PARIS"
expected_memo = "150120 CB****5874"
@ -64,21 +47,21 @@ class BpvfLineTestCase(unittest.TestCase):
self.assertEqual(result_memo, expected_memo)
def test_get_date(self):
transaction = BpvfTransaction(dtposted=datetime.datetime(2020, 1, 23, 1, 2, 3))
transaction = OfxTransaction(dtposted=datetime.datetime(2020, 1, 23, 1, 2, 3))
expected_date = "2020-01-23"
result_date = BpvfLine(transaction).get_date()
self.assertEqual(result_date, expected_date)
def test_get_amount_positive(self):
transaction = BpvfTransaction(trnamt=52.2)
transaction = OfxTransaction(trnamt=52.2)
expected_amount = 52200
result_amount = BpvfLine(transaction).get_amount()
self.assertEqual(result_amount, expected_amount)
def test_get_amount_negative(self):
transaction = BpvfTransaction(trnamt=-52.2)
transaction = OfxTransaction(trnamt=-52.2)
expected_amount = -52200
result_amount = BpvfLine(transaction).get_amount()

18
tests/test_end_to_end.py
View file

@ -8,6 +8,7 @@ from click.testing import CliRunner
from ofx_processor.processors.bpvf import BpvfProcessor
from ofx_processor.processors.ce import CeProcessor
from ofx_processor.processors.lcl import LclProcessor
from ofx_processor.processors.revolut import RevolutProcessor
from ofx_processor.utils import utils
from ofx_processor.utils import ynab
@ -25,6 +26,7 @@ class UtilsTestCase(unittest.TestCase):
ce_main = CeProcessor.main
bpvf_main = BpvfProcessor.main
revolut_main = RevolutProcessor.main
lcl_main = LclProcessor.main
runner = CliRunner()
with mock.patch("click.core.Group.add_command") as add_command:
from ofx_processor.main import cli
@ -34,6 +36,7 @@ class UtilsTestCase(unittest.TestCase):
call(ce_main),
call(bpvf_main),
call(revolut_main),
call(lcl_main),
call(config, name="config"),
]
add_command.assert_has_calls(calls, any_order=True)
@ -264,3 +267,18 @@ class DataTestCase(unittest.TestCase):
post.assert_called_once_with(
expected_url, json=expected_data, headers=expected_headers
)
@mock.patch("requests.post")
def test_lcl_sends_to_ynab(self, post):
with open("tests/samples/lcl_transactions.json", encoding="utf-8") as f:
expected_data = json.load(f)
expected_headers = {"Authorization": f"Bearer <YOUR API TOKEN>"}
expected_url = f"{ynab.BASE_URL}/budgets/<YOUR BUDGET ID>/transactions"
runner = CliRunner()
runner.invoke(self.cli, ["lcl", "tests/samples/lcl.ofx"])
post.assert_called_once_with(
expected_url, json=expected_data, headers=expected_headers
)

73
tests/test_lcl_processor.py Normal file
View file

@ -0,0 +1,73 @@
import datetime
import json
import unittest
from ofx_processor.processors.lcl import LclLine, LclProcessor
from tests.utils import OfxTransaction
class LclLineTestCase(unittest.TestCase):
def test_get_name(self):
name = "VIR INST"
transaction = OfxTransaction(name=name)
result_name = LclLine(transaction).get_payee()
self.assertEqual(result_name, name)
def test_get_memo(self):
memo = "VIR INST"
transaction = OfxTransaction(memo=memo)
result_memo = LclLine(transaction).get_memo()
self.assertEqual(result_memo, memo)
def test_get_date(self):
transaction = OfxTransaction(dtposted=datetime.datetime(2020, 1, 23, 1, 2, 3))
expected_date = "2020-01-23"
result_date = LclLine(transaction).get_date()
self.assertEqual(result_date, expected_date)
def test_get_amount_positive(self):
transaction = OfxTransaction(trnamt=52.2)
expected_amount = 52200
result_amount = LclLine(transaction).get_amount()
self.assertEqual(result_amount, expected_amount)
def test_get_amount_negative(self):
transaction = OfxTransaction(trnamt=-52.2)
expected_amount = -52200
result_amount = LclLine(transaction).get_amount()
self.assertEqual(result_amount, expected_amount)
class LclProcessorTestCase(unittest.TestCase):
def test_file_not_found(self):
with self.assertRaises(SystemExit):
LclProcessor("filenotfound.ofx").get_transactions()
def test_file(self):
transactions = LclProcessor("tests/samples/lcl.ofx").get_transactions()
with open("tests/samples/lcl_expected.json") as f:
expected_transactions = json.load(f)
self.assertListEqual(transactions, expected_transactions)
def test_file_as_downloaded(self):
transactions = LclProcessor(
"tests/samples/lcl_as_downloaded.ofx"
).get_transactions()
with open("tests/samples/lcl_expected.json") as f:
expected_transactions = json.load(f)
self.assertListEqual(transactions, expected_transactions)
def test_file_malformed(self):
with self.assertRaises(SystemExit):
LclProcessor("tests/samples/lcl_malformed.ofx")
if __name__ == "__main__":
unittest.main() # pragma: nocover

19
tests/utils.py Normal file
View file

@ -0,0 +1,19 @@
import datetime
class OfxTransaction:
"""
Mimick what is retrieved via ofxtools when parsing the file
"""
def __init__(
self,
name: str = "",
memo: str = "",
dtposted: datetime.datetime = None,
trnamt: float = 0,
):
self.dtposted = dtposted
self.memo = memo
self.trnamt = trnamt
self.name = name