fetch-transactions

#!/usr/bin/env python
"""
This script fetches JSON from nordigen
"""

import argparse
import csv
import datetime
import logging
import json
import sys
import requests
import re
from pathlib import Path
from pprint import pprint
from uuid import uuid4
from nordigen import NordigenClient

root = Path(__file__).parents[1]

session = {
    'refresh_token': ''
}
session_file = None
transaction_keys = [
    'internal_reference', # Must be 1st!
    'date_transaction',
    'date_book',
    'external-id',
    'account-id',
    'opposing-name',
    'opposing-iban',
    'amount',
    'currency-code',
    'tags-comma',
    'description',
]

transaction_import_config = {
    "version": 3,
    "date": "Y-m-d",
    "default_account": 0, # ! updates below !
    "delimiter": "comma",
    "headers": True,
    "rules": True,
    "add_import_tag": True,
    "roles": transaction_keys,
    "mapping": [],
    "duplicate_detection_method": "cell",
    "ignore_duplicate_lines": True,
    "unique_column_index": 0,
    "unique_column_type": "internal_reference",
    "flow": "file",
    "map_all_data": True,
    "accounts": [],
    "date_range": "",
    "date_range_number": 365, # ! updates below !
    "date_range_unit": "d", # ! updates below !
    "date_not_before": "",
    "date_not_after": "",
    "nordigen_country": "",
    "nordigen_bank": "",
    "nordigen_requisitions": [],
    "nordigen_max_days": "90",
    "conversion": False,
    "ignore_duplicate_transactions": True,
}

iso_date_regex = re.compile(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}')

# Max errors per service, to throw an error
max_errors = 5

def session_save():
    with open(str(session_file), 'w') as fp:
        logging.info("Save config to %s", session_file)
        json.dump(session, fp)

def json_save(filename, data):
    with open(filename, "w") as outfile:
        outfile.write(json.dumps(data, indent=4))

def filter_field(data, field_name, search_string):
    if field_name in data:
        if data[field_name] == search_string:
            del data[field_name]
    return data

def main():
    """
        Entry point
    """
    global session
    global session_file
    global config

    parser = argparse.ArgumentParser(
        description='Fetch nordigen JSON'
    )

    parser.add_argument(
        '-l', '--log',
        choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
        default='INFO',
        help='set log level',
    )

    parser.add_argument(
        '--days',
        default=60,
        type=int,
        help='days to fetch',
    )

    parser.add_argument(
        '--cfg-dir',
        default=Path("cfg"),
        help='path to get json config files and store session',
    )

    parser.add_argument(
        '--data-dir',
        default=Path("data"),
        help='path to get json config files and store session',
    )

    parser.add_argument(
        '--test',
        action='store_true',
        help='Do not contact Nordigen, re-use previous data',
    )

    args = parser.parse_args()

    log_level = args.log
    logging.basicConfig(level=log_level)

    logging.debug("Config path %s", args.cfg_dir)
    config_dir = Path(args.cfg_dir)

    data_dir = Path(args.data_dir)
    data_dir.mkdir(parents=True, exist_ok=True)

    # Prepare filter fields
    date_to = datetime.datetime.now(datetime.timezone.utc)
    date_from = date_to - datetime.timedelta(days=args.days)
    date_with_month = date_to.strftime("%Y-%m")
    date_with_hour = date_to.strftime("%Y-%m-%d-%H")

    logging.info(
        "Fetching from %s to %s",
        date_from.strftime("%Y-%m-%d"),
        date_to.strftime("%Y-%m-%d"),
    )

    # Check - do we have session valid?

    session_file = Path(config_dir) / "session.json"

    # Load session file (if present)
    if session_file.exists():
        session = json.load(open(session_file))

    if 'account' not in session: session['account'] = {}

    # Load accounts
    config_nordigen = json.load(open(config_dir / "nordigen.json", "r"))
    config_firefly = json.load(open(config_dir / "firefly.json", "r"))

    # Fill some default account (it does not required, as we set if in each transaction,
    # but this is required by importer)
    transaction_import_config['default_account'] = list(config_nordigen['accounts'].values())[0]['firefly_account_id']
    # Set range limit
    transaction_import_config['date_range_number'] = args.days
    transaction_import_config['date_range_unit'] = "d"

    if not args.test:
        # Establish Nordigen
        client = NordigenClient(
            secret_id = config_nordigen['id'],
            secret_key = config_nordigen['key'],
            timeout = 120,
        )
        token_info = client.generate_token()
        session['access'] = token_info['access']
        session_save()
    else:
        logging.warning("Test mode activated, will use prevous data!")

    data = []

    account_errors = []

    # Load transactions
    for account_id, account in config_nordigen['accounts'].items():
        logging.info("Processing: %s", account_id)
        if account_id not in session['account']: session['account'][account_id] = {'status':''}
        if 'connection_errors' not in session['account'][account_id]: session['account'][account_id]['connection_errors'] = 0
        # Check account status
        account_status = session['account'][account_id].get('status', None)
        connection_url = f"https://bankaccountdata.gocardless.com/psd2/start/{account['id']}/{account['bank_id']}"

        if account_status == 'suspend':
            # Account suspended, skip it
            logging.error("Account suspended! Skipping")
            account_errors.append(f"Account {account_id}: suspended status, need to be checked and status cleared, connection url: {connection_url}")
            continue

        if not args.test:
            account_api = client.account_api(id=account['account'])
            try:
                transactions_api = account_api.get_transactions(
                    date_from=date_from.strftime("%Y-%m-%d"),
                    date_to=date_to.strftime("%Y-%m-%d")
                )
            except requests.exceptions.HTTPError as e:
                logging.error("HTTP error: %s", str(e))

                if e.response.status_code == 503:
                    # Institution service unavailable
                    logging.error("Service unavailable")
                    session['account'][account_id]['connection_errors'] = session['account'][account_id]['connection_errors'] + 1

                    if session['account'][account_id]['connection_errors'] >= max_errors:
                        logging.error(f"Max erros reached for {account_id}")
                        exit(43)
                    session_save()
                    continue

                if e.response.status_code == 409:
                    # Account suspended!
                    logging.error("Account suspended!")
                    session['account'][account_id]['status'] = 'suspend'
                    session_save()
                    exit(42)

                if e.response.json()['type'] == 'AccessExpiredError':
                    # We have expired token!
                    logging.error("Token expired! We have to get new connection!")
                    session['account'][account_id]['status'] = 'suspend'
                    session_save()
                    # Request new connection link
                    logging.error(f"New connection: {connection_url}")
                    exit(41)

                # Rethrow unknown
                raise e

            # We get something, reset error counter
            session['account'][account_id]['connection_errors'] = 0

            # Dump data (with month)
            json_save(data_dir / f"transactions-{date_with_month}-{account_id}.json", transactions_api)
            # Dump data (with month+day+hour)
            json_save(data_dir / f"transactions-{date_with_hour}-{account_id}.json", transactions_api)
            session_save()
        else:
            # Re-use previous dump
            transactions_api = json.load(open(data_dir / f"transactions-{date_with_month}-{account_id}.json"))

        for transaction_raw in transactions_api['transactions']['booked']:
            # Prepare empty object
            transaction_csv = {}
            for csv_key in transaction_keys: transaction_csv[csv_key] = ''
            # Add some tags, good for automation (e.g. transaction origin to filter duplicates)
            transaction_tags = []
            # Fetched via Nordigen
            transaction_tags.append(f"nordigen_{account_id}")
            # Fetched via API
            transaction_tags.append(f"api_{account_id}")
            # Common fields
            transaction_csv['date_transaction'] = transaction_raw['valueDate']
            transaction_csv['date_book'] = transaction_raw['bookingDate']
            transaction_csv['internal_reference'] = transaction_raw['internalTransactionId']
            transaction_csv['account-id'] = account['firefly_account_id']
            # Bank reference?
            if 'endToEndId' in transaction_raw:
                transaction_csv['external-id'] = transaction_raw['endToEndId']
            # Amount
            transaction_csv['amount'] = float(transaction_raw['transactionAmount']['amount'])
            # Skip empty ("informational") ones
            if transaction_csv['amount'] == 0:
                continue
            # Credit / debit and currency
            is_income = transaction_csv['amount'] >= 0
            transaction_csv['currency-code'] = transaction_raw['transactionAmount']['currency']
            # The 'opposite' account
            # We should use: Creditor = money OUT, Debitor = money IN
            field_name = 'creditor'
            if is_income: field_name = 'debtor'
            # Try to read other Name (ok, if missing)
            try:
                transaction_csv['opposing-name'] = transaction_raw[f"{field_name}Name"]
            except: pass
            # Try to read other IBAN (ok, if missing)
            try:
                transaction_csv['opposing-iban'] = transaction_raw[f"{field_name}Account"]['iban']
            except: pass

            # Filter out "-" in values (sent by N26)
            transaction_raw = filter_field(transaction_raw, 'remittanceInformationUnstructured', '-')

            # Description
            # Some of banks (N26) have only remittanceInformationUnstructured
            # , and some (Commerzbank) have more detailed remittanceInformationStructured
            if 'remittanceInformationStructured' in transaction_raw:
                transaction_csv['description'] = transaction_raw['remittanceInformationStructured']
            elif 'remittanceInformationUnstructured' in transaction_raw:
                transaction_csv['description'] = transaction_raw['remittanceInformationUnstructured']
            # Sometimes it could be evenpip missing!
            # Populate it with some opposite-account information
            elif 'opposing-name' in transaction_csv:
                transaction_csv['description'] = transaction_csv['opposing-name']
            elif 'opposing-iban' in transaction_csv:
                transaction_csv['description'] = transaction_csv['opposing-iban']

            # Search date / time from description, some banks (like Commerzbank) may
            # do like: NORDSEE GMBH - 1234 KARLSRUHE DEU 2023-06-01T09:19:15 Kartenzahlung
            iso_date_match = iso_date_regex.search(transaction_csv['description'])
            if iso_date_match:
                iso_date_match = iso_date_match.group(0)
                iso_date_match = datetime.datetime.fromisoformat(iso_date_match).strftime("%Y-%m-%d")

                if iso_date_match != transaction_csv['date_book']:
                    # logging.debug(
                    #     "Trnsaction has date in description: %s, date %s -> %s",
                    #     transaction_csv['description'],
                    #     transaction_csv['date_book'],
                    #     iso_date_match,
                    # )
                    transaction_csv['date_book'] = iso_date_match
                    transaction_csv['date_transaction'] = iso_date_match

            # Build tags
            transaction_csv['tags-comma'] = ','.join(transaction_tags)

            # Add record to common list
            data.append(transaction_csv)

    logging.info("Grabbed %d transactions", len(data))

    # Dump out files
    json_save(data_dir / 'import-config.json', transaction_import_config)
    with open(data_dir / 'import-data.csv', mode='w') as csv_file:
        writer = csv.DictWriter(csv_file, transaction_keys)
        writer.writeheader()
        writer.writerows(data)

    # Send transactions
    # Provide 2 files to importer: config json and csv data
    # See documentation: https://docs.firefly-iii.org/data-importer/advanced/automation/#automated-imports-using-the-web-post

    post_files = {
        'importable': open(data_dir / 'import-data.csv', 'rb'),
        'json': open(data_dir / 'import-config.json', 'rb'),
    }

    post_headers = {
        'accept': 'application/json',
        'authorization': f"Bearer {config_firefly['importer_client_key']}",
    }

    logging.info("Sending data to %s", config_firefly['importer_url'])

    resp = requests.post(
        url=f"{config_firefly['importer_url']}/autoupload?secret={config_firefly['importer_import_secret']}",
        files = post_files,
        headers=post_headers,
        verify=False, # Skip SSL check (self-signed cert)
    )

    logging.info("Response: %s", resp.reason)

    if not resp.ok:
        logging.info("Error sending data!")
        return 8

    # If we have erorrs recorded, say it
    if len(account_errors):
        logging.info("Errors in accounts detected")
        for err in account_errors:
            logging.info(err)
        return 9

    return 0

if __name__ == "__main__":
    sys.exit(main())