<reponame>Tommoa/dots<filename>lib/python/up_api.py<gh_stars>1-10
#!/usr/bin/env python3

from io import TextIOBase
from os import getenv
import requests
import json

from beancount.core.amount import Amount, Decimal
from datetime import datetime, date, time
from dateutil import tz
from typing import Any, Dict, List, Optional, cast
from enum import IntEnum
from configparser import ConfigParser


class Account(object):
    """A class representing an Up Account
    """
    name: str
    id: str
    value: Amount
    attributes: Optional[Dict[str, Any]]
    created: datetime
    closed: Optional[datetime] = None

    def __init__(self, id: str, name: str, attributes: Dict[str, Any] = None):
        self.name = name
        self.id = id
        self.attributes = attributes
        self.created = datetime.fromisoformat(
            attributes.get('createdAt') or attributes['createdat'])
        self.closed = attributes.get('closedat')
        if self.closed:
            self.value = Amount(Decimal(0),
                                attributes.get('currency') or 'NOTHING')
            self.closed = datetime.fromisoformat(self.closed)
        else:
            balance = attributes['balance']
            self.value = Amount(Decimal(balance['value']),
                                balance['currencyCode'])

    @staticmethod
    def from_id(id: str, auth: str) -> 'Account':
        endpoint = f'https://api.up.com.au/api/v1/accounts/{id}'
        r = requests.get(endpoint, headers={'authorization': 'Bearer ' + auth})
        returned_data = json.loads(r.text)
        print(id, returned_data)
        attributes = returned_data['data']['attributes']
        return Account(id, attributes['displayName'], attributes)

    def __str__(self):
        return self.name

    def transactions(self, auth: str, since: date):
        endpoint = f"https://api.up.com.au/api/v1/accounts/{self.id}/transactions"
        parameters = {
            "filter[since]":
            datetime.combine(since, time(), tzinfo=tz.tzlocal()).isoformat(),
            "filter[status]":
            "SETTLED",
            "page[size]":
            100,
        }
        dates = []
        amounts = []
        from_account = []
        while endpoint:
            r = requests.get(endpoint,
                             headers={'authorization': 'Bearer ' + auth},
                             params=parameters)
            returned_data = json.loads(r.text)
            links = returned_data['links']
            returned_data = returned_data['data']
            attributes = list(x['attributes'] for x in returned_data)
            dates.extend(
                datetime.fromisoformat(x.get('settledAt')).date()
                for x in attributes)
            amounts.extend((x['amount']['value'], x['amount']['currencyCode'])
                           for x in attributes)
            from_account.extend(x['relationships']['account']['data']
                                for x in returned_data)
            parameters = None
            endpoint = links['next']


class TransactionStatus(IntEnum):
    HELD = 0
    SETTLED = 1

    @staticmethod
    def fromstr(input: str) -> 'TransactionStatus':
        if input == 'HELD':
            return TransactionStatus.HELD
        elif input == 'SETTLED':
            return TransactionStatus.SETTLED
        raise ValueError()


class Transaction(object):
    id: str
    status: TransactionStatus
    raw_text: str
    description: str
    message: str
    amount: Amount
    roundup: Optional[Amount] = None
    created: datetime
    settled: Optional[datetime] = None
    account: Optional[Account] = None

    def __init__(self, json_obj, id_to_account: Dict[str, Account]):
        self.id = json_obj['id']
        attributes = json_obj['attributes']
        account = id_to_account.get(
            json_obj['relationships']['account']['data']['id'])
        self.account = account
        self.status = TransactionStatus.fromstr(attributes['status'])
        self.raw_text = attributes['rawText']
        self.description = attributes['description']
        self.message = attributes['message']
        self.amount = Amount(Decimal(attributes['amount']['value']),
                             attributes['amount']['currencyCode'])
        if roundup := attributes['roundUp']:
            self.roundup = Amount(Decimal(roundup['amount']['value']),
                                  roundup['amount']['currencyCode'])
        if attributes['settledAt']:
            self.settled = datetime.fromisoformat(attributes['settledAt'])
        self.created = datetime.fromisoformat(attributes['createdAt'])

    def __repr__(self) -> str:
        return f"{self.account}: {self.description or self.raw_text} -- {self.message} ({self.amount})"


class Client(object):
    auth: str
    accounts: Optional[Dict[str, Account]] = None
    other_accounts: Optional[Dict[str, Account]] = None

    def __init__(self,
                 auth,
                 other_accounts: Optional[Dict[str, Account]] = None):
        self.auth = auth
        self.other_accounts = other_accounts

    def get_accounts(self) -> Dict[str, Account]:
        endpoint = 'https://api.up.com.au/api/v1/accounts'
        parameters = {
            "page[size]": 100,
        }
        attributes = []
        ids = []
        while endpoint:
            r = requests.get(endpoint,
                             headers={'authorization': 'Bearer ' + self.auth},
                             params=parameters)
            returned_data = json.loads(r.text)
            links = returned_data['links']
            returned_data = returned_data['data']
            attributes.extend(x['attributes'] for x in returned_data)
            ids.extend(x['id'] for x in returned_data)
            endpoint = links['next']
            parameters = None
        return dict(
            self.other_accounts or dict(), **{
                uid: Account(uid, attributes['displayName'], attributes)
                for uid, attributes in zip(ids, attributes)
            })

    def get_transactions(self,
                         since: Optional[date] = None,
                         until: Optional[date] = None) -> List[Transaction]:
        if not self.accounts:
            self.accounts = self.get_accounts()
        endpoint = 'https://api.up.com.au/api/v1/transactions'
        parameters: Optional[Dict[str, Any]] = {
            "page[size]": 100,
        }
        if since:
            parameters["filter[since]"] = datetime.combine(
                since, time(), tzinfo=tz.tzlocal()).isoformat()
        if until:
            parameters["filter[until]"] = datetime.combine(
                until, time(), tzinfo=tz.tzlocal()).isoformat()
        transactions = []
        while endpoint:
            r = requests.get(endpoint,
                             headers={'authorization': 'Bearer ' + self.auth},
                             params=parameters)
            returned_data = json.loads(r.text)
            links = returned_data['links']
            returned_data = returned_data['data']
            transactions.extend(
                Transaction(transaction, self.accounts)
                for transaction in returned_data)
            parameters = None
            endpoint = links['next']
        return transactions


def get_config(file=None) -> Dict[str, object]:
    from pathlib import Path

    configuration = ConfigParser()
    if not file:
        if conf_dir := getenv('XDG_CONFIG_DIR'):
            file = Path(conf_dir, "up.conf")
        else:
            file = Path.home() / ".config" / "up.conf"
    if isinstance(file, TextIOBase):
        configuration.read_file(file)
    else:
        with file.open('r') as conf:
            configuration.read_file(conf)

    auth_key = configuration["DEFAULT"].get("auth")
    if not auth_key:
        from subprocess import run
        from sys import stderr
        process = run([configuration.get("DEFAULT", "auth_cmd")],
                      shell=True,
                      capture_output=True,
                      text=True)
        print(process.stderr, file=stderr, end='')
        auth_key = process.stdout.rstrip()
    return dict({
        "_auth_key": auth_key,
    }, **{
        key: Account(key, configuration[key].get('name'),
                     dict(configuration.items(key)))
        for key in configuration.sections()
    })


if __name__ == '__main__':
    from argparse import ArgumentParser, FileType

    def sum_accounts(client: Client, parsed, **_):
        accounts = client.get_accounts().values()
        print(sum(x.value.number for x in accounts))

    def print_accounts(client: Client, parsed, **_):
        accounts = client.get_accounts().values()
        for account in accounts:
            print(account.name + ":", account.value)

    parser = ArgumentParser()
    parser.set_defaults(func=sum_accounts)
    parser.add_argument("-c",
                        "--config",
                        required=False,
                        type=FileType('r'),
                        default=None)

    subparser = parser.add_subparsers(title="command")

    sum_parser = subparser.add_parser("sum")
    sum_parser.set_defaults(func=sum_accounts)

    print_parser = subparser.add_parser("accounts", aliases=["print"])
    print_parser.set_defaults(func=print_accounts)

    parsed = parser.parse_args()

    configuration = get_config(parsed.config)

    client = Client(configuration.pop("_auth_key"),
                    cast(Dict[str, Account], configuration))

    parsed.func(client, parsed)

#!/usr/bin/env python3

from io import TextIOBase
from os import getenv
import requests
import json

from beancount.core.amount import Amount, Decimal
from datetime import datetime, date, time
from dateutil import tz
from typing import Any, Dict, List, Optional, cast
from enum import IntEnum
from configparser import ConfigParser


class Account(object):
    """A class representing an Up Account
    """
    name: str
    id: str
    value: Amount
    attributes: Optional[Dict[str, Any]]
    created: datetime
    closed: Optional[datetime] = None

    def __init__(self, id: str, name: str, attributes: Dict[str, Any] = None):
        self.name = name
        self.id = id
        self.attributes = attributes
        self.created = datetime.fromisoformat(
            attributes.get('createdAt') or attributes['createdat'])
        self.closed = attributes.get('closedat')
        if self.closed:
            self.value = Amount(Decimal(0),
                                attributes.get('currency') or 'NOTHING')
            self.closed = datetime.fromisoformat(self.closed)
        else:
            balance = attributes['balance']
            self.value = Amount(Decimal(balance['value']),
                                balance['currencyCode'])

    @staticmethod
    def from_id(id: str, auth: str) -> 'Account':
        endpoint = f'https://api.up.com.au/api/v1/accounts/{id}'
        r = requests.get(endpoint, headers={'authorization': 'Bearer ' + auth})
        returned_data = json.loads(r.text)
        print(id, returned_data)
        attributes = returned_data['data']['attributes']
        return Account(id, attributes['displayName'], attributes)

    def __str__(self):
        return self.name

    def transactions(self, auth: str, since: date):
        endpoint = f"https://api.up.com.au/api/v1/accounts/{self.id}/transactions"
        parameters = {
            "filter[since]":
            datetime.combine(since, time(), tzinfo=tz.tzlocal()).isoformat(),
            "filter[status]":
            "SETTLED",
            "page[size]":
            100,
        }
        dates = []
        amounts = []
        from_account = []
        while endpoint:
            r = requests.get(endpoint,
                             headers={'authorization': 'Bearer ' + auth},
                             params=parameters)
            returned_data = json.loads(r.text)
            links = returned_data['links']
            returned_data = returned_data['data']
            attributes = list(x['attributes'] for x in returned_data)
            dates.extend(
                datetime.fromisoformat(x.get('settledAt')).date()
                for x in attributes)
            amounts.extend((x['amount']['value'], x['amount']['currencyCode'])
                           for x in attributes)
            from_account.extend(x['relationships']['account']['data']
                                for x in returned_data)
            parameters = None
            endpoint = links['next']


class TransactionStatus(IntEnum):
    HELD = 0
    SETTLED = 1

    @staticmethod
    def fromstr(input: str) -> 'TransactionStatus':
        if input == 'HELD':
            return TransactionStatus.HELD
        elif input == 'SETTLED':
            return TransactionStatus.SETTLED
        raise ValueError()


class Transaction(object):
    id: str
    status: TransactionStatus
    raw_text: str
    description: str
    message: str
    amount: Amount
    roundup: Optional[Amount] = None
    created: datetime
    settled: Optional[datetime] = None
    account: Optional[Account] = None

    def __init__(self, json_obj, id_to_account: Dict[str, Account]):
        self.id = json_obj['id']
        attributes = json_obj['attributes']
        account = id_to_account.get(
            json_obj['relationships']['account']['data']['id'])
        self.account = account
        self.status = TransactionStatus.fromstr(attributes['status'])
        self.raw_text = attributes['rawText']
        self.description = attributes['description']
        self.message = attributes['message']
        self.amount = Amount(Decimal(attributes['amount']['value']),
                             attributes['amount']['currencyCode'])
        if roundup := attributes['roundUp']:
            self.roundup = Amount(Decimal(roundup['amount']['value']),
                                  roundup['amount']['currencyCode'])
        if attributes['settledAt']:
            self.settled = datetime.fromisoformat(attributes['settledAt'])
        self.created = datetime.fromisoformat(attributes['createdAt'])

    def __repr__(self) -> str:
        return f"{self.account}: {self.description or self.raw_text} -- {self.message} ({self.amount})"


class Client(object):
    auth: str
    accounts: Optional[Dict[str, Account]] = None
    other_accounts: Optional[Dict[str, Account]] = None

    def __init__(self,
                 auth,
                 other_accounts: Optional[Dict[str, Account]] = None):
        self.auth = auth
        self.other_accounts = other_accounts

    def get_accounts(self) -> Dict[str, Account]:
        endpoint = 'https://api.up.com.au/api/v1/accounts'
        parameters = {
            "page[size]": 100,
        }
        attributes = []
        ids = []
        while endpoint:
            r = requests.get(endpoint,
                             headers={'authorization': 'Bearer ' + self.auth},
                             params=parameters)
            returned_data = json.loads(r.text)
            links = returned_data['links']
            returned_data = returned_data['data']
            attributes.extend(x['attributes'] for x in returned_data)
            ids.extend(x['id'] for x in returned_data)
            endpoint = links['next']
            parameters = None
        return dict(
            self.other_accounts or dict(), **{
                uid: Account(uid, attributes['displayName'], attributes)
                for uid, attributes in zip(ids, attributes)
            })

    def get_transactions(self,
                         since: Optional[date] = None,
                         until: Optional[date] = None) -> List[Transaction]:
        if not self.accounts:
            self.accounts = self.get_accounts()
        endpoint = 'https://api.up.com.au/api/v1/transactions'
        parameters: Optional[Dict[str, Any]] = {
            "page[size]": 100,
        }
        if since:
            parameters["filter[since]"] = datetime.combine(
                since, time(), tzinfo=tz.tzlocal()).isoformat()
        if until:
            parameters["filter[until]"] = datetime.combine(
                until, time(), tzinfo=tz.tzlocal()).isoformat()
        transactions = []
        while endpoint:
            r = requests.get(endpoint,
                             headers={'authorization': 'Bearer ' + self.auth},
                             params=parameters)
            returned_data = json.loads(r.text)
            links = returned_data['links']
            returned_data = returned_data['data']
            transactions.extend(
                Transaction(transaction, self.accounts)
                for transaction in returned_data)
            parameters = None
            endpoint = links['next']
        return transactions


def get_config(file=None) -> Dict[str, object]:
    from pathlib import Path

    configuration = ConfigParser()
    if not file:
        if conf_dir := getenv('XDG_CONFIG_DIR'):
            file = Path(conf_dir, "up.conf")
        else:
            file = Path.home() / ".config" / "up.conf"
    if isinstance(file, TextIOBase):
        configuration.read_file(file)
    else:
        with file.open('r') as conf:
            configuration.read_file(conf)

    auth_key = configuration["DEFAULT"].get("auth")
    if not auth_key:
        from subprocess import run
        from sys import stderr
        process = run([configuration.get("DEFAULT", "auth_cmd")],
                      shell=True,
                      capture_output=True,
                      text=True)
        print(process.stderr, file=stderr, end='')
        auth_key = process.stdout.rstrip()
    return dict({
        "_auth_key": auth_key,
    }, **{
        key: Account(key, configuration[key].get('name'),
                     dict(configuration.items(key)))
        for key in configuration.sections()
    })


if __name__ == '__main__':
    from argparse import ArgumentParser, FileType

    def sum_accounts(client: Client, parsed, **_):
        accounts = client.get_accounts().values()
        print(sum(x.value.number for x in accounts))

    def print_accounts(client: Client, parsed, **_):
        accounts = client.get_accounts().values()
        for account in accounts:
            print(account.name + ":", account.value)

    parser = ArgumentParser()
    parser.set_defaults(func=sum_accounts)
    parser.add_argument("-c",
                        "--config",
                        required=False,
                        type=FileType('r'),
                        default=None)

    subparser = parser.add_subparsers(title="command")

    sum_parser = subparser.add_parser("sum")
    sum_parser.set_defaults(func=sum_accounts)

    print_parser = subparser.add_parser("accounts", aliases=["print"])
    print_parser.set_defaults(func=print_accounts)

    parsed = parser.parse_args()

    configuration = get_config(parsed.config)

    client = Client(configuration.pop("_auth_key"),
                    cast(Dict[str, Account], configuration))

    parsed.func(client, parsed)
#!/usr/bin/env python3 A class representing an Up Account