path: root/beancount_extras_kris7t/importers/transferwise/client.py

'''
Importer for Transferwise API transaction history from the command line.
'''
__copyright__ = 'Copyright (c) 2020  Kristóf Marussy <kristof@marussy.com>'
__license__ = 'GNU GPLv2'

import datetime as dt
import logging
import os
from typing import Any, Dict, Optional, Tuple, Set

import beancount
from beancount.core import data

from beancount_extras_kris7t.importers.transferwise.transferwise_json import Accounts, \
        DATE_FORMAT, Importer

LOG = logging.getLogger('importers.transferwise.client')


def _parse_date_arg(date_str: str) -> dt.date:
    return dt.datetime.strptime(date_str, '%Y-%m-%d').date()


def _import_config(config_path: str) -> Importer:
    import runpy

    config = runpy.run_path(config_path)
    importer = config['TRANSFERWISE_CONFIG']  # type: ignore
    if isinstance(importer, Importer):
        LOG.info('Loaded configuration from %s', config_path)
        return importer
    else:
        raise ValueError(f'Invalid configuration: {config_path}')


def _get_reference(transaction: data.Transaction) -> Optional[str]:
    for link in transaction.links:
        if link.startswith('transferwise_'):
            return link[13:]
    return None


def _get_last_transaction_date(ledger_path: str, skip_references: Set[str]) -> Optional[dt.date]:
    from beancount.parser import parser

    LOG.info('Checking %s for already imported transactions', ledger_path)
    entries, _, _ = parser.parse_file(ledger_path)
    date: Optional[dt.date] = None
    skip: Set[str] = set()
    for entry in entries:
        if isinstance(entry, data.Transaction):
            reference = _get_reference(entry)
            if not reference:
                continue
            if date is None or date < entry.date:
                date = entry.date
                skip.clear()
            if date == entry.date:
                skip.add(reference)
    skip_references.update(skip)
    return date


def _get_date_range(from_date: Optional[dt.date],
                    to_date: dt.date,
                    ledger_path: Optional[str]) -> Tuple[dt.date, dt.date, Set[str]]:
    skip_references = set()
    if not from_date and ledger_path:
        from_date = _get_last_transaction_date(ledger_path, skip_references)
    if not from_date:
        from_date = to_date - dt.timedelta(days=365)
    LOG.info('Fetching transactions from %s to %s', from_date, to_date)
    return from_date, to_date, skip_references


def _get_secrets(importer: Importer,
                 api_key: Optional[str],
                 proxy_uri: Optional[str]) -> Tuple[str, Optional[str]]:
    import urllib.parse

    if proxy_uri:
        uri_parts = urllib.parse.urlsplit(proxy_uri)
    else:
        uri_parts = None
    if api_key and (not uri_parts or not uri_parts.username or uri_parts.password):
        return api_key, proxy_uri

    from contextlib import closing

    import secretstorage

    with closing(secretstorage.dbus_init()) as connection:
        collection = secretstorage.get_default_collection(connection)
        if not api_key:
            items = collection.search_items({
                'profile_id': str(importer.profile_id),
                'borderless_account_id': str(importer.borderless_account_id),
                'xdg:schema': 'com.marussy.beancount.importer.TransferwiseAPIKey',
            })
            item = next(items, None)
            if not item:
                raise ValueError('No API key found in SecretService')
            LOG.info('Found API key secret "%s" from SecretService', item.get_label())
            api_key = item.get_secret().decode('utf-8')
        if uri_parts and uri_parts.username and not uri_parts.password:
            host = uri_parts.hostname or uri_parts.netloc
            items = collection.search_items({
                'host': host,
                'port': str(uri_parts.port or 1080),
                'user': uri_parts.username,
                'xdg:schema': 'org.freedesktop.Secret.Generic',
            })
            item = next(items, None)
            if item:
                LOG.info('Found proxy password secret "%s" from SecretService', item.get_label())
                password = urllib.parse.quote_from_bytes(item.get_secret())
                uri = f'{uri_parts.scheme}://{uri_parts.username}:{password}@{host}'
                if uri_parts.port:
                    proxy_uri = f'{uri}:{uri_parts.port}'
                else:
                    proxy_uri = uri
            else:
                LOG.info('No proxy password secret was found in SecretService')
    assert api_key  # Make pyright happy
    return api_key, proxy_uri


def _fetch_statements(importer: Importer,
                      from_date: dt.date,
                      to_date: dt.date,
                      api_key: str,
                      proxy_uri: Optional[str]) -> Dict[str, Any]:
    import json

    import requests

    now = dt.datetime.utcnow().time()
    from_time_str = dt.datetime.combine(from_date, dt.datetime.min.time()).strftime(DATE_FORMAT)
    to_time_str = dt.datetime.combine(to_date, now).strftime(DATE_FORMAT)
    uri_prefix = f'https://api.transferwise.com/v3/profiles/{importer.profile_id}/' + \
        f'borderless-accounts/{importer.borderless_account_id}/statement.json' + \
        f'?intervalStart={from_time_str}&intervalEnd={to_time_str}&type=COMPACT&currency='
    headers = {
        'User-Agent': f'Beancount {beancount.__version__} Transferwise importer {__copyright__}',
        'Authorization': f'Bearer {api_key}',
    }
    proxy_dict: Dict[str, str] = {}
    if proxy_uri:
        proxy_dict['https'] = proxy_uri
    statements: Dict[str, Any] = {}
    for currency in importer.currencies:
        result = requests.get(uri_prefix + currency, headers=headers, proxies=proxy_dict)
        if result.status_code != 200:
            LOG.error(
                'Fetcing %s statement failed with HTTP status code %d: %s',
                currency,
                result.status_code,
                result.text)
        else:
            try:
                statement = json.loads(result.text)
            except json.JSONDecodeError as exc:
                LOG.error('Failed to decode %s statement', currency, exc_info=exc)
            else:
                statements[currency] = statement
                LOG.info('Fetched %s statement', currency)
    return statements


def _print_statements(from_date: dt.date,
                      to_date: dt.date,
                      entries: data.Entries) -> None:
    from beancount.parser import printer
    print(f'*** Transferwise from {from_date} to {to_date}', flush=True)
    printer.print_entries(entries)


def _determine_path(dir: str, currency: str, to_date: dt.date) -> str:
    date_str = to_date.strftime('%Y-%m-%d')
    simple_path = os.path.join(dir, f'{date_str}.transferwise_{currency}.json')
    if not os.path.exists(simple_path):
        return simple_path
    for i in range(2, 10):
        path = os.path.join(dir, f'{date_str}.transferwise_{currency}_{i}.json')
        if not os.path.exists(path):
            return path
    raise ValueError(f'Cannot find unused name for {simple_path}')


def _archive_statements(documents_path: str,
                        to_date: dt.date,
                        accounts: Accounts,
                        statements: Dict[str, Any]):
    import json

    from beancount.core.account import sep

    for currency, statement in statements.items():
        dir = os.path.join(documents_path, *accounts.get_borderless_account(currency).split(sep))
        os.makedirs(dir, exist_ok=True)
        path = _determine_path(dir, currency, to_date)
        with open(path, 'w') as file:
            json.dump(statement, file, indent=2)
        LOG.info('Saved %s statement as %s', currency, path)


def main():
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument('--verbose', '-v', action='store_true')
    parser.add_argument('--from-date', '-f', required=False, type=_parse_date_arg)
    parser.add_argument('--to-date', '-t', default=dt.date.today(), type=_parse_date_arg)
    parser.add_argument('--api-key', '-k', required=False, type=str)
    parser.add_argument('--proxy', '-p', required=False, type=str)
    parser.add_argument('--archive', '-a', required=False, type=str)
    parser.add_argument('config', nargs=1)
    parser.add_argument('ledger', nargs='?')
    args = parser.parse_args()
    if args.verbose:
        log_level = logging.INFO
    else:
        log_level = logging.WARN
    logging.basicConfig(level=log_level)
    importer = _import_config(args.config[0])
    from_date, to_date, skip = _get_date_range(args.from_date, args.to_date, args.ledger)
    if skip:
        LOG.info('Skipping %s', skip)
    api_key, proxy_uri = _get_secrets(importer, args.api_key, args.proxy)
    statements = _fetch_statements(importer, from_date, to_date, api_key, proxy_uri)
    if args.archive:
        _archive_statements(args.archive, to_date, importer.accounts, statements)
    entries = importer.extract_objects(statements.values(), skip)
    if entries:
        _print_statements(from_date, to_date, entries)