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

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

from collections import defaultdict
import datetime as dt
import json
import logging
import re
from os import path
from typing import Any, Callable, Dict, Iterable, List, NamedTuple, Optional, Set

from beancount.core import account, data
import beancount.core.amount as am
from beancount.core.amount import Amount
from beancount.core.flags import FLAG_WARNING
from beancount.core.inventory import Inventory
from beancount.core.number import ZERO
from beancount.ingest.cache import _FileMemo as FileMemo
from beancount.ingest.importer import ImporterProtocol

import beancount_extras_kris7t.importers.utils as utils
from beancount_extras_kris7t.importers.utils import COMMENT_META, InvalidEntry, PAYEE_META

DATE_FORMAT = '%Y-%m-%dT%H:%M:%SZ'
DATE_FORMAT_FRACTIONAL = '%Y-%m-%dT%H:%M:%S.%fZ'
CATEGORY_META = 'transferwise-category'
ENTRY_TYPE_META = 'transferwise-entry-type'
TRANSFERWISE_JSON_TAG = 'transferwise-json'
CD_DEBIT = 'DEBIT'
CD_CREDIT = 'CREDIT'
MONEY_ADDED_TYPE = 'MONEY_ADDED'
CARD_TYPE = 'CARD'
CARD_REGEX = re.compile(r'^Card transaction of.*issued by', re.IGNORECASE)


def _parse_date(date_str: str) -> dt.date:
    # TODO Handle time zones accurately
    try:
        return utils.parse_date(date_str, DATE_FORMAT)
    except InvalidEntry:
        return utils.parse_date(date_str, DATE_FORMAT_FRACTIONAL)


def _parse_json_amount(data: Any, cd: Optional[str] = None) -> Amount:
    # TODO Handle precision better
    amount = Amount(round(utils.parse_number(data['value']), 2), data['currency'])
    if cd == CD_DEBIT:
        return -amount
    else:
        return amount


def _validate_cd(amount: Amount, cd: str) -> None:
    if cd == CD_DEBIT:
        if amount.number >= ZERO:
            raise InvalidEntry(f'Invalid debit amount: {amount}')
    elif cd == CD_CREDIT:
        if amount.number <= ZERO:
            raise InvalidEntry(f'Invalid credit amount: {amount}')
    else:
        raise InvalidEntry(f'Invalid credit/debit type: {cd}')


class Reference(NamedTuple):
    type: str
    reference_number: str


class _ConversionResult(NamedTuple):
    converted_fraction: Amount
    price: Optional[Amount]
    fudge: Optional[Amount]


class _Conversion(NamedTuple):
    native_amount: Amount
    converted_amount: Optional[Amount]

    def get_fraction(self,
                     total_converted_amount: Amount,
                     assigned_amount: Amount,
                     diverted_fees: Inventory) -> _ConversionResult:
        fees = diverted_fees.get_currency_units(self.native_amount.currency)
        native_amount = am.add(self.native_amount, fees)
        fraction = am.div(assigned_amount, total_converted_amount.number).number
        if self.converted_amount:
            converted_fraction = am.mul(self.converted_amount, fraction)
            price = am.div(native_amount, self.converted_amount.number)
            assert price.number is not None
            price = price._replace(number=round(price.number, 4))
            # TODO Do we have to calculate the fudge here?
            return _ConversionResult(converted_fraction, price, None)
        else:
            return _ConversionResult(am.mul(native_amount, fraction), None, None)


class Accounts(NamedTuple):
    borderless_root_asset: str
    fees_expense: str

    def get_borderless_account(self, currency: str) -> str:
        return account.sep.join([self.borderless_root_asset, currency])


class Row(utils.Row):
    cd: str
    _transacted_amount: Amount
    date: dt.date
    _conversions: List[_Conversion]
    _inputs: Inventory
    _fees: Inventory
    divert_fees: Optional[str]

    def __init__(self, reference: Reference, transaction_list: List[Any]):
        assert len(transaction_list) >= 1
        first_transaction = transaction_list[0]
        details = first_transaction['details']
        entry_type = details.get('type', None)
        merchant = details.get('merchant', None)
        if merchant:
            payee = merchant['name']
        else:
            payee = None
        comment = details['description']
        super().__init__('<transferwise>', 0, entry_type, payee, comment)
        if entry_type:
            self.meta[ENTRY_TYPE_META] = entry_type
        category = details.get('category', None)
        if category:
            self.meta[CATEGORY_META] = category
        if merchant:
            if 'category' in merchant and merchant['category'] == category:
                del merchant['category']
            not_null_elements = [str(value).strip() for _, value in merchant.items() if value]
            self.meta[PAYEE_META] = '; '.join(not_null_elements)
        self.meta[COMMENT_META] = comment
        self.tags.add(TRANSFERWISE_JSON_TAG)
        self.links.add(f'transferwise_{reference.reference_number}')
        self.date = _parse_date(first_transaction['date'])
        self.cd = reference.type
        self.divert_fees = None
        self._conversions = []
        self._inputs = Inventory()
        self._fees = Inventory()
        self._compute_transacted_amount(transaction_list)
        for transaction in transaction_list:
            self._add_json_transaction(transaction)

    def _compute_transacted_amount(self, transaction_list: List[Any]) -> None:
        first_transaction = transaction_list[0]
        details = first_transaction['details']
        transacted_json = details.get('amount', None)
        if transacted_json:
            self._transacted_amount = _parse_json_amount(transacted_json, self.cd)
        else:
            if len(transaction_list) != 1:
                raise InvalidEntry('Cannot determine transaction amount')
            if exchange_details := first_transaction.get('exchangeDetails', None):
                self._transacted_amount = _parse_json_amount(exchange_details, self.cd)
            else:
                self._transacted_amount = _parse_json_amount(first_transaction['amount'])
            _validate_cd(self._transacted_amount, self.cd)

    def _add_json_transaction(self, transaction: Any) -> None:
        if exchange := transaction.get('exchangeDetails', None):
            native = _parse_json_amount(exchange['fromAmount'], self.cd)
            converted = _parse_json_amount(exchange['toAmount'], self.cd)
            _validate_cd(converted, self.cd)
        else:
            native = _parse_json_amount(transaction['amount'])
            converted = None
        _validate_cd(native, self.cd)
        if total_fees := transaction.get('totalFees', None):
            fee = -_parse_json_amount(total_fees)
            if fee.number > ZERO:
                raise InvalidEntry(f'Invalid transaction fee: {fee}')
            if fee.number != ZERO:
                self._fees.add_amount(fee)
            native_after_fees = am.sub(native, fee)
        else:
            native_after_fees = native
        self._conversions.append(_Conversion(native_after_fees, converted))
        self._inputs.add_amount(native)

    @property
    def transacted_amount(self) -> Amount:
        return self._transacted_amount

    def _to_transaction(self, accounts: Accounts) -> data.Transaction:
        postings = self._get_postings(accounts)
        return data.Transaction(self.meta, self.date, self.flag, self.payee, self.comment,
                                self.tags, self.links, postings)

    def _get_postings(self, accounts: Accounts) -> List[data.Posting]:
        postings: List[data.Posting] = []
        for units, cost in self._inputs:
            assert cost is None
            postings.append(data.Posting(
                accounts.get_borderless_account(units.currency), units, None, None, None, None))
        if self.divert_fees:
            for units, cost in self._fees:
                assert cost is None
                postings.append(data.Posting(
                    self.divert_fees, units, None, None, None, None))
            diverted_fees = self._fees
        else:
            diverted_fees = Inventory()
        # Also add the "fudge" amounts to the fees generated by rounding currency conversions.
        all_fees = Inventory()
        all_fees.add_inventory(self._fees)
        for acc, assigned_units in self.postings:
            for conversion in self._conversions:
                units, price, fudge = conversion.get_fraction(
                    self._transacted_amount, assigned_units, diverted_fees)
                postings.append(data.Posting(acc, -units, None, price, None, None))
                if fudge:
                    all_fees.add_amount(-fudge)
        for units, cost in all_fees:
            assert cost is None
            postings.append(data.Posting(
                accounts.fees_expense, -units, None, None, None, None))
        return postings


Extractor = Callable[[Row], None]


def extract_card_transaction(payment_processors: Dict[str, Optional[str]] = {}) -> Extractor:
    regexes = [(re.compile(f'^\\s*{key}\\s*\\*', re.IGNORECASE), value)
               for key, value in payment_processors.items()]

    def do_extract(row: Row) -> None:
        if row.entry_type == CARD_TYPE and CARD_REGEX.search(row.comment):
            if row.cd == CD_DEBIT:
                row.comment = ''
            else:
                row.comment = 'Refund'
                # Most manually add posting for refunded fees.
                row.flag = FLAG_WARNING
        if row.payee:
            for key, value in regexes:
                if match := key.search(row.payee):
                    if value:
                        row.tags.add(value)
                    row.payee = row.payee[match.end():].strip()
    return do_extract


def extract_add_money(add_money_asset: str, add_money_fees_asset: str) -> Extractor:
    def do_extract(row: Row) -> None:
        if row.entry_type == MONEY_ADDED_TYPE:
            row.payee = 'Transferwise'
            row.divert_fees = add_money_fees_asset
            row.assign_to_account(add_money_asset)
    return do_extract


class Importer(ImporterProtocol):
    _log: logging.Logger
    profile_id: int
    borderless_account_id: int
    currencies: List[str]
    accounts: Accounts
    _extractors: List[Extractor]

    def __init__(self,
                 profile_id: int,
                 borderless_account_id: int,
                 currencies: Iterable[str],
                 accounts: Accounts,
                 extractors: List[Extractor]):
        self._log = logging.getLogger(type(self).__qualname__)
        self.profile_id = profile_id
        self.borderless_account_id = borderless_account_id
        self.currencies = list(currencies)
        self.accounts = accounts
        self._extractors = extractors

    def _parse_file(self, file: FileMemo) -> Any:
        def parse_json(path: str) -> Any:
            with open(path, 'r') as json_file:
                try:
                    return json.load(json_file)
                except json.JSONDecodeError as exc:
                    self._log.info('Invalid JSON: %s', path, exc_info=exc)
                    return None

        return file.convert(parse_json)

    def identify(self, file: FileMemo) -> bool:
        _, extension = path.splitext(file.name)
        if extension.lower() != '.json':
            return False
        contents = self._parse_file(file)
        try:
            query = contents['query']
            return query['accountId'] == self.borderless_account_id and \
                query['currency'] in self.currencies
        except (KeyError, TypeError):
            return False

    def file_name(self, file: FileMemo) -> str:
        return 'statement.json'

    def file_account(self, file: FileMemo) -> str:
        contents = self._parse_file(file)
        try:
            currency = contents['query']['currency']
        except (KeyError, TypeError) as exc:
            raise ValueError(f'Invalid account statement: {file.name}') from exc
        if not isinstance(currency, str):
            raise ValueError(f'Invalid account statement: {file.name}')
        return self.accounts.get_borderless_account(currency)

    def file_date(self, file: FileMemo) -> dt.date:
        contents = self._parse_file(file)
        try:
            date_str = contents['query']['intervalEnd']
        except (KeyError, TypeError) as exc:
            raise ValueError(f'Invalid account statement: {file.name}') from exc
        if not isinstance(date_str, str):
            raise ValueError(f'Invalid account statement: {file.name}')
        return _parse_date(date_str)

    def extract(self, file: FileMemo) -> data.Entries:
        contents = self._parse_file(file)
        if contents:
            return self.extract_objects([contents])
        else:
            return []

    def extract_objects(self,
                        statements: Iterable[Any],
                        skip_references: Set[str] = set()) -> data.Entries:
        transactions: Dict[Reference, List[Any]] = defaultdict(lambda: [])
        for statement in statements:
            for transaction in statement['transactions']:
                reference_number = transaction['referenceNumber']
                if reference_number in skip_references:
                    continue
                reference = Reference(transaction['type'], reference_number)
                transactions[reference].append(transaction)
        entries: data.Entries = []
        for reference, transaction_list in transactions.items():
            if not transaction_list:
                continue
            try:
                row = Row(reference, transaction_list)
            except (TypeError, KeyError, InvalidEntry) as exc:
                self._log.warn('Invalid entry: %s', reference, exc_info=exc)
                continue
            try:
                utils.run_row_extractors(row, self._extractors)
            except InvalidEntry as exc:
                self._log.warn('Invalid entry: %s', reference, exc_info=exc)
                continue
            entries.append(row._to_transaction(self.accounts))
        entries.sort(key=lambda entry: entry.date)
        if entries:
            self._extract_closing_balances(statements, entries)
        return entries

    def _extract_closing_balances(self,
                                  statements: Iterable[Any],
                                  entries: data.Entries) -> None:
        for statement in statements:
            query = statement['query']
            end_date = _parse_date(query['intervalEnd']) + dt.timedelta(days=1)
            currency = query['currency']
            balance = statement['endOfStatementBalance']
            amount = Amount(utils.parse_number(balance['value']), balance['currency'])
            meta = data.new_metadata(f'<transferwise:{currency}>', 0)
            account = self.accounts.get_borderless_account(currency)
            entries.append(data.Balance(meta, end_date, account, amount, None, None))