ofx-processor/ofx_processor/utils/ynab.py

91 lines
2.6 KiB
Python
Raw Normal View History

2020-02-08 19:07:05 +01:00
import configparser
import os
import click
import requests
BASE_URL = "https://api.youneedabudget.com/v1"
DEFAULT_CONFIG_DIR = click.get_app_dir("ofx_processor")
DEFAULT_CONFIG_FILENAME = "config.ini"
def get_default_config():
default_config = configparser.ConfigParser()
default_config["DEFAULT"] = {
"token": "<YOUR API TOKEN>",
"budget": "<YOUR BUDGET ID>",
}
default_config["bpvf"] = {"account": "<YOUR ACCOUNT ID>"}
default_config["revolut"] = {"account": "<YOUR ACCOUNT ID>"}
default_config["ce"] = {"account": "<YOUR ACCOUNT ID>"}
2020-02-08 19:07:05 +01:00
return default_config
2020-02-22 15:34:40 +01:00
def get_config_file_name():
config_file = os.path.join(DEFAULT_CONFIG_DIR, DEFAULT_CONFIG_FILENAME)
return config_file
@click.group()
def config():
2020-02-29 11:12:22 +01:00
pass # This function acts only as a group to collect other subcommands
# As such, it intentionally has no body
2020-02-22 15:34:40 +01:00
@click.command()
def edit_config():
config_file = get_config_file_name()
click.edit(filename=config_file)
config.add_command(edit_config, "edit")
2020-02-08 19:07:05 +01:00
def push_transactions(transactions, account):
if not transactions:
click.secho("No transaction, nothing to do.", fg="yellow")
return
2020-02-08 19:07:05 +01:00
config = configparser.ConfigParser()
2020-02-22 15:34:40 +01:00
config_file = get_config_file_name()
2020-02-08 19:07:05 +01:00
if not os.path.isfile(config_file):
os.makedirs(DEFAULT_CONFIG_DIR, exist_ok=True)
config = get_default_config()
with open(config_file, "w") as file_:
config.write(file_)
click.secho("Editing config file...")
click.pause()
click.edit(filename=config_file)
config.read(config_file)
section = config[account]
budget_id = section["budget"]
url = f"{BASE_URL}/budgets/{budget_id}/transactions"
for transaction in transactions:
transaction["account_id"] = section["account"]
2020-02-11 23:58:08 +01:00
2020-02-08 19:07:05 +01:00
data = {"transactions": transactions}
token = section["token"]
headers = {"Authorization": f"Bearer {token}"}
2020-02-11 23:58:08 +01:00
2020-02-08 19:07:05 +01:00
res = requests.post(url, json=data, headers=headers)
res.raise_for_status()
data = res.json()["data"]
2020-02-11 23:58:08 +01:00
created = set()
for transaction in data["transactions"]:
matched_id = transaction["matched_transaction_id"]
if not matched_id or matched_id not in created:
created.add(transaction["id"])
2020-02-22 15:34:40 +01:00
if created:
click.secho(
f"{len(created)} transactions created in YNAB.", fg="green", bold=True
)
2020-02-11 23:58:08 +01:00
duplicates = data["duplicate_import_ids"]
2020-02-08 19:07:05 +01:00
if duplicates:
click.secho(
f"{len(duplicates)} transactions ignored (duplicates).", fg="yellow"
)