Luís Murta
fd6793b4f4
and as a result, had to fix a LOT of minor potential future issue. It also reorders and clears unused imports. When exporting transactions, it will sort by date.
136 lines
3.6 KiB
Python
136 lines
3.6 KiB
Python
import datetime as dt
|
|
import dotenv
|
|
import json
|
|
import nordigen
|
|
import os
|
|
import requests
|
|
import time
|
|
import uuid
|
|
|
|
import pfbudget.db.model as t
|
|
import pfbudget.utils as utils
|
|
|
|
from .input import Input
|
|
|
|
dotenv.load_dotenv()
|
|
|
|
|
|
class NordigenInput(Input):
|
|
redirect_url = "https://murta.dev"
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
if not (key := os.environ.get("SECRET_KEY")) or not (
|
|
id := os.environ.get("SECRET_ID")
|
|
):
|
|
raise
|
|
|
|
self._client = nordigen.NordigenClient(
|
|
secret_key=key,
|
|
secret_id=id,
|
|
)
|
|
|
|
self._client.token = self.__token()
|
|
self._start = dt.date.min
|
|
self._end = dt.date.max
|
|
|
|
def parse(self) -> list[t.BankTransaction]:
|
|
transactions = []
|
|
assert len(self._banks) > 0
|
|
|
|
for bank in self._banks:
|
|
print(f"Downloading from {bank}...")
|
|
requisition = self.client.requisition.get_requisition_by_id(
|
|
bank.nordigen.requisition_id
|
|
)
|
|
|
|
print(requisition)
|
|
for acc in requisition["accounts"]:
|
|
account = self._client.account_api(acc)
|
|
|
|
retries = 0
|
|
downloaded = {}
|
|
while retries < 3:
|
|
try:
|
|
downloaded = account.get_transactions()
|
|
break
|
|
except requests.ReadTimeout:
|
|
retries += 1
|
|
print(f"Request #{retries} timed-out, retrying in 1s")
|
|
time.sleep(1)
|
|
except requests.HTTPError as e:
|
|
retries += 1
|
|
print(f"Request #{retries} failed with {e}, retrying in 1s")
|
|
time.sleep(1)
|
|
|
|
if not downloaded:
|
|
print(f"Couldn't download transactions for {account}")
|
|
continue
|
|
|
|
with open("json/" + bank.name + ".json", "w") as f:
|
|
json.dump(downloaded, f)
|
|
|
|
converted = [
|
|
utils.convert(t, bank) for t in downloaded["transactions"]["booked"]
|
|
]
|
|
|
|
transactions.extend(
|
|
[t for t in converted if self._start <= t.date <= self._end]
|
|
)
|
|
|
|
return sorted(transactions)
|
|
|
|
def token(self):
|
|
token = self._client.generate_token()
|
|
print(f"New access token: {token}")
|
|
return token
|
|
|
|
def requisition(self, institution: str, country: str = "PT"):
|
|
id = self._client.institution.get_institution_id_by_name(country, institution)
|
|
requisition = self._client.initialize_session(
|
|
redirect_uri=self.redirect_url,
|
|
institution_id=id,
|
|
reference_id=str(uuid.uuid4()),
|
|
)
|
|
return requisition.link, requisition.requisition_id
|
|
|
|
def country_banks(self, country: str):
|
|
return self._client.institution.get_institutions(country)
|
|
|
|
@property
|
|
def client(self):
|
|
return self._client
|
|
|
|
@property
|
|
def banks(self):
|
|
return self._banks
|
|
|
|
@banks.setter
|
|
def banks(self, value):
|
|
self._banks = value
|
|
|
|
@property
|
|
def start(self):
|
|
return self._start
|
|
|
|
@start.setter
|
|
def start(self, value):
|
|
self._start = value
|
|
|
|
@property
|
|
def end(self):
|
|
return self._end
|
|
|
|
@end.setter
|
|
def end(self, value):
|
|
self._end = value
|
|
|
|
def __token(self):
|
|
if token := os.environ.get("TOKEN"):
|
|
return token
|
|
else:
|
|
token = self._client.generate_token()
|
|
print(f"New access token: {token}")
|
|
return token["access"]
|