Simplify code needed to add a new bank processor

This commit is contained in:
Gabriel Augendre 2020-02-23 09:23:42 +01:00
parent 21053657b7
commit d341239978
No known key found for this signature in database
GPG key ID: 1E693F4CE4AEE7B4
13 changed files with 42 additions and 57 deletions

View file

@ -1,9 +0,0 @@
import click
from ofx_processor.bpvf_processor.bpvf_processor import BpvfProcessor
@click.command("bpvf", help="Process BPVF bank statement (OFX)")
@click.argument("ofx_filename")
def main(ofx_filename):
BpvfProcessor(ofx_filename).push_to_ynab()

View file

@ -1,9 +0,0 @@
import click
from ofx_processor.ce_processor.ce_processor import CeProcessor
@click.command("ce", help="Process CE bank statement (OFX)")
@click.argument("ofx_filename")
def main(ofx_filename):
CeProcessor(ofx_filename).push_to_ynab()

View file

@ -1,5 +0,0 @@
from ofx_processor.bpvf_processor.bpvf_processor import BpvfProcessor
class CeProcessor(BpvfProcessor):
account_name = "ce"

View file

@ -59,3 +59,9 @@ class BpvfProcessor(Processor):
sys.exit(1) sys.exit(1)
return ofx.statements[0].transactions return ofx.statements[0].transactions
@staticmethod
@click.command("bpvf", help="Process BPVF bank statement (OFX)")
@click.argument("ofx_filename")
def main(ofx_filename):
BpvfProcessor(ofx_filename).push_to_ynab()

View file

@ -0,0 +1,13 @@
import click
from ofx_processor.processors.bpvf import BpvfProcessor
class CeProcessor(BpvfProcessor):
account_name = "ce"
@staticmethod
@click.command("ce", help="Process CE bank statement (OFX)")
@click.argument("ofx_filename")
def main(ofx_filename):
CeProcessor(ofx_filename).push_to_ynab()

View file

@ -1,5 +1,6 @@
import csv import csv
import click
import dateparser import dateparser
from ofx_processor.utils.processor import Processor, Line from ofx_processor.utils.processor import Processor, Line
@ -54,3 +55,9 @@ class RevolutProcessor(Processor):
with open(self.filename) as f: with open(self.filename) as f:
reader = csv.DictReader(f, delimiter=";") reader = csv.DictReader(f, delimiter=";")
return reader return reader
@staticmethod
@click.command("revolut", help="Process Revolut bank statement (CSV)")
@click.argument("csv_filename")
def main(csv_filename):
RevolutProcessor(csv_filename).push_to_ynab()

View file

@ -1,9 +0,0 @@
import click
from ofx_processor.revolut_processor.revolut_processor import RevolutProcessor
@click.command("revolut", help="Process Revolut bank statement (CSV)")
@click.argument("csv_filename")
def main(csv_filename):
RevolutProcessor(csv_filename).push_to_ynab()

View file

@ -3,34 +3,25 @@ import pkgutil
import click import click
import ofx_processor from ofx_processor import processors
def discover_processors(cli: click.Group): def discover_processors(cli: click.Group):
""" """
Discover processors. To be discovered, CLIs need to match the following structure: Discover processors.
ofx_processor To be discovered, processors must:
__init__.py * Be in the `processors` package.
<package_name>_processor * Declare a <BankName>Processor class
   __init__.py * Declare a static main function in this class,
   <module_name>_cli.py which must be a click command
   <module_name>_processor.py
Preferably, package_name and module_name are the same, but there is no restriction.
Finally, a "main" function will be looked for inside the <module_name>_cli module.
This function must be a Click command with a unique name and will be added to the
main command line interface.
:param cli: The main CLI to add discovered processors to. :param cli: The main CLI to add discovered processors to.
""" """
prefix = ofx_processor.__name__ + "." prefix = processors.__name__ + "."
for package in pkgutil.iter_modules(ofx_processor.__path__, prefix): for module in pkgutil.iter_modules(processors.__path__, prefix):
if package.name.endswith("_processor"):
module_prefix = package.name + "."
package = importlib.import_module(package.name)
for module in pkgutil.iter_modules(package.__path__, module_prefix):
if module.name.endswith("_cli"):
module = importlib.import_module(module.name) module = importlib.import_module(module.name)
if "main" in dir(module): for item in dir(module):
cli.add_command(module.main) if item.endswith("Processor") and item != "Processor":
cls = getattr(module, item)
cli.add_command(cls.main)

View file

@ -1,6 +1,6 @@
import unittest import unittest
from ofx_processor.bpvf_processor.bpvf_processor import _process_name_and_memo from ofx_processor.processors.bpvf import _process_name_and_memo
class MyTestCase(unittest.TestCase): class MyTestCase(unittest.TestCase):

View file

@ -1,7 +1,7 @@
import datetime import datetime
import unittest import unittest
from ofx_processor.revolut_processor.revolut_processor import ( from ofx_processor.processors.revolut import (
_amount_str_to_float, _amount_str_to_float,
RevolutLine, RevolutLine,
) )