johnmee · September 7, 2022 23:06
diff --git a/upbank_importer.py b/upbank_importer.py
 import json

 from datetime import date

 import beancount
 from beancount.core import data
 from beancount.core import flags
 from beancount.core import (account, amount, number)
 from beancount.ingest import importer

 # Upbank only operates in AUD, afaik.
 CURRENCY = "AUD"


 class UpbankImporter(importer.ImporterProtocol):
    """Interface that all source importers need to comply with.
    """

    # A flag to use on new transaction. Override this flag as you prefer.
    FLAG = beancount.core.flags.FLAG_OKAY

    def __init__(self, account_name="Assets:Bank:Upbank", tags=data.EMPTY_SET):
        """

        Args:
            account_name: beancount account name for the upbank account.
                eg:  "Assets:Bank:Upbank"
        """
        self.account_name = account_name
        self.tags = tags

    def name(self):
        """Return a unique id/name for this importer.

        Returns:
          A string which uniquely identifies this importer.
        """
        return "Upbank"

    __str__ = name

    def identify(self, file):
        """Return true if this importer matches the given file.

        Args:
          file: A cache.FileMemo instance.
        Returns:
          A boolean, true if this importer can handle this file.
        """
        try:
            # Is a json file with specific structure and attributes.
            transactions = json.loads(file.contents())
            assert isinstance(transactions, list), "JSON file is not a list of transactions."
            assert len(transactions) > 0, "JSON list of transactions is empty"
            for trans in transactions:
                assert isinstance(trans, dict), "JSON list contains a non-dictionary item."
                assert "type" in trans
                assert "id" in trans
                assert "attributes" in trans
                assert "relationships" in trans
                assert "links" in trans

            # Use the filename to determine which account.
            assert any(tag in file.name for tag in self.tags), \
                "Tag '%s' not found in filename '%s'" % (self.tags, file.name)

            return True
        except json.JSONDecodeError:
            pass
        except AssertionError:
            pass
        return False

    def extract(self, file, existing_entries=None):
        """Extract transactions from a file.

        Args:
          file: A cache.FileMemo instance.
          existing_entries: An optional list of existing directives loaded from
            the ledger which is intended to contain the extracted entries. This
            is only provided if the user provides them via a flag in the
            extractor program.
        Returns:
          A list of new, imported directives (usually mostly Transactions)
          extracted from the file.
        """
        # Open the file as json
        transactions = json.loads(file.contents())
        entries = []

        for trans in reversed(transactions):
            trans_id = trans['id']   # Could be used to flag "__duplicate__"s.
            date_ = date.fromisoformat(trans['attributes']['createdAt'][:10])
            raw_text = trans['attributes']['rawText']
            description = trans['attributes']['description']
            value = amount.Amount(
                beancount.core.number.D(trans['attributes']['amount']['value']),
                CURRENCY
            )
            posting = data.Posting(self.account_name, value, None, None, None, None)
            txn = data.Transaction(
                meta=data.new_metadata(file.name, trans_id),
                date=date_,
                flag=beancount.core.flags.FLAG_OKAY,
                payee=description,
                tags=self.tags,
                links=data.EMPTY_SET,
                narration=raw_text,
                postings=[posting],
            )
            entries.append(txn)

        # TODO: add categories:
        #  use the 'existing_entries' for categories? (add a second posting)
        #  use the upbank provided categories?
        #  use my own static description matching categorizer?
        #  build a history-inspecting categorizer?
        # TODO: insert a balance line?
        return entries

        #     try:
        #         category = ':'.join([
        #             trans['relationships']['parentCategory']['data']['id'],
        #             trans['relationships']['category']['data']['id'],
        #         ])
        #     except TypeError:
        #         category = ''
        #     txt_description = f"\"{raw_text} [{description}]\""
        #     entry = f'{date_str} * {txt_description:66s}  {tags}\n'
        #     entry += f"    {accountname}\n"
        #     entry += "    %-46s %10.2f AUD\n" % (
        #         f'ACCOUNT_UNKNOWN [{category}]', float(value))
        #
        #     click.echo(entry)
        #     change += float(value)
        # balance = balance - change
        # balance_str = f"{date_str} balance {accountname:36s} {balance:.2f} AUD"
        # click.echo(balance_str)

    def file_account(self, file):
        """Return an account associated with the given file.

        Note: If you don't implement this method you won't be able to move the
        files into its preservation hierarchy; the bean-file command won't
        work.

        Also, normally the returned account is not a function of the input
        file--just of the importer--but it is provided anyhow.

        Args:
          file: A cache.FileMemo instance.
        Returns:
          The name of the account that corresponds to this importer.
        """
        return self.account_name

    def file_name(self, file):
        """A filter that optionally renames a file before filing.

        This is used to make tidy filenames for filed/stored document files. If
        you don't implement this and return None, the same filename is used.
        Note that if you return a filename, a simple, RELATIVE filename must be
        returned, not an absolute filename.

        Args:
          file: A cache.FileMemo instance.
        Returns:
          The tidied up, new filename to store it as.
        """
        # Use the account name.
        account_str = self.account_name.split(":")[-1]
        return f"{account_str}.json"

    def file_date(self, file):
        """Attempt to obtain a date that corresponds to the given file.

        Args:
          file: A cache.FileMemo instance.
        Returns:
          A date object, if successful, or None if a date could not be extracted.
          (If no date is returned, the file creation time is used. This is the
          default.)
        """
        # Date of last transaction in the file.
        return self.extract(file)[-1].date
	import json

	from datetime import date

	import beancount
	from beancount.core import data
	from beancount.core import flags
	from beancount.core import (account, amount, number)
	from beancount.ingest import importer

	# Upbank only operates in AUD, afaik.
	CURRENCY = "AUD"


	class UpbankImporter(importer.ImporterProtocol):
	"""Interface that all source importers need to comply with.
	"""

	# A flag to use on new transaction. Override this flag as you prefer.
	FLAG = beancount.core.flags.FLAG_OKAY

	def __init__(self, account_name="Assets:Bank:Upbank", tags=data.EMPTY_SET):
	"""

	Args:
	account_name: beancount account name for the upbank account.
	eg: "Assets:Bank:Upbank"
	"""
	self.account_name = account_name
	self.tags = tags

	def name(self):
	"""Return a unique id/name for this importer.

	Returns:
	A string which uniquely identifies this importer.
	"""
	return "Upbank"

	__str__ = name

	def identify(self, file):
	"""Return true if this importer matches the given file.

	Args:
	file: A cache.FileMemo instance.
	Returns:
	A boolean, true if this importer can handle this file.
	"""
	try:
	# Is a json file with specific structure and attributes.
	transactions = json.loads(file.contents())
	assert isinstance(transactions, list), "JSON file is not a list of transactions."
	assert len(transactions) > 0, "JSON list of transactions is empty"
	for trans in transactions:
	assert isinstance(trans, dict), "JSON list contains a non-dictionary item."
	assert "type" in trans
	assert "id" in trans
	assert "attributes" in trans
	assert "relationships" in trans
	assert "links" in trans

	# Use the filename to determine which account.
	assert any(tag in file.name for tag in self.tags), \
	"Tag '%s' not found in filename '%s'" % (self.tags, file.name)

	return True
	except json.JSONDecodeError:
	pass
	except AssertionError:
	pass
	return False

	def extract(self, file, existing_entries=None):
	"""Extract transactions from a file.

	Args:
	file: A cache.FileMemo instance.
	existing_entries: An optional list of existing directives loaded from
	the ledger which is intended to contain the extracted entries. This
	is only provided if the user provides them via a flag in the
	extractor program.
	Returns:
	A list of new, imported directives (usually mostly Transactions)
	extracted from the file.
	"""
	# Open the file as json
	transactions = json.loads(file.contents())
	entries = []

	for trans in reversed(transactions):
	trans_id = trans['id'] # Could be used to flag "__duplicate__"s.
	date_ = date.fromisoformat(trans['attributes']['createdAt'][:10])
	raw_text = trans['attributes']['rawText']
	description = trans['attributes']['description']
	value = amount.Amount(
	beancount.core.number.D(trans['attributes']['amount']['value']),
	CURRENCY
	)
	posting = data.Posting(self.account_name, value, None, None, None, None)
	txn = data.Transaction(
	meta=data.new_metadata(file.name, trans_id),
	date=date_,
	flag=beancount.core.flags.FLAG_OKAY,
	payee=description,
	tags=self.tags,
	links=data.EMPTY_SET,
	narration=raw_text,
	postings=[posting],
	)
	entries.append(txn)

	# TODO: add categories:
	# use the 'existing_entries' for categories? (add a second posting)
	# use the upbank provided categories?
	# use my own static description matching categorizer?
	# build a history-inspecting categorizer?
	# TODO: insert a balance line?
	return entries

	# try:
	# category = ':'.join([
	# trans['relationships']['parentCategory']['data']['id'],
	# trans['relationships']['category']['data']['id'],
	# ])
	# except TypeError:
	# category = ''
	# txt_description = f"\"{raw_text} [{description}]\""
	# entry = f'{date_str} * {txt_description:66s} {tags}\n'
	# entry += f" {accountname}\n"
	# entry += " %-46s %10.2f AUD\n" % (
	# f'ACCOUNT_UNKNOWN [{category}]', float(value))
	#
	# click.echo(entry)
	# change += float(value)
	# balance = balance - change
	# balance_str = f"{date_str} balance {accountname:36s} {balance:.2f} AUD"
	# click.echo(balance_str)

	def file_account(self, file):
	"""Return an account associated with the given file.

	Note: If you don't implement this method you won't be able to move the
	files into its preservation hierarchy; the bean-file command won't
	work.

	Also, normally the returned account is not a function of the input
	file--just of the importer--but it is provided anyhow.

	Args:
	file: A cache.FileMemo instance.
	Returns:
	The name of the account that corresponds to this importer.
	"""
	return self.account_name

	def file_name(self, file):
	"""A filter that optionally renames a file before filing.

	This is used to make tidy filenames for filed/stored document files. If
	you don't implement this and return None, the same filename is used.
	Note that if you return a filename, a simple, RELATIVE filename must be
	returned, not an absolute filename.

	Args:
	file: A cache.FileMemo instance.
	Returns:
	The tidied up, new filename to store it as.
	"""
	# Use the account name.
	account_str = self.account_name.split(":")[-1]
	return f"{account_str}.json"

	def file_date(self, file):
	"""Attempt to obtain a date that corresponds to the given file.

	Args:
	file: A cache.FileMemo instance.
	Returns:
	A date object, if successful, or None if a date could not be extracted.
	(If no date is returned, the file creation time is used. This is the
	default.)
	"""
	# Date of last transaction in the file.
	return self.extract(file)[-1].date