diff --git a/pfbudget/extract/parsers.py b/pfbudget/extract/parsers.py index 00c2375..cee0c3c 100644 --- a/pfbudget/extract/parsers.py +++ b/pfbudget/extract/parsers.py @@ -1,58 +1,45 @@ -from collections import namedtuple +from __future__ import annotations from decimal import Decimal from importlib import import_module from pathlib import Path import datetime as dt +from typing import Any, Callable, NamedTuple, Optional import yaml from pfbudget.common.types import NoBankSelected -from pfbudget.db.model import Transaction +from pfbudget.db.model import BankTransaction from pfbudget.utils import utils -Index = namedtuple( - "Index", ["date", "text", "value", "negate"], defaults=[-1, -1, -1, False] -) -Options = namedtuple( - "Options", - [ - "encoding", - "separator", - "date_fmt", - "start", - "end", - "debit", - "credit", - "additional_parser", - "category", - "VISA", - "MasterCard", - "AmericanExpress", - ], - defaults=[ - "", - "", - "", - 1, - None, - Index(), - Index(), - False, - None, - None, - None, - None, - ], -) + +class Index(NamedTuple): + date: int = -1 + text: int = -1 + value: int = -1 + negate: bool = False -def parse_data(filename: Path, args: dict) -> list[Transaction]: - cfg: dict = yaml.safe_load(open("parsers.yaml")) +class Options(NamedTuple): + encoding: str + separator: str + date_fmt: str + start: int = 1 + end: Optional[int] = None + debit: Index = Index() + credit: Index = Index() + additional_parser: bool = False + VISA: Optional[Options] = None + MasterCard: Optional[Options] = None + AmericanExpress: Optional[Options] = None + + +def parse_data(filename: Path, args: dict[str, Any]) -> list[BankTransaction]: + cfg: dict[str, Any] = yaml.safe_load(open("parsers.yaml")) assert ( "Banks" in cfg ), "parsers.yaml is missing the Banks section with the list of available banks" if not args["bank"]: - bank, creditcard = utils.find_credit_institution( + bank, creditcard = utils.find_credit_institution( # type: ignore filename, cfg.get("Banks"), cfg.get("CreditCards") ) else: @@ -60,7 +47,7 @@ def parse_data(filename: Path, args: dict) -> list[Transaction]: creditcard = None if not args["creditcard"] else args["creditcard"][0] try: - options: dict = cfg[bank] + options: dict[str, Any] = cfg[bank] except KeyError as e: banks = cfg["Banks"] raise NoBankSelected(f"{e} not a valid bank, try one of {banks}") @@ -73,9 +60,6 @@ def parse_data(filename: Path, args: dict) -> list[Transaction]: raise NoBankSelected(f"{e} not a valid bank, try one of {creditcards}") bank += creditcard - if args["category"]: - options["category"] = args["category"][0] - if options.get("additional_parser"): parser = getattr(import_module("pfbudget.extract.parsers"), bank) transactions = parser(filename, bank, options).parse() @@ -86,7 +70,7 @@ def parse_data(filename: Path, args: dict) -> list[Transaction]: class Parser: - def __init__(self, filename: Path, bank: str, options: dict): + def __init__(self, filename: Path, bank: str, options: dict[str, Any]): self.filename = filename self.bank = bank @@ -97,10 +81,10 @@ class Parser: self.options = Options(**options) - def func(self, transaction: Transaction): + def func(self, transaction: BankTransaction): pass - def parse(self) -> list[Transaction]: + def parse(self) -> list[BankTransaction]: transactions = [ Parser.transaction(line, self.bank, self.options, self.func) for line in list(open(self.filename, encoding=self.options.encoding))[ @@ -111,7 +95,8 @@ class Parser: return transactions @staticmethod - def index(line: list, options: Options) -> Index: + def index(line: list[str], options: Options) -> Index: + index = None if options.debit.date != -1 and options.credit.date != -1: if options.debit.value != options.credit.value: if line[options.debit.value]: @@ -138,28 +123,22 @@ class Parser: else: raise IndexError("No debit not credit indexes available") - return index + return index if index else Index() @staticmethod - def transaction(line: str, bank: str, options: Options, func) -> Transaction: - line = line.rstrip().split(options.separator) + def transaction( + line_: str, bank: str, options: Options, func: Callable[[BankTransaction], None] + ) -> BankTransaction: + line = line_.rstrip().split(options.separator) index = Parser.index(line, options) - date = ( - dt.datetime.strptime(line[index.date].strip(), options.date_fmt) - .date() - .isoformat() - ) + date = dt.datetime.strptime(line[index.date].strip(), options.date_fmt).date() text = line[index.text] value = utils.parse_decimal(line[index.value]) if index.negate: value = -value - if options.category: - category = line[options.category] - transaction = Transaction(date, text, bank, value, category) - else: - transaction = Transaction(date, text, bank, value) + transaction = BankTransaction(date, text, value, bank=bank) if options.additional_parser: func(transaction) @@ -167,20 +146,26 @@ class Parser: class Bank1(Parser): - def __init__(self, filename: str, bank: str, options: dict): + def __init__(self, filename: Path, bank: str, options: dict[str, Any]): super().__init__(filename, bank, options) - self.transfers = [] + self.transfers: list[dt.date] = [] self.transaction_cost = -Decimal("1") - def func(self, transaction: Transaction): - if "transf" in transaction.description.lower() and transaction.value < 0: - transaction.value -= self.transaction_cost + def func(self, transaction: BankTransaction): + if ( + transaction.description + and "transf" in transaction.description.lower() + and transaction.amount < 0 + ): + transaction.amount -= self.transaction_cost self.transfers.append(transaction.date) - def parse(self) -> list: + def parse(self) -> list[BankTransaction]: transactions = super().parse() for date in self.transfers: transactions.append( - Transaction(date, "Transaction cost", self.bank, self.transaction_cost) + BankTransaction( + date, "Transaction cost", self.transaction_cost, bank=self.bank + ) ) return transactions