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