Fetching info from YNAB

src/ynrc/main.py

@@ -18,6 +18,10 @@ Note: This skeleton file can be safely removed if not needed!
 import argparse
 import logging
 import sys
+import datetime
+
+import hammock
+import requests
 
 from ynrc import __version__
 
@@ -43,6 +47,9 @@ def parse_args(args):
         action="version",
         version="ynrc {ver}".format(ver=__version__),
     )
+    parser.add_argument("--ynab-api-key", help="YNAB API key", type=str, required=True)
+    parser.add_argument("--ynab-budget-name", help="YNAB API key", type=str, required=True)
+    parser.add_argument("--rocketchat-webhook", help="YNAB API key", type=str, required=True)
     parser.add_argument(
         "-v",
         "--verbose",
@@ -74,6 +81,77 @@ def setup_logging(loglevel):
     )
 
 
+def _get_group_name_by_id(groups, group_id):
+    for group in groups:
+        if group_id == group['id']:
+            return group['name']
+    return ''
+
+
+def fetch_from_ynab(api_key, budget_name):
+    _logger.debug("Fetching YNAB budget %s...", budget_name)
+    ynab = hammock.Hammock(
+        "https://api.youneedabudget.com/v1",
+        headers={"Authorization": "Bearer %s" % api_key},
+    )
+
+    budgets = ynab.budgets.GET().json()['data']['budgets']
+    for budget in budgets:
+        if budget['name'] == budget_name:
+            budget_id = budget['id']
+            budget_full = ynab.budgets(budget_id).GET().json()['data']['budget']
+            category_groups = budget_full['category_groups']
+            break
+    else:
+        raise Exception('Budget with name "%s" not found' % budget_name)
+
+    _logger.debug('Using %s (%s)', budget_name, budget_id)
+
+    today = datetime.date.today()
+    week_ago = today - datetime.timedelta(days=7)
+    since_date = week_ago.strftime('%Y-%m-%d')
+
+    _logger.debug('Fetching all transactions since %s...', since_date)
+
+    transactions = ynab.budgets(budget_id).transactions.GET('?since_date=%s' % since_date).json()['data']['transactions']
+
+    # Calculate spending
+    res = {}
+    for tx in transactions:
+        if tx['subtransactions']:
+            tx = tx['subtransactions']
+        else:
+            tx = [tx]
+        for t in tx:
+            if t['amount'] > 0:
+                continue
+            tid = t['category_id']
+            if tid not in res:
+                res[tid] = {
+                    'amount': 0.0,
+                }
+            res[tid]['name'] = t.get('category_name', None)
+            res[tid]['amount'] += t['amount']
+
+    # Get category names
+    for tid, t in res.items():
+        if tid is None:
+            res[tid]['name'] = 'Transfer'
+            continue
+        cat = ynab.budgets(budget_id).categories(tid).GET().json()['data']['category']
+        parent_cat = _get_group_name_by_id(category_groups, cat['category_group_id'])
+        res[tid]['name'] = parent_cat + ': ' + cat['name']
+
+    # Sort top 10 results
+    out = []
+    for r in (sorted(res.values(), key=lambda k: k['amount'])):
+        amount = u'%4.2f €' % abs(r['amount'] / 1000)
+        out.append((u'%s' % r['name'], amount, ))
+        if len(out) >= 10:
+            break
+
+    print(out)
+
 def main_call(args):
     """Main entry point allowing external calls
 
@@ -83,6 +161,9 @@ def main_call(args):
     args = parse_args(args)
     setup_logging(args.loglevel)
     _logger.debug("Starting ynrc...")
+
+    fetch_from_ynab(args.ynab_api_key, args.ynab_budget_name)
+
     _logger.debug("Finished running ynrc")