Add plugins and importers from private config

The importers are missing tests, because not having any specifications
for the import formats means  we must use real, private data as test inputs

1 files changed, 377 insertions, 0 deletions

diff --git a/beancount_extras_kris7t/importers/transferwise/transferwise_json.py b/beancount_extras_kris7t/importers/transferwise/transferwise_json.py
new file mode 100644
index 0000000..c42de90
--- /dev/null
+++ b/beancount_extras_kris7t/importers/transferwise/transferwise_json.py
@@ -0,0 +1,377 @@
+'''
+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))