|
import logging |
|
from uuid import uuid4 |
|
import yaml |
|
import requests |
|
import json |
|
from cleantext import clean |
|
|
|
logging.basicConfig(format="%(levelname)s: %(message)s", level=logging.DEBUG) |
|
|
|
|
|
class SplitCategories(object): |
|
|
|
def __init__(self, api_key, config): |
|
self.api_token = api_key |
|
self.config=config |
|
|
|
def run(self,what=None): |
|
all_categories = self.get_all_categories() |
|
for budget, categories in self.config.items(): |
|
budget_id = self.find_budget_id(budget) |
|
budget_categories = all_categories[budget_id] |
|
logging.debug(f"Action:{what}") |
|
if what == 'categories': |
|
# this utils to test the settings/categories |
|
for item in categories: |
|
(payee, split_categories), = item.items() |
|
for split_category in split_categories: |
|
split_category=clean(split_category,no_emoji=True) |
|
logging.debug(f"{split_category} - {str(budget_categories[split_category])}") |
|
else: |
|
transactions = self.get_all_transactions(budget_id) |
|
# for all the settings that have categories |
|
for item in categories: |
|
# we get the data |
|
(payee, split_categories), = item.items() |
|
for transaction in transactions: |
|
# if the transaction has the payee |
|
# and is in our settings, |
|
# we update the transaction |
|
if not transaction['payee_name']: |
|
continue |
|
if payee in transaction['payee_name']: |
|
if transaction['memo'] != 'parsed': |
|
subtransactions = [] |
|
new_transaction = { |
|
"account_id": transaction['account_id'], |
|
"date": transaction['date'], |
|
"payee_name": transaction['payee_name'], |
|
"amount": transaction['amount'], |
|
"memo": transaction['memo'], |
|
"category_id": transaction['category_id'], |
|
"cleared": "cleared", |
|
"import_id": str(uuid4()), # we need a fake value |
|
"payee_id": transaction['payee_id'], |
|
"category_id": transaction['category_id'], |
|
"approved": transaction['approved'], |
|
# "flag_color": "red", |
|
"memo":"parsed", |
|
|
|
} |
|
total = transaction['amount'] |
|
for index, split_category in enumerate(split_categories): |
|
|
|
if type(split_category) == dict: |
|
(k, v), = split_category.items() |
|
logging.debug(f"{k},{v}") |
|
# this to ensure that the sum of the sub categories |
|
# matches with the total. |
|
if index == len(split_categories)-1: |
|
amount = total |
|
else: |
|
amount = int(transaction['amount']*v/100) |
|
# this works only for outflow, inflow will probably break here. |
|
total -= amount |
|
name = k |
|
else: |
|
if index == len(split_categories)-1: |
|
amount = total |
|
else: |
|
amount = int(transaction['amount']/len(split_categories)) |
|
total -= amount |
|
name = split_category |
|
name=clean(name,no_emoji=True) |
|
category_id = budget_categories[name] |
|
subtransactions.append({ |
|
# the amount is *100. so 2$ is 200 |
|
"amount": amount, |
|
"payee_id": transaction['payee_id'], |
|
# not needed |
|
# "payee_name": transaction['payee_name'], |
|
# "category_name": split_category, |
|
"category_id": category_id, |
|
"memo": "" |
|
}) |
|
new_transaction['subtransactions']= subtransactions |
|
self.put_transaction(budget_id, transaction['id'], {'transaction': new_transaction}) |
|
|
|
def put_transaction(self, budget_id,transaction_id, transaction): |
|
""" |
|
Update the transaction |
|
""" |
|
# send our data to API |
|
logging.info("Uploading transactions to YNAB...") |
|
|
|
url = ( |
|
"https://api.youneedabudget.com/v1/budgets/" |
|
+ "{}/transactions/{}?access_token={}".format( |
|
budget_id, transaction_id, self.api_token |
|
) |
|
) |
|
logging.debug(json.dumps(transaction, indent=2)) |
|
|
|
post_response = requests.put(url, json=transaction) |
|
|
|
# response handling - TODO: make this more thorough! |
|
try: |
|
self.process_api_response(json.loads(post_response.text)["error"]) |
|
except KeyError: |
|
logging.info("Success!") |
|
|
|
def find_budget_id(self, budget): |
|
""" |
|
gets id of a budget |
|
""" |
|
budgets = self.api_read(None, "budgets") |
|
if budgets[0] == "ERROR": |
|
return budgets |
|
for budget_ynab in budgets: |
|
logging.debug(f"{budget_ynab['name']} vs {budget}") |
|
if budget_ynab['name']==budget: |
|
return budget_ynab["id"] |
|
return "ERROR" |
|
|
|
def _parse_cartegories(self, group_categories): |
|
""" |
|
Util to return a dict where key is the category name |
|
and value the id. |
|
""" |
|
res = {} |
|
for group in group_categories: |
|
for category in group['categories']: |
|
res[clean(category['name'],no_emoji=True)]=category['id'] |
|
return res |
|
|
|
def get_all_categories(self): |
|
""" |
|
Get all the categories for all the budegets |
|
""" |
|
res = {} |
|
for budget, _ in self.config.items(): |
|
budget_id = self.find_budget_id(budget) |
|
all_categories = self._get_all_categories_api(budget_id) |
|
res[budget_id]=self._parse_cartegories(all_categories) |
|
return res |
|
|
|
def _get_all_categories_api(self, budget_id): |
|
""" |
|
Api call to get all transactions |
|
""" |
|
logging.debug(f"Budget {budget_id}") |
|
categories = self.api_read(budget_id, "categories","category_groups") |
|
if categories[0] == "ERROR": |
|
return categories |
|
if len(categories) > 0: |
|
logging.debug("Listing categories:") |
|
return categories |
|
else: |
|
logging.debug("no categories found") |
|
|
|
def get_all_transactions(self, budget_id): |
|
""" |
|
Function to get all transactions |
|
""" |
|
logging.debug(f"Budget {budget_id}") |
|
transactions = self.api_read(budget_id, "transactions") |
|
if transactions[0] == "ERROR": |
|
return transactions |
|
|
|
if len(transactions) > 0: |
|
logging.debug("Listing transactions:") |
|
return transactions |
|
else: |
|
logging.debug("no transactions found") |
|
|
|
def api_read(self, budget_id, kwd, kwd_ret=None): |
|
""" |
|
STOLEN FROM bank2ynab |
|
General function for reading data from YNAB API |
|
:param budget: boolean indicating if there's a default budget |
|
:param kwd: keyword for data type, e.g. transactions |
|
:param kwd_ret: the item to retrurn (some api do not match kwd / kwd_ret) |
|
:return error_codes: if it fails we return our error |
|
""" |
|
api_t = self.api_token |
|
base_url = "https://api.youneedabudget.com/v1/budgets/" |
|
|
|
if budget_id is None: |
|
# only happens when we're looking for the list of budgets |
|
url = base_url + "?access_token={}".format(api_t) |
|
else: |
|
url = base_url + "{}/{}?access_token={}".format( |
|
budget_id, kwd, api_t |
|
) |
|
|
|
logging.info(url) |
|
response = requests.get(url) |
|
try: |
|
if not kwd_ret: |
|
kwd_ret=kwd |
|
read_data = response.json()["data"][kwd_ret] |
|
except KeyError: |
|
logging.error(response.json()) |
|
# the API has returned an error so let's handle it |
|
return self.process_api_response(response.json()["error"]) |
|
return read_data |
|
|
|
def process_api_response(self, details): |
|
""" |
|
STOLEN FROM bank2ynab |
|
Prints details about errors returned by the YNAB api |
|
:param details: dictionary of returned error info from the YNAB api |
|
:return id: HTTP error ID |
|
:return detail: human-understandable explanation of error |
|
""" |
|
# TODO: make this function into a general response handler instead |
|
errors = { |
|
"400": "Bad syntax or validation error", |
|
"401": "API access token missing, invalid, revoked, or expired", |
|
"403.1": "The subscription for this account has lapsed.", |
|
"403.2": "The trial for this account has expired.", |
|
"404.1": "The specified URI does not exist.", |
|
"404.2": "Resource not found", |
|
"409": "Conflict error", |
|
"429": "Too many requests. Wait a while and try again.", |
|
"500": "Unexpected error", |
|
} |
|
id = details["id"] |
|
name = details["name"] |
|
detail = errors[id] |
|
logging.debug(details) |
|
logging.error("{} - {} ({})".format(id, detail, name)) |
|
|
|
return ["ERROR", id, detail] |
|
|
|
def categories(self): |
|
print(api_read()) |
|
|
|
import argparse |
|
parser = argparse.ArgumentParser( |
|
prog='SplitTransaction', |
|
description='What the program does', |
|
epilog='Text at the bottom of help') |
|
parser.add_argument('-c', '--categories', dest='categories', |
|
action='store_true') |
|
args = parser.parse_args() |
|
|
|
if __name__ == "__main__": |
|
import yaml |
|
with open("split.yaml", 'r') as stream: |
|
split_config = yaml.safe_load(stream) |
|
|
|
api_key = split_config.pop('api_key') |
|
|
|
split = SplitCategories(api_key, split_config) |
|
if args.categories: |
|
split.run('categories') |
|
else: |
|
split.run() |
