From e7648b7f04a2122e7af565c59124f21f85737e98 Mon Sep 17 00:00:00 2001 From: Patrick Ruckstuhl Date: Fri, 27 Dec 2024 15:45:44 +0000 Subject: [PATCH] Upgrade for beancount3/beangulp --- .pre-commit-config.yaml | 3 +- setup.cfg | 5 +- src/tariochbctools/importers/bcge/importer.py | 7 +- .../importers/bitst/importer.py | 42 ++++---- .../importers/blockchain/importer.py | 17 ++-- .../importers/cembrastatement/importer.py | 72 ++++++++------ .../importers/general/mailAdapterImporter.py | 25 +++-- .../importers/general/mt940importer.py | 37 ++++---- .../importers/general/priceLookup.py | 12 +-- src/tariochbctools/importers/ibkr/importer.py | 39 ++++---- src/tariochbctools/importers/neon/importer.py | 32 ++++--- .../importers/netbenefits/importer.py | 40 ++++---- .../importers/nordigen/importer.py | 14 +-- .../importers/nordigen/nordigen_config.py | 14 +-- .../importers/postfinance/importer.py | 37 ++++---- .../importers/quickfile/importer.py | 22 ++--- .../importers/raiffeisench/importer.py | 10 +- .../importers/revolut/importer.py | 41 ++++---- .../importers/schedule/importer.py | 19 ++-- .../importers/swisscard/importer.py | 32 ++++--- .../importers/transferwise/importer.py | 14 +-- .../importers/truelayer/importer.py | 26 ++--- .../importers/viseca/importer.py | 40 ++++---- src/tariochbctools/importers/zak/importer.py | 95 ++++++++++--------- src/tariochbctools/importers/zkb/importer.py | 5 +- src/tariochbctools/plugins/prices/ibkr.py | 9 +- .../importers/test_quickfile.py | 5 +- .../importers/test_truelayer.py | 5 +- 28 files changed, 391 insertions(+), 328 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index bf36254..24c9091 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -52,4 +52,5 @@ repos: rev: v1.10.1 hooks: - id: mypy - args: [--install-types, --non-interactive, --ignore-missing-imports] + args: [--install-types, --non-interactive, --ignore-missing-imports, --disallow-incomplete-defs] +# args: [--install-types, --non-interactive, --ignore-missing-imports, --disallow-untyped-defs] diff --git a/setup.cfg b/setup.cfg index bd88e51..91808e9 100644 --- a/setup.cfg +++ b/setup.cfg @@ -38,7 +38,9 @@ package_dir = # install_requires = numpy; scipy install_requires = importlib-metadata; python_version<"3.8" - beancount>=2,<3 + beancount>=3 + beangulp + beanprice bitstampclient mt-940 pyyaml @@ -49,6 +51,7 @@ install_requires = blockcypher imap-tools undictify + rsa # The usage of test_requires is discouraged, see `Dependency Management` docs # tests_require = pytest; pytest-cov # Require a specific Python version, e.g. Python 2.7 or >= 3.4 diff --git a/src/tariochbctools/importers/bcge/importer.py b/src/tariochbctools/importers/bcge/importer.py index 3742bb6..2d26321 100644 --- a/src/tariochbctools/importers/bcge/importer.py +++ b/src/tariochbctools/importers/bcge/importer.py @@ -1,14 +1,15 @@ import re +from typing import Any from tariochbctools.importers.general import mt940importer -def strip_newline(string): +def strip_newline(string: str) -> str: return string.replace("\n", "").replace("\r", "") class BCGEImporter(mt940importer.Importer): - def prepare_payee(self, trxdata): + def prepare_payee(self, trxdata: dict[str, Any]) -> str: transaction_details = strip_newline(trxdata["transaction_details"]) payee = re.search(r"ORDP/([^/]+)", transaction_details) if payee is None: @@ -16,7 +17,7 @@ def prepare_payee(self, trxdata): else: return payee.group(1) - def prepare_narration(self, trxdata): + def prepare_narration(self, trxdata: dict[str, Any]) -> str: transaction_details = strip_newline(trxdata["transaction_details"]) extra_details = strip_newline(trxdata["extra_details"]) beneficiary = re.search(r"/BENM/([^/]+)", transaction_details) diff --git a/src/tariochbctools/importers/bitst/importer.py b/src/tariochbctools/importers/bitst/importer.py index d8b009b..fe890b5 100644 --- a/src/tariochbctools/importers/bitst/importer.py +++ b/src/tariochbctools/importers/bitst/importer.py @@ -1,36 +1,38 @@ from datetime import date from os import path +from typing import Any +import beangulp import bitstamp.client import yaml from beancount.core import amount, data from beancount.core.number import MISSING, D -from beancount.ingest import importer from dateutil.parser import parse from dateutil.relativedelta import relativedelta from tariochbctools.importers.general.priceLookup import PriceLookup -class Importer(importer.ImporterProtocol): +class Importer(beangulp.Importer): """An importer for Bitstamp.""" - def identify(self, file): - return path.basename(file.name).endswith("bitstamp.yaml") + def identify(self, filepath: str) -> bool: + return path.basename(filepath).endswith("bitstamp.yaml") - def file_account(self, file): + def account(self, filepath: str) -> data.Account: return "" - def extract(self, file, existing_entries): - self.priceLookup = PriceLookup(existing_entries, "CHF") + def extract(self, filepath: str, existing: data.Entries) -> data.Entries: + self.priceLookup = PriceLookup(existing, "CHF") - config = yaml.safe_load(file.contents()) + with open(filepath) as file: + config = yaml.safe_load(file) self.config = config self.client = bitstamp.client.Trading( username=config["username"], key=config["key"], secret=config["secret"] ) self.currencies = config["currencies"] - self.account = config["account"] + self._account = config["account"] self.otherExpensesAccount = config["otherExpensesAccount"] self.capGainAccount = config["capGainAccount"] @@ -46,7 +48,7 @@ def extract(self, file, existing_entries): return result - def fetchSingle(self, trx): + def fetchSingle(self, trx: dict[str, Any]) -> data.Transaction: id = int(trx["id"]) type = int(trx["type"]) date = parse(trx["datetime"]).date() @@ -67,12 +69,13 @@ def fetchSingle(self, trx): if type == 0: narration = "Deposit" - cost = data.Cost( - self.priceLookup.fetchPriceAmount(posCcy, date), "CHF", None, None - ) + if posCcy: + cost = data.Cost( + self.priceLookup.fetchPriceAmount(posCcy, date), "CHF", None, None + ) postings = [ data.Posting( - self.account + ":" + posCcy, + self._account + ":" + posCcy, amount.Amount(posAmt, posCcy), cost, None, @@ -84,7 +87,7 @@ def fetchSingle(self, trx): narration = "Withdrawal" postings = [ data.Posting( - self.account + ":" + negCcy, + self._account + ":" + negCcy, amount.Amount(negAmt, negCcy), None, None, @@ -94,14 +97,15 @@ def fetchSingle(self, trx): ] elif type == 2: fee = D(trx["fee"]) - if posCcy.lower() + "_" + negCcy.lower() in trx: + if posCcy and negCcy and posCcy.lower() + "_" + negCcy.lower() in trx: feeCcy = negCcy negAmt -= fee else: feeCcy = posCcy posAmt -= fee - rateFiatCcy = self.priceLookup.fetchPriceAmount(feeCcy, date) + if feeCcy: + rateFiatCcy = self.priceLookup.fetchPriceAmount(feeCcy, date) if feeCcy == posCcy: posCcyCost = None posCcyPrice = amount.Amount(rateFiatCcy, "CHF") @@ -119,7 +123,7 @@ def fetchSingle(self, trx): postings = [ data.Posting( - self.account + ":" + posCcy, + self._account + ":" + posCcy, amount.Amount(posAmt, posCcy), posCcyCost, posCcyPrice, @@ -127,7 +131,7 @@ def fetchSingle(self, trx): None, ), data.Posting( - self.account + ":" + negCcy, + self._account + ":" + negCcy, amount.Amount(negAmt, negCcy), negCcyCost, negCcyPrice, diff --git a/src/tariochbctools/importers/blockchain/importer.py b/src/tariochbctools/importers/blockchain/importer.py index 1e9f138..8a3bac3 100644 --- a/src/tariochbctools/importers/blockchain/importer.py +++ b/src/tariochbctools/importers/blockchain/importer.py @@ -1,28 +1,29 @@ from os import path +import beangulp import blockcypher import yaml from beancount.core import amount, data from beancount.core.number import D -from beancount.ingest import importer from tariochbctools.importers.general.priceLookup import PriceLookup -class Importer(importer.ImporterProtocol): +class Importer(beangulp.Importer): """An importer for Blockchain data.""" - def identify(self, file): - return path.basename(file.name).endswith("blockchain.yaml") + def identify(self, filepath: str) -> bool: + return path.basename(filepath).endswith("blockchain.yaml") - def file_account(self, file): + def account(self, filepath: str) -> data.Entries: return "" - def extract(self, file, existing_entries): - config = yaml.safe_load(file.contents()) + def extract(self, filepath: str, existing: data.Entries) -> data.Entries: + with open(filepath) as file: + config = yaml.safe_load(file) self.config = config baseCcy = config["base_ccy"] - priceLookup = PriceLookup(existing_entries, baseCcy) + priceLookup = PriceLookup(existing, baseCcy) entries = [] for address in self.config["addresses"]: diff --git a/src/tariochbctools/importers/cembrastatement/importer.py b/src/tariochbctools/importers/cembrastatement/importer.py index 4925401..ddcd1d9 100644 --- a/src/tariochbctools/importers/cembrastatement/importer.py +++ b/src/tariochbctools/importers/cembrastatement/importer.py @@ -1,26 +1,31 @@ +import datetime import re -from datetime import datetime, timedelta +from datetime import timedelta +import beangulp import camelot from beancount.core import amount, data from beancount.core.number import D -from beancount.ingest import importer -from beancount.ingest.importers.mixins import identifier -class Importer(identifier.IdentifyMixin, importer.ImporterProtocol): +class Importer(beangulp.Importer): """An importer for Cembra Card Statement PDF files.""" - def __init__(self, regexps, account): - identifier.IdentifyMixin.__init__(self, matchers=[("filename", regexps)]) - self.account = account + def __init__(self, filepattern: str, account: data.Account): + self._filepattern = filepattern + self._account = account self.currency = "CHF" - def file_account(self, file): - return self.account + def identify(self, filepath: str) -> bool: + return re.search(self._filepattern, filepath) is not None - def createEntry(self, file, date, amt, text): - meta = data.new_metadata(file.name, 0) + def account(self, filepath: str) -> data.Account: + return self._account + + def createEntry( + self, filepath: str, date: datetime.date, amt: data.Decimal, text: str + ) -> data.Transaction: + meta = data.new_metadata(filepath, 0) return data.Transaction( meta, date, @@ -30,19 +35,21 @@ def createEntry(self, file, date, amt, text): data.EMPTY_SET, data.EMPTY_SET, [ - data.Posting(self.account, amt, None, None, None, None), + data.Posting(self._account, amt, None, None, None, None), ], ) - def createBalanceEntry(self, file, date, amt): - meta = data.new_metadata(file.name, 0) - return data.Balance(meta, date, self.account, amt, None, None) + def createBalanceEntry( + self, filepath: str, date: datetime.date, amt: data.Decimal + ) -> data.Balance: + meta = data.new_metadata(filepath, 0) + return data.Balance(meta, date, self._account, amt, None, None) - def extract(self, file, existing_entries): + def extract(self, filepath: str, existing: data.Entries) -> data.Entries: entries = [] tables = camelot.read_pdf( - file.name, pages="2-end", flavor="stream", table_areas=["50,700,560,50"] + filepath, pages="2-end", flavor="stream", table_areas=["50,700,560,50"] ) for table in tables: df = table.df @@ -63,7 +70,7 @@ def extract(self, file, existing_entries): # Transaction entry try: - book_date = datetime.strptime(book_date, "%d.%m.%Y").date() + book_date = datetime.datetime.strptime(book_date, "%d.%m.%Y").date() except Exception: book_date = None @@ -71,17 +78,24 @@ def extract(self, file, existing_entries): amount = self.getAmount(debit, credit) if amount: - entries.append(self.createEntry(file, book_date, amount, text)) + entries.append( + self.createEntry(filepath, book_date, amount, text) + ) continue # Balance entry try: - book_date = re.search( - r"Saldo per (\d\d\.\d\d\.\d\d\d\d) zu unseren Gunsten CHF", text - ).group(1) - book_date = datetime.strptime(book_date, "%d.%m.%Y").date() - # add 1 day: cembra provides balance at EOD, but beancount checks it at SOD - book_date = book_date + timedelta(days=1) + m = re.search( + r"Saldo per (\d\d\.\d\d\.\d\d\d\d) zu unseren Gunsten CHF", + text, + ) + if m: + book_date = m.group(1) + book_date = datetime.datetime.strptime( + book_date, "%d.%m.%Y" + ).date() + # add 1 day: cembra provides balance at EOD, but beancount checks it at SOD + book_date = book_date + timedelta(days=1) except Exception: book_date = None @@ -89,14 +103,16 @@ def extract(self, file, existing_entries): amount = self.getAmount(debit, credit) if amount: - entries.append(self.createBalanceEntry(file, book_date, amount)) + entries.append( + self.createBalanceEntry(filepath, book_date, amount) + ) return entries - def cleanDecimal(self, formattedNumber): + def cleanDecimal(self, formattedNumber: str) -> data.Decimal: return D(formattedNumber.replace("'", "")) - def getAmount(self, debit, credit): + def getAmount(self, debit: str, credit: str) -> data.Amount: amt = -self.cleanDecimal(debit) if debit else self.cleanDecimal(credit) if amt: return amount.Amount(amt, self.currency) diff --git a/src/tariochbctools/importers/general/mailAdapterImporter.py b/src/tariochbctools/importers/general/mailAdapterImporter.py index 16a8b13..7a558ec 100644 --- a/src/tariochbctools/importers/general/mailAdapterImporter.py +++ b/src/tariochbctools/importers/general/mailAdapterImporter.py @@ -2,24 +2,26 @@ from os import path import yaml -from beancount.ingest import cache, importer +from beancount.core import data +from beangulp import Importer from imap_tools import MailBox -class MailAdapterImporter(importer.ImporterProtocol): +class MailAdapterImporter(Importer): """An importer adapter that fetches file from mails and then calls another importer.""" - def __init__(self, importers): + def __init__(self, importers: list[Importer]): self.importers = importers - def identify(self, file): - return "mail.yaml" == path.basename(file.name) + def identify(self, filepath: str) -> bool: + return "mail.yaml" == path.basename(filepath) - def file_account(self, file): + def account(self, filepath: str) -> data.Account: return "" - def extract(self, file, existing_entries): - config = yaml.safe_load(file.contents()) + def extract(self, filepath: str, existing: data.Entries) -> data.Entries: + with open(filepath) as file: + config = yaml.safe_load(file) with MailBox(config["host"]).login( config["user"], config["password"], initial_folder=config["folder"] @@ -33,13 +35,10 @@ def extract(self, file, existing_entries): with open(attFileName, "wb") as attFile: attFile.write(att.payload) attFile.flush() - fileMemo = cache.get_file(attFileName) for delegate in self.importers: - if delegate.identify(fileMemo): - newEntries = delegate.extract( - fileMemo, existing_entries - ) + if delegate.identify(attFileName): + newEntries = delegate.extract(attFileName, existing) result.extend(newEntries) processed = True diff --git a/src/tariochbctools/importers/general/mt940importer.py b/src/tariochbctools/importers/general/mt940importer.py index d8fb475..24979fc 100644 --- a/src/tariochbctools/importers/general/mt940importer.py +++ b/src/tariochbctools/importers/general/mt940importer.py @@ -1,29 +1,28 @@ +import re +from typing import Any + +import beangulp import mt940 from beancount.core import amount, data from beancount.core.number import D -from beancount.ingest import importer -from beancount.ingest.importers.mixins import identifier -class Importer(identifier.IdentifyMixin, importer.ImporterProtocol): +class Importer(beangulp.Importer): """An importer for MT940 files.""" - def __init__(self, regexps, account): - identifier.IdentifyMixin.__init__(self, matchers=[("filename", regexps)]) - self.account = account - - def identify(self, file): - if file.mimetype() != "text/plain": - return False + def __init__(self, filepattern: str, account: data.Account): + self._filepattern = filepattern + self._account = account - return super().identify(file) + def identify(self, filepath: str) -> bool: + return re.search(self._filepattern, filepath) is not None - def file_account(self, file): - return self.account + def account(self, filepath: str) -> data.Account: + return self._account - def extract(self, file, existing_entries): + def extract(self, filepath: str, existing: data.Entries) -> data.Entries: entries = [] - transactions = mt940.parse(file.contents()) + transactions = mt940.parse(filepath) for trx in transactions: trxdata = trx.data ref = trxdata["bank_reference"] @@ -31,7 +30,7 @@ def extract(self, file, existing_entries): metakv = {"ref": ref} else: metakv = None - meta = data.new_metadata(file.name, 0, metakv) + meta = data.new_metadata(filepath, 0, metakv) if "entry_date" in trxdata: date = trxdata["entry_date"] else: @@ -46,7 +45,7 @@ def extract(self, file, existing_entries): data.EMPTY_SET, [ data.Posting( - self.account, + self._account, amount.Amount( D(trxdata["amount"].amount), trxdata["amount"].currency ), @@ -61,8 +60,8 @@ def extract(self, file, existing_entries): return entries - def prepare_payee(self, trxdata): + def prepare_payee(self, trxdata: dict[str, Any]) -> str: return "" - def prepare_narration(self, trxdata): + def prepare_narration(self, trxdata: dict[str, Any]) -> str: return trxdata["transaction_details"] + " " + trxdata["extra_details"] diff --git a/src/tariochbctools/importers/general/priceLookup.py b/src/tariochbctools/importers/general/priceLookup.py index 05090d0..d304a54 100644 --- a/src/tariochbctools/importers/general/priceLookup.py +++ b/src/tariochbctools/importers/general/priceLookup.py @@ -1,18 +1,18 @@ from datetime import date -from beancount.core import amount, prices +from beancount.core import amount, data, prices from beancount.core.number import D class PriceLookup: - def __init__(self, existing_entries, baseCcy: str): - if existing_entries: - self.priceMap = prices.build_price_map(existing_entries) + def __init__(self, existing: data.Entries, baseCcy: str): + if existing: + self.priceMap = prices.build_price_map(existing) else: self.priceMap = None self.baseCcy = baseCcy - def fetchPriceAmount(self, instrument: str, date: date): + def fetchPriceAmount(self, instrument: str, date: date) -> data.Amount: if self.priceMap: price = prices.get_price( self.priceMap, tuple([instrument, self.baseCcy]), date @@ -21,7 +21,7 @@ def fetchPriceAmount(self, instrument: str, date: date): else: return D(1) - def fetchPrice(self, instrument: str, date: date): + def fetchPrice(self, instrument: str, date: date) -> data.Amount: if instrument == self.baseCcy: return None diff --git a/src/tariochbctools/importers/ibkr/importer.py b/src/tariochbctools/importers/ibkr/importer.py index 1d7deee..1893a08 100644 --- a/src/tariochbctools/importers/ibkr/importer.py +++ b/src/tariochbctools/importers/ibkr/importer.py @@ -2,27 +2,30 @@ from datetime import date from decimal import Decimal from os import path +from typing import Any +import beangulp import yaml from beancount.core import amount, data from beancount.core.number import D -from beancount.ingest import importer from ibflex import Types, client, parser from ibflex.enums import CashAction from tariochbctools.importers.general.priceLookup import PriceLookup -class Importer(importer.ImporterProtocol): +class Importer(beangulp.Importer): """An importer for Interactive Broker using the flex query service.""" - def identify(self, file): - return path.basename(file.name).endswith("ibkr.yaml") + def identify(self, filepath: str) -> bool: + return path.basename(filepath).endswith("ibkr.yaml") - def file_account(self, file): + def account(self, filepath: str) -> data.Account: return "" - def matches(self, trx, t, account): + def matches( + self, trx: Types.CashTransaction, t: Any, account: data.Account + ) -> bool: p = re.compile(r".* (?P\d+\.?\d+) PER SHARE") trxPerShareGroups = p.search(trx.description) @@ -38,13 +41,13 @@ def matches(self, trx, t, account): and t["account"] == account ) - def extract(self, file, existing_entries): - with open(file.name, "r") as f: + def extract(self, filepath: str, existing: data.Entries) -> data.Entries: + with open(filepath, "r") as f: config = yaml.safe_load(f) token = config["token"] queryId = config["queryId"] - priceLookup = PriceLookup(existing_entries, config["baseCcy"]) + priceLookup = PriceLookup(existing, config["baseCcy"]) response = client.download(token, queryId) statement = parser.parse(response) @@ -52,7 +55,7 @@ def extract(self, file, existing_entries): result = [] for stmt in statement.FlexStatements: - transactions = [] + transactions: list = [] account = stmt.accountId for trx in stmt.Trades: result.append( @@ -147,7 +150,7 @@ def createDividen( priceLookup: PriceLookup, description: str, account: str, - ): + ) -> data.Transaction: narration = "Dividend: " + description liquidityAccount = self.getLiquidityAccount(account, currency) incomeAccount = self.getIncomeAccount(account) @@ -190,7 +193,7 @@ def createDividen( def createBuy( self, date: date, - account: str, + account: data.Account, asset: str, quantity: Decimal, currency: str, @@ -199,7 +202,7 @@ def createBuy( netCash: amount.Amount, baseCcy: str, fxRateToBase: Decimal, - ): + ) -> data.Transaction: narration = "Buy" feeAccount = self.getFeeAccount(account) liquidityAccount = self.getLiquidityAccount(account, currency) @@ -238,17 +241,17 @@ def createBuy( meta, date, "*", "", narration, data.EMPTY_SET, data.EMPTY_SET, postings ) - def getAssetAccount(self, account: str, asset: str): + def getAssetAccount(self, account: str, asset: str) -> data.Account: return f"Assets:{account}:Investment:IB:{asset}" - def getLiquidityAccount(self, account: str, currency: str): + def getLiquidityAccount(self, account: str, currency: str) -> data.Account: return f"Assets:{account}:Liquidity:IB:{currency}" - def getReceivableAccount(self, account: str): + def getReceivableAccount(self, account: str) -> data.Account: return f"Assets:{account}:Receivable:Verrechnungssteuer" - def getIncomeAccount(self, account: str): + def getIncomeAccount(self, account: str) -> data.Account: return f"Income:{account}:Interest" - def getFeeAccount(self, account: str): + def getFeeAccount(self, account: str) -> data.Account: return f"Expenses:{account}:Fees" diff --git a/src/tariochbctools/importers/neon/importer.py b/src/tariochbctools/importers/neon/importer.py index 49023cd..3c03da2 100644 --- a/src/tariochbctools/importers/neon/importer.py +++ b/src/tariochbctools/importers/neon/importer.py @@ -1,30 +1,32 @@ import csv -from io import StringIO +import re +import beangulp from beancount.core import amount, data from beancount.core.number import D -from beancount.ingest import importer -from beancount.ingest.importers.mixins import identifier from dateutil.parser import parse -class Importer(identifier.IdentifyMixin, importer.ImporterProtocol): +class Importer(beangulp.Importer): """An importer for Neon CSV files.""" - def __init__(self, regexps, account): - identifier.IdentifyMixin.__init__(self, matchers=[("filename", regexps)]) - self.account = account + def __init__(self, filepattern: str, account: data.Account): + self._filepattern = filepattern + self._account = account - def name(self): - return super().name() + self.account + def identify(self, filepath: str) -> bool: + return re.search(self._filepattern, filepath) is not None - def file_account(self, file): - return self.account + def name(self) -> str: + return super().name() + self._account - def extract(self, file, existing_entries): + def account(self, filepath: str) -> data.Account: + return self._account + + def extract(self, filepath: str, existing: data.Entries) -> data.Entries: entries = [] - with StringIO(file.contents()) as csvfile: + with open(filepath) as csvfile: reader = csv.DictReader( csvfile, [ @@ -55,7 +57,7 @@ def extract(self, file, existing_entries): metakv["original_amount"] = row["Original amount"] metakv["exchange_rate"] = row["Exchange rate"] - meta = data.new_metadata(file.name, 0, metakv) + meta = data.new_metadata(filepath, 0, metakv) description = row["Description"].strip() if row["Subject"].strip() != "": description = description + ": " + row["Subject"].strip() @@ -69,7 +71,7 @@ def extract(self, file, existing_entries): data.EMPTY_SET, data.EMPTY_SET, [ - data.Posting(self.account, amt, None, None, None, None), + data.Posting(self._account, amt, None, None, None, None), ], ) entries.append(entry) diff --git a/src/tariochbctools/importers/netbenefits/importer.py b/src/tariochbctools/importers/netbenefits/importer.py index b01e506..aacd75f 100644 --- a/src/tariochbctools/importers/netbenefits/importer.py +++ b/src/tariochbctools/importers/netbenefits/importer.py @@ -1,24 +1,23 @@ import csv +import re from collections.abc import Iterable from datetime import date -from io import StringIO +import beangulp from beancount.core import amount, data from beancount.core.number import D from beancount.core.position import CostSpec -from beancount.ingest import importer -from beancount.ingest.importers.mixins import identifier from dateutil.parser import parse from tariochbctools.importers.general.priceLookup import PriceLookup -class Importer(identifier.IdentifyMixin, importer.ImporterProtocol): +class Importer(beangulp.Importer): """An importer for Fidelity Netbenefits Activity CSV files.""" def __init__( self, - regexps: str | Iterable[str], + filepattern: str, cashAccount: str, investmentAccount: str, dividendAccount: str, @@ -28,7 +27,7 @@ def __init__( ignoreTypes: Iterable[str], baseCcy: str, ): - identifier.IdentifyMixin.__init__(self, matchers=[("filename", regexps)]) + self.filepattern = filepattern self.cashAccount = cashAccount self.investmentAccount = investmentAccount self.dividendAccount = dividendAccount @@ -38,18 +37,21 @@ def __init__( self.ignoreTypes = ignoreTypes self.baseCcy = baseCcy - def name(self): + def name(self) -> str: return super().name() + self.cashAccount - def file_account(self, file): + def identify(self, filepath: str) -> bool: + return re.search(self.filepattern, filepath) is not None + + def account(self, filepath: str) -> data.Account: return self.cashAccount - def extract(self, file, existing_entries): + def extract(self, filepath: str, existing: data.Entries) -> data.Entries: entries = [] - self.priceLookup = PriceLookup(existing_entries, self.baseCcy) + self.priceLookup = PriceLookup(existing, self.baseCcy) - with StringIO(file.contents()) as csvfile: + with open(filepath) as csvfile: reader = csv.DictReader( csvfile, [ @@ -76,12 +78,12 @@ def extract(self, file, existing_entries): if row["Shares"] != "-": shares = amount.Amount(D(row["Shares"]), self.symbol) - metakv = {} + metakv: data.Meta = {} if not amt and not shares: continue - meta = data.new_metadata(file.name, 0, metakv) + meta = data.new_metadata(filepath, 0, metakv) description = row["Transaction type"].strip() if "TAX" in description: @@ -120,7 +122,9 @@ def extract(self, file, existing_entries): return entries - def __createBuy(self, amt: amount, shares: amount, book_date: date): + def __createBuy( + self, amt: amount, shares: amount, book_date: date + ) -> list[data.Posting]: price = self.priceLookup.fetchPrice("USD", book_date) cost = CostSpec( number_per=None, @@ -137,7 +141,9 @@ def __createBuy(self, amt: amount, shares: amount, book_date: date): return postings - def __createSell(self, amt: amount, shares: amount, book_date: date): + def __createSell( + self, amt: amount, shares: amount, book_date: date + ) -> list[data.Posting]: price = self.priceLookup.fetchPrice("USD", book_date) cost = CostSpec( number_per=None, @@ -155,7 +161,9 @@ def __createSell(self, amt: amount, shares: amount, book_date: date): return postings - def __createDividend(self, amt: amount, book_date: date, incomeAccount: str): + def __createDividend( + self, amt: amount, book_date: date, incomeAccount: str + ) -> list[data.Posting]: price = self.priceLookup.fetchPrice("USD", book_date) postings = [ data.Posting( diff --git a/src/tariochbctools/importers/nordigen/importer.py b/src/tariochbctools/importers/nordigen/importer.py index a55dbe1..ddf58d9 100644 --- a/src/tariochbctools/importers/nordigen/importer.py +++ b/src/tariochbctools/importers/nordigen/importer.py @@ -1,28 +1,28 @@ from datetime import date from os import path +import beangulp import requests import yaml from beancount.core import amount, data from beancount.core.number import D -from beancount.ingest import importer class HttpServiceException(Exception): pass -class Importer(importer.ImporterProtocol): +class Importer(beangulp.Importer): """An importer for Nordigen API (e.g. for Revolut).""" - def identify(self, file): - return path.basename(file.name).endswith("nordigen.yaml") + def identify(self, filepath: str) -> bool: + return path.basename(filepath).endswith("nordigen.yaml") - def file_account(self, file): + def account(self, filepath: str) -> data.Entries: return "" - def extract(self, file, existing_entries): - with open(file.name, "r") as f: + def extract(self, filepath: str, existing: data.Entries) -> data.Entries: + with open(filepath, "r") as f: config = yaml.safe_load(f) r = requests.post( diff --git a/src/tariochbctools/importers/nordigen/nordigen_config.py b/src/tariochbctools/importers/nordigen/nordigen_config.py index 60f8fda..618a685 100644 --- a/src/tariochbctools/importers/nordigen/nordigen_config.py +++ b/src/tariochbctools/importers/nordigen/nordigen_config.py @@ -4,7 +4,7 @@ import requests -def build_header(token): +def build_header(token: str) -> dict[str, str]: return {"Authorization": "Bearer " + token} @@ -15,7 +15,7 @@ def check_result(result): raise Exception(e, e.response.text) -def get_token(secret_id, secret_key): +def get_token(secret_id: str, secret_key: str) -> str: r = requests.post( "https://bankaccountdata.gocardless.com/api/v2/token/new/", data={ @@ -28,7 +28,7 @@ def get_token(secret_id, secret_key): return r.json()["access"] -def list_bank(token, country): +def list_bank(token: str, country: str) -> None: r = requests.get( "https://bankaccountdata.gocardless.com/api/v2/institutions/", params={"country": country}, @@ -40,7 +40,7 @@ def list_bank(token, country): print(asp["name"] + ": " + asp["id"]) # noqa: T201 -def create_link(token, reference, bank): +def create_link(token: str, reference: str, bank: str) -> None: if not bank: raise Exception("Please specify --bank it is required for create_link") requisitionId = _find_requisition_id(token, reference) @@ -61,7 +61,7 @@ def create_link(token, reference, bank): print(f"Go to {link} for connecting to your bank.") # noqa: T201 -def list_accounts(token): +def list_accounts(token: str) -> None: headers = build_header(token) r = requests.get( "https://bankaccountdata.gocardless.com/api/v2/requisitions/", headers=headers @@ -93,7 +93,7 @@ def list_accounts(token): print(f"{account}: {asp} {owner} {iban} {currency}") # noqa: T201 -def delete_link(token, reference): +def delete_link(token: str, reference: str) -> None: requisitionId = _find_requisition_id(token, reference) if requisitionId: r = requests.delete( @@ -103,7 +103,7 @@ def delete_link(token, reference): check_result(r) -def _find_requisition_id(token, userId): +def _find_requisition_id(token: str, userId: str) -> str | None: headers = build_header(token) r = requests.get( "https://bankaccountdata.gocardless.com/api/v2/requisitions/", headers=headers diff --git a/src/tariochbctools/importers/postfinance/importer.py b/src/tariochbctools/importers/postfinance/importer.py index c05535e..bedb7b2 100644 --- a/src/tariochbctools/importers/postfinance/importer.py +++ b/src/tariochbctools/importers/postfinance/importer.py @@ -1,49 +1,52 @@ import csv import logging +import re from datetime import datetime, timedelta from decimal import Decimal +import beangulp from beancount.core import data -from beancount.ingest.importer import ImporterProtocol -from beancount.ingest.importers.mixins.identifier import IdentifyMixin -class Importer(IdentifyMixin, ImporterProtocol): +class Importer(beangulp.Importer): """An importer for PostFinance CSV.""" - def __init__(self, regexps, account, currency="CHF"): - IdentifyMixin.__init__(self, matchers=[("filename", regexps)]) - self.account = account + def __init__(self, filepattern: str, account: data.Account, currency: str = "CHF"): + self._filepattern = filepattern + self._account = account self.currency = currency - def file_account(self, file): - return self.account + def identify(self, filepath: str) -> bool: + return re.search(self._filepattern, filepath) is not None - def extract(self, file, existing_entries): - csvfile = open(file=file.name, encoding="windows_1252") + def account(self, filepath: str) -> data.Account: + return self._account + + def extract(self, filepath: str, existing: data.Entries) -> data.Entries: + csvfile = open(file=filepath, encoding="windows_1252") reader = csv.reader(csvfile, delimiter=";") - meta = data.new_metadata(file.name, 0) + meta = data.new_metadata(filepath, 0) entries = [] for row in reader: try: - book_date, text, credit, debit, val_date, balance = tuple(row) - book_date = datetime.strptime(book_date, "%Y-%m-%d").date() + book_date_str, text, credit, debit, val_date, balance_str = tuple(row) + book_date = datetime.strptime(book_date_str, "%Y-%m-%d").date() if credit: amount = data.Amount(Decimal(credit), self.currency) elif debit: amount = data.Amount(Decimal(debit), self.currency) else: amount = None - if balance: - balance = data.Amount(Decimal(balance), self.currency) + if balance_str: + balance = data.Amount(Decimal(balance_str), self.currency) else: balance = None except Exception as e: logging.debug(e) else: logging.debug((book_date, text, amount, val_date, balance)) - posting = data.Posting(self.account, amount, None, None, None, None) + posting = data.Posting(self._account, amount, None, None, None, None) entry = data.Transaction( meta, book_date, @@ -59,7 +62,7 @@ def extract(self, file, existing_entries): book_date = book_date + timedelta(days=1) if balance and book_date.day == 1: entry = data.Balance( - meta, book_date, self.account, balance, None, None + meta, book_date, self._account, balance, None, None ) entries.append(entry) diff --git a/src/tariochbctools/importers/quickfile/importer.py b/src/tariochbctools/importers/quickfile/importer.py index 6449fa3..9f1e5ca 100644 --- a/src/tariochbctools/importers/quickfile/importer.py +++ b/src/tariochbctools/importers/quickfile/importer.py @@ -5,11 +5,11 @@ from os import path from typing import Dict, List, NamedTuple +import beangulp import requests import yaml from beancount.core import amount, data from beancount.core.number import D -from beancount.ingest import importer from undictify import type_checked_constructor @@ -152,32 +152,32 @@ def bank_search( return QuickFileBankSearch(**body) -class Importer(importer.ImporterProtocol): +class Importer(beangulp.Importer): """An importer for QuickFile""" def __init__(self): self.quickfile = None self.config = None - self.existing_entries = None + self.existing = None - def _configure(self, file, existing_entries): - with open(file.name, "r") as config_file: + def _configure(self, filepath, existing): + with open(filepath, "r") as config_file: self.config = yaml.safe_load(config_file) self.quickfile = QuickFile( account_number=self.config["account_number"], api_key=self.config["api_key"], app_id=self.config["app_id"], ) - self.existing_entries = existing_entries + self.existing = existing - def identify(self, file): - return path.basename(file.name) == "quickfile.yaml" + def identify(self, filepath): + return path.basename(filepath) == "quickfile.yaml" - def file_account(self, file): + def account(self, filepath): return "" - def extract(self, file, existing_entries=None): - self._configure(file, existing_entries) + def extract(self, filepath, existing=None): + self._configure(filepath, existing) entries = [] for bank_account in self.config["accounts"].keys(): diff --git a/src/tariochbctools/importers/raiffeisench/importer.py b/src/tariochbctools/importers/raiffeisench/importer.py index 201bfa9..eeb40bb 100644 --- a/src/tariochbctools/importers/raiffeisench/importer.py +++ b/src/tariochbctools/importers/raiffeisench/importer.py @@ -1,4 +1,5 @@ import re +from typing import Any from tariochbctools.importers.general import mt940importer @@ -8,21 +9,16 @@ class RaiffeisenCHImporter(mt940importer.Importer): """To get the correct file, choose SWIFT -> 'Période prédéfinie du relevé de compte' -> Sans détails""" - def prepare_payee(self, trxdata): + def prepare_payee(self, trxdata: dict[str, Any]) -> str: return "" - def prepare_narration(self, trxdata): + def prepare_narration(self, trxdata: dict[str, Any]) -> str: extra = trxdata["extra_details"] details = trxdata["transaction_details"] - extraReplacements = {} - detailsReplacements = {} detailsReplacements[r"\n"] = ", " - for pattern, replacement in extraReplacements.items(): - extra = re.sub(pattern, replacement, extra) - for pattern, replacement in detailsReplacements.items(): details = re.sub(pattern, replacement, details) diff --git a/src/tariochbctools/importers/revolut/importer.py b/src/tariochbctools/importers/revolut/importer.py index a9bf965..f23b513 100644 --- a/src/tariochbctools/importers/revolut/importer.py +++ b/src/tariochbctools/importers/revolut/importer.py @@ -1,34 +1,39 @@ import csv import logging +import re from datetime import timedelta -from io import StringIO +from typing import Any +import beangulp from beancount.core import amount, data from beancount.core.number import ZERO, D -from beancount.ingest import importer -from beancount.ingest.importers.mixins import identifier from dateutil.parser import parse -class Importer(identifier.IdentifyMixin, importer.ImporterProtocol): +class Importer(beangulp.Importer): """An importer for Revolut CSV files.""" - def __init__(self, regexps, account, currency, fee=None): - identifier.IdentifyMixin.__init__(self, matchers=[("filename", regexps)]) - self.account = account + def __init__( + self, filepattern: str, account: data.Account, currency: str, fee: Any = None + ): + self._filepattern = filepattern + self._account = account self.currency = currency self._fee = fee - def name(self): - return super().name() + self.account + def name(self) -> str: + return super().name() + self._account - def file_account(self, file): - return self.account + def identify(self, filepath: str) -> bool: + return re.search(self._filepattern, filepath) is not None - def extract(self, file, existing_entries): + def account(self, filepath: str) -> data.Account: + return self._account + + def extract(self, filepath: str, existing: data.Entries) -> data.Entries: entries = [] - with StringIO(file.contents()) as csvfile: + with open(filepath) as csvfile: reader = csv.DictReader( csvfile, [ @@ -65,12 +70,12 @@ def extract(self, file, existing_entries): continue postings = [ - data.Posting(self.account, amt, None, None, None, None), + data.Posting(self._account, amt, None, None, None, None), ] description = row["Description"].strip() if is_fee_mode: postings = [ - data.Posting(self.account, fee, None, None, None, None), + data.Posting(self._account, fee, None, None, None, None), data.Posting( self._fee["account"], -fee, None, None, None, None ), @@ -82,7 +87,7 @@ def extract(self, file, existing_entries): ), "Actual type of description is " + str(type(description)) entry = data.Transaction( - data.new_metadata(file.name, 0, {}), + data.new_metadata(filepath, 0, {}), book_date, "*", "", @@ -98,9 +103,9 @@ def extract(self, file, existing_entries): try: book_date = book_date + timedelta(days=1) entry = data.Balance( - data.new_metadata(file.name, 0, {}), + data.new_metadata(filepath, 0, {}), book_date, - self.account, + self._account, balance, None, None, diff --git a/src/tariochbctools/importers/schedule/importer.py b/src/tariochbctools/importers/schedule/importer.py index 6fa9ea9..27596c4 100644 --- a/src/tariochbctools/importers/schedule/importer.py +++ b/src/tariochbctools/importers/schedule/importer.py @@ -1,24 +1,25 @@ import datetime from os import path +from typing import Any +import beangulp import yaml from beancount.core import amount, data from beancount.core.number import D -from beancount.ingest import importer from dateutil.relativedelta import relativedelta -class Importer(importer.ImporterProtocol): +class Importer(beangulp.Importer): """An importer for Scheduled/Recurring Transactions.""" - def identify(self, file): - return path.basename(file.name).endswith("schedule.yaml") + def identify(self, filepath: str) -> bool: + return path.basename(filepath).endswith("schedule.yaml") - def file_account(self, file): + def account(self, filepath: str) -> data.Account: return "" - def extract(self, file, existing_entries): - with open(file.name, "r") as f: + def extract(self, filepath: str, existing: data.Entries) -> data.Entries: + with open(filepath, "r") as f: config = yaml.safe_load(f) self.transactions = config["transactions"] @@ -30,7 +31,9 @@ def extract(self, file, existing_entries): return result - def createForDate(self, trx, date): + def createForDate( + self, trx: dict[str, Any], date: datetime.date + ) -> data.Transaction: postings = [] for post in trx["postings"]: amt = None diff --git a/src/tariochbctools/importers/swisscard/importer.py b/src/tariochbctools/importers/swisscard/importer.py index 0cfb6a3..3f60473 100644 --- a/src/tariochbctools/importers/swisscard/importer.py +++ b/src/tariochbctools/importers/swisscard/importer.py @@ -1,29 +1,31 @@ import csv -from io import StringIO +import re from beancount.core import amount, data from beancount.core.number import D -from beancount.ingest import importer -from beancount.ingest.importers.mixins import identifier +from beangulp import Importer from dateutil.parser import parse -class SwisscardImporter(identifier.IdentifyMixin, importer.ImporterProtocol): +class SwisscardImporter(Importer): """An importer for Swisscard's cashback CSV files.""" - def __init__(self, regexps, account): - identifier.IdentifyMixin.__init__(self, matchers=[("filename", regexps)]) - self.account = account + def __init__(self, filepattern: str, account: data.Account): + self._filepattern = filepattern + self._account = account - def name(self): - return super().name() + self.account + def name(self) -> str: + return super().name() + self._account - def file_account(self, file): - return self.account + def identify(self, filepath: str) -> bool: + return re.search(self._filepattern, filepath) is not None - def extract(self, file, existing_entries): + def account(self, filepath: str) -> data.Account: + return self._account + + def extract(self, filepath: str, existing: data.Entries) -> data.Entries: entries = [] - with StringIO(file.contents()) as csvfile: + with open(filepath) as csvfile: reader = csv.DictReader( csvfile, delimiter=",", @@ -36,7 +38,7 @@ def extract(self, file, existing_entries): "merchant": row["Merchant Category"], "category": row["Registered Category"], } - meta = data.new_metadata(file.name, 0, metakv) + meta = data.new_metadata(filepath, 0, metakv) description = row["Description"].strip() entry = data.Transaction( meta, @@ -47,7 +49,7 @@ def extract(self, file, existing_entries): data.EMPTY_SET, data.EMPTY_SET, [ - data.Posting(self.account, amt, None, None, None, None), + data.Posting(self._account, amt, None, None, None, None), ], ) entries.append(entry) diff --git a/src/tariochbctools/importers/transferwise/importer.py b/src/tariochbctools/importers/transferwise/importer.py index 24cdb5b..a06f446 100644 --- a/src/tariochbctools/importers/transferwise/importer.py +++ b/src/tariochbctools/importers/transferwise/importer.py @@ -4,6 +4,7 @@ from os import path from urllib.parse import urlencode +import beangulp import dateutil.parser import requests import rsa @@ -11,19 +12,18 @@ import yaml from beancount.core import amount, data from beancount.core.number import D -from beancount.ingest import importer from dateutil.relativedelta import relativedelta http = urllib3.PoolManager() -class Importer(importer.ImporterProtocol): +class Importer(beangulp.Importer): """An importer for Transferwise using the API.""" - def identify(self, file): - return path.basename(file.name).endswith("transferwise.yaml") + def identify(self, filepath): + return path.basename(filepath).endswith("transferwise.yaml") - def file_account(self, file): + def account(self, filepath): return "" def __init__(self, *args, **kwargs): @@ -115,8 +115,8 @@ def _do_sca_challenge(self): return signature - def extract(self, file, existing_entries): - with open(file.name, "r") as f: + def extract(self, filepath, existing): + with open(filepath, "r") as f: config = yaml.safe_load(f) self.api_token = config["token"] baseAccount = config["baseAccount"] diff --git a/src/tariochbctools/importers/truelayer/importer.py b/src/tariochbctools/importers/truelayer/importer.py index daa5974..ec490c1 100644 --- a/src/tariochbctools/importers/truelayer/importer.py +++ b/src/tariochbctools/importers/truelayer/importer.py @@ -2,12 +2,12 @@ from datetime import timedelta from os import path +import beangulp import dateutil.parser import requests import yaml from beancount.core import amount, data from beancount.core.number import D -from beancount.ingest import importer # https://docs.truelayer.com/#retrieve-account-transactions @@ -24,7 +24,7 @@ ) -class Importer(importer.ImporterProtocol): +class Importer(beangulp.Importer): """An importer for Truelayer API (e.g. for Revolut).""" def __init__(self): @@ -33,17 +33,17 @@ def __init__(self): self.clientSecret = None self.refreshToken = None self.sandbox = None - self.existing_entries = None + self.existing = None self.domain = "truelayer.com" - def _configure(self, file, existing_entries): - with open(file.name, "r") as f: + def _configure(self, filepath, existing): + with open(filepath, "r") as f: self.config = yaml.safe_load(f) self.clientId = self.config["client_id"] self.clientSecret = self.config["client_secret"] self.refreshToken = self.config["refresh_token"] self.sandbox = self.clientId.startswith("sandbox") - self.existing_entries = existing_entries + self.existing = existing if self.sandbox: self.domain = "truelayer-sandbox.com" @@ -51,14 +51,14 @@ def _configure(self, file, existing_entries): if "account" not in self.config and "accounts" not in self.config: raise KeyError("At least one of `account` or `accounts` must be specified") - def identify(self, file): - return path.basename(file.name).endswith("truelayer.yaml") + def identify(self, filepath): + return path.basename(filepath).endswith("truelayer.yaml") - def file_account(self, file): + def account(self, filepath): return "" - def extract(self, file, existing_entries=None): - self._configure(file, existing_entries) + def extract(self, filepath, existing=None): + self._configure(filepath, existing) r = requests.post( f"https://auth.{self.domain}/connect/token", @@ -187,8 +187,8 @@ def _extract_transaction(self, trx, local_account, transactions, invert_sign): if trx["transaction_id"] == transactions[-1]["transaction_id"]: balDate = trxDate + timedelta(days=1) metakv = {} - if self.existing_entries is not None: - for exEntry in self.existing_entries: + if self.existing is not None: + for exEntry in self.existing: if ( isinstance(exEntry, data.Balance) and exEntry.date == balDate diff --git a/src/tariochbctools/importers/viseca/importer.py b/src/tariochbctools/importers/viseca/importer.py index 03ec86d..63aacf0 100644 --- a/src/tariochbctools/importers/viseca/importer.py +++ b/src/tariochbctools/importers/viseca/importer.py @@ -1,26 +1,32 @@ import re from datetime import datetime +import beangulp import camelot from beancount.core import amount, data from beancount.core.number import D -from beancount.ingest import importer -from beancount.ingest.importers.mixins import identifier -class Importer(identifier.IdentifyMixin, importer.ImporterProtocol): +class Importer(beangulp.Importer): """An importer for Viseca One Card Statement PDF files.""" - def __init__(self, regexps, account): - identifier.IdentifyMixin.__init__(self, matchers=[("filename", regexps)]) - self.account = account + def __init__(self, filepattern: str, account: data.Account): + self._filepattern = filepattern + self._account = account self.currency = "CHF" - def file_account(self, file): - return self.account + def identify(self, filepath: str) -> bool: + return re.search(self._filepattern, filepath) is not None - def createEntry(self, file, date, entryAmount, text): + def account(self, filepath: str) -> data.Account: + return self._account + + def createEntry( + self, filepath: str, date: str, entryAmount: str | None, text: str + ) -> data.Transaction: amt = None + if not entryAmount: + entryAmount = "" entryAmount = entryAmount.replace("'", "") if "-" in entryAmount: amt = amount.Amount(D(entryAmount.strip(" -")), "CHF") @@ -29,7 +35,7 @@ def createEntry(self, file, date, entryAmount, text): book_date = datetime.strptime(date, "%d.%m.%y").date() - meta = data.new_metadata(file.name, 0) + meta = data.new_metadata(filepath, 0) return data.Transaction( meta, book_date, @@ -39,11 +45,11 @@ def createEntry(self, file, date, entryAmount, text): data.EMPTY_SET, data.EMPTY_SET, [ - data.Posting(self.account, amt, None, None, None, None), + data.Posting(self._account, amt, None, None, None, None), ], ) - def extract(self, file, existing_entries): + def extract(self, filepath: str, existing: data.Entries) -> data.Entries: entries = [] p = re.compile(r"^\d\d\.\d\d\.\d\d$") @@ -51,7 +57,7 @@ def extract(self, file, existing_entries): columns = ["100,132,400,472,523"] firstPageTables = camelot.read_pdf( - file.name, + filepath, flavor="stream", pages="1", table_regions=["65,450,585,50"], @@ -59,7 +65,7 @@ def extract(self, file, existing_entries): split_text=True, ) otherPageTables = camelot.read_pdf( - file.name, + filepath, flavor="stream", pages="2-end", table_regions=["65,650,585,50"], @@ -91,7 +97,9 @@ def extract(self, file, existing_entries): if amountChf: if lastTrxDate: entries.append( - self.createEntry(file, lastTrxDate, lastAmount, lastDetails) + self.createEntry( + filepath, lastTrxDate, lastAmount, lastDetails + ) ) lastTrxDate = trxDate @@ -102,7 +110,7 @@ def extract(self, file, existing_entries): if lastTrxDate: entries.append( - self.createEntry(file, lastTrxDate, lastAmount, lastDetails) + self.createEntry(filepath, lastTrxDate, lastAmount, lastDetails) ) return entries diff --git a/src/tariochbctools/importers/zak/importer.py b/src/tariochbctools/importers/zak/importer.py index ba22fda..fa822a2 100644 --- a/src/tariochbctools/importers/zak/importer.py +++ b/src/tariochbctools/importers/zak/importer.py @@ -1,32 +1,37 @@ import re from datetime import timedelta +import beangulp import camelot import pandas as pd from beancount.core import amount, data from beancount.core.number import D -from beancount.ingest import importer -from beancount.ingest.importers.mixins import identifier from dateutil.parser import parse -class Importer(identifier.IdentifyMixin, importer.ImporterProtocol): +class Importer(beangulp.Importer): """An importer for Bank Cler ZAK PDF files files.""" - def __init__(self, regexps, account): - identifier.IdentifyMixin.__init__(self, matchers=[("filename", regexps)]) - self.account = account + def __init__(self, filepattern: str, account: data.Account): + self._filepattern = filepattern + self._account = account - def file_account(self, file): - return self.account + def account(self, filepath: str) -> data.Account: + return self._account - def createEntry(self, file, date, amt, text): + def identify(self, filepath: str) -> bool: + return re.search(self._filepattern, filepath) is not None + + def createEntry( + self, filepath: str, date: str, amt: str, text: str + ) -> data.Transaction: bookingNrRgexp = re.compile(r"BC Buchungsnr. (?P\d+)$") m = bookingNrRgexp.search(text) - bookingRef = m.group("bookingRef") - text = re.sub(bookingNrRgexp, "", text) + if m: + bookingRef = m.group("bookingRef") + text = re.sub(bookingNrRgexp, "", text) - meta = data.new_metadata(file.name, 0, {"zakref": bookingRef}) + meta = data.new_metadata(filepath, 0, {"zakref": bookingRef}) return data.Transaction( meta, parse(date.strip(), dayfirst=True).date(), @@ -37,41 +42,41 @@ def createEntry(self, file, date, amt, text): data.EMPTY_SET, [ data.Posting( - self.account, amount.Amount(D(amt), "CHF"), None, None, None, None + self._account, amount.Amount(D(amt), "CHF"), None, None, None, None ), ], ) - def createBalanceEntry(self, file, date, amt): - meta = data.new_metadata(file.name, 0) + def createBalanceEntry(self, filepath: str, date: str, amt: str) -> data.Balance: + meta = data.new_metadata(filepath, 0) return data.Balance( meta, parse(date.strip(), dayfirst=True).date() + timedelta(days=1), - self.account, + self._account, amount.Amount(D(amt), "CHF"), None, None, ) - def cleanNumber(self, number): + def cleanNumber(self, number: str | data.Decimal) -> data.Decimal: if isinstance(number, str): return D(number.replace("'", "")) else: return number - def extract(self, file, existing_entries): + def extract(self, filepath: str, existing: data.Entries) -> data.Entries: entries = [] firstPageTables = camelot.read_pdf( - file.name, flavor="stream", pages="1", table_regions=["60,450,600,170"] + filepath, flavor="stream", pages="1", table_regions=["60,450,600,170"] ) otherPageTables = camelot.read_pdf( - file.name, flavor="stream", pages="2-end", table_regions=["60,630,600,170"] + filepath, flavor="stream", pages="2-end", table_regions=["60,630,600,170"] ) tables = [*firstPageTables, *otherPageTables] - df = None + df: pd.DataFrame | None = None for table in tables: cur_df = table.df new_header = cur_df.iloc[0] @@ -87,36 +92,38 @@ def extract(self, file, existing_entries): text = "" amount = None saldo = None - for row in df.itertuples(): - if row.Saldo: - if date and amount: - entries.append(self.createEntry(file, date, amount, text)) + if df: + for row in df.itertuples(): + if row.Saldo: + if date and amount: + entries.append(self.createEntry(filepath, date, amount, text)) - date = None - amount = None - text = "" + date = None + amount = None + text = "" - if row.Valuta: - date = row.Valuta + if row.Valuta: + date = row.Valuta - if row.Text: - text += " " + row.Text + if row.Text: + text += " " + row.Text - if row.Belastung: - amount = -self.cleanNumber(row.Belastung) + if row.Belastung: + amount = -self.cleanNumber(row.Belastung) - if row.Gutschrift: - amount = self.cleanNumber(row.Gutschrift) + if row.Gutschrift: + amount = self.cleanNumber(row.Gutschrift) - if row.Saldo: - saldo = self.cleanNumber(row.Saldo) + if row.Saldo: + saldo = self.cleanNumber(row.Saldo) - if date and amount: - entries.append(self.createEntry(file, date, amount, text)) + if date and amount: + entries.append(self.createEntry(filepath, date, amount, text)) - dateRegexp = re.compile(r"\d\d\.\d\d\.\d\d\d\d") - m = dateRegexp.search(text) - date = m.group() - entries.append(self.createBalanceEntry(file, date, saldo)) + dateRegexp = re.compile(r"\d\d\.\d\d\.\d\d\d\d") + m = dateRegexp.search(text) + if m and saldo: + date = m.group() + entries.append(self.createBalanceEntry(filepath, date, saldo)) return entries diff --git a/src/tariochbctools/importers/zkb/importer.py b/src/tariochbctools/importers/zkb/importer.py index 06b4a72..d69b493 100644 --- a/src/tariochbctools/importers/zkb/importer.py +++ b/src/tariochbctools/importers/zkb/importer.py @@ -1,13 +1,14 @@ import re +from typing import Any from tariochbctools.importers.general import mt940importer class ZkbImporter(mt940importer.Importer): - def prepare_payee(self, trxdata): + def prepare_payee(self, trxdata: dict[str, Any]) -> str: return "" - def prepare_narration(self, trxdata): + def prepare_narration(self, trxdata: dict[str, Any]) -> str: extra = trxdata["extra_details"] details = trxdata["transaction_details"] diff --git a/src/tariochbctools/plugins/prices/ibkr.py b/src/tariochbctools/plugins/prices/ibkr.py index 4c326cd..52c046b 100644 --- a/src/tariochbctools/plugins/prices/ibkr.py +++ b/src/tariochbctools/plugins/prices/ibkr.py @@ -1,15 +1,16 @@ from datetime import datetime from os import environ from time import sleep +from typing import Optional from beancount.core.number import D -from beancount.prices import source +from beanprice import source from dateutil import tz from ibflex import client, parser class Source(source.Source): - def get_latest_price(self, ticker: str): + def get_latest_price(self, ticker: str) -> source.SourcePrice | None: token: str = environ["IBKR_TOKEN"] queryId: str = environ["IBKR_QUERY_ID"] @@ -36,5 +37,7 @@ def get_latest_price(self, ticker: str): return None - def get_historical_price(self, ticker, time): + def get_historical_price( + self, ticker: str, time: datetime + ) -> Optional[source.SourcePrice]: return None diff --git a/tests/tariochbctools/importers/test_quickfile.py b/tests/tariochbctools/importers/test_quickfile.py index 6736f64..082ddf7 100644 --- a/tests/tariochbctools/importers/test_quickfile.py +++ b/tests/tariochbctools/importers/test_quickfile.py @@ -4,7 +4,6 @@ import pytest from beancount.core import data from beancount.core.number import D -from beancount.ingest import cache from tariochbctools.importers.quickfile import importer as qfimp @@ -83,7 +82,7 @@ def tmp_config_fixture(tmp_path): config = tmp_path / "quickfile.yaml" config.write_bytes(TEST_CONFIG) - yield cache.get_file(config) # a FileMemo, not a Path + yield config @pytest.fixture(name="importer") @@ -103,7 +102,7 @@ def _importer_with_config(custom_config): config = tmp_path / "quickfile.yaml" config.write_bytes(custom_config) importer = qfimp.Importer() - importer._configure(cache.get_file(config), []) + importer._configure(config, []) return importer yield _importer_with_config diff --git a/tests/tariochbctools/importers/test_truelayer.py b/tests/tariochbctools/importers/test_truelayer.py index 5829865..b1476ca 100644 --- a/tests/tariochbctools/importers/test_truelayer.py +++ b/tests/tariochbctools/importers/test_truelayer.py @@ -3,7 +3,6 @@ import pytest import yaml from beancount.core.amount import Decimal as D -from beancount.ingest import cache from tariochbctools.importers.truelayer import importer as tlimp @@ -73,7 +72,7 @@ def tmp_config_fixture(tmp_path): config = tmp_path / "truelayer.yaml" config.write_bytes(TEST_CONFIG) - yield cache.get_file(config) # a FileMemo, not a Path + yield config @pytest.fixture(name="importer") @@ -93,7 +92,7 @@ def _importer_with_config(custom_config): config = tmp_path / "truelayer.yaml" config.write_bytes(custom_config) importer = tlimp.Importer() - importer._configure(cache.get_file(config), []) + importer._configure(config, []) return importer yield _importer_with_config