import json import datetime from ynab_splitwise_connect.transaction import Transaction from ynab_splitwise_connect.ynab_utils import YNABClient from ynab_splitwise_connect.splitwise_utils import SplitwiseClient def read_config(config_file_path): with open(config_file_path, 'r') as config_file: config = json.load(config_file) return config["ynab"], config["splitwise"], config["general"] def read_transactions(transaction_file_path): with open(transaction_file_path, 'r') as j: transactions = json.loads(j.read()) return transactions def process_general_config(general_config): transactions_path = general_config["transactions_path"] last_transactions_path = general_config["last_transactions_path"] days_included = general_config["days_included"] return transactions_path, last_transactions_path, days_included def calculate_dates(days_included): date_today = datetime.datetime.now() datetime_cutoff = date_today - datetime.timedelta(days=days_included) date_cutoff = datetime_cutoff.isoformat() return date_today, date_cutoff def update_transactions_file(transactions_dict, transactions_path): with open(transactions_path, "w") as outfile: json.dump(transactions_dict, outfile, indent=2) def get_category_ids_from_ynab_transactions(transactions_from_ynab): category_ids_from_ynab_transactions = {} for txn in transactions_from_ynab: category_ids_from_ynab_transactions[txn.id] = txn.category_id return category_ids_from_ynab_transactions def filter_transactions(transactions, check_date, payee_name): filtered_transactions = [] for txn in transactions: if check_date(txn['date']) and txn['payee_name'] == payee_name: filtered_transactions.append(txn) return filtered_transactions def check_valid_date(str_date, datetime_after): date_to_check = datetime.datetime.strptime(str_date, '%Y-%m-%d') difference = (date_to_check - datetime_after).days return difference >= -1 def main(): ynab_config, sw_config, general_config = read_config("config.json") transactions_path, last_transactions_path, days_included = process_general_config(general_config) transactions = read_transactions(transactions_path) date_today, date_cutoff = calculate_dates(days_included) ynab = YNABClient(ynab_config) sw = SplitwiseClient(sw_config) sw.set_expenses_included(date_cutoff) sw_transactions = [] for exp in sw.expenses_included: exp_origin = "splitwise" exp_description = exp.getDescription() exp_date = exp.getDate()[:10] exp_sw_id = exp.getId() exp_creditor_sw_id = exp.getRepayments()[0].getToUser() exp_debtor_sw_id = exp.getRepayments()[0].getFromUser() exp_debt = "-" + exp.getRepayments()[0].getAmount() exp_credit = str(round(float(exp.getCost()) + float(exp_debt), 2)) exp_deleted_sw = 'true' if exp.getDeletedBy() else 'false' exp_debtor = sw.debtor_name(exp_debtor_sw_id) exp_creditor = sw.creditor_name(exp_creditor_sw_id) txn = Transaction(origin=exp_origin, description=exp_description, date=exp_date, creditor_id=exp_creditor_sw_id, debtor_id=exp_debtor_sw_id, debt=exp_debt, credit=exp_credit, deleted_sw=exp_deleted_sw, debtor=exp_debtor, creditor=exp_creditor, sw_id=exp_sw_id) debtor = sw.debtor_name(txn.debtor_sw_id) creditor = sw.creditor_name(txn.creditor_sw_id) txn.set_debtor_and_creditor(debtor, creditor) sw_transactions.append(txn) transactions_dict = {txn.sw_id: txn for txn in transactions} # Lists to keep track of transactions that need to be created or updated in YNAB ynab_transactions_to_create = [] ynab_transactions_to_update = [] # The attributes that will be compared between the existing transaction and the splitwise transactions attributes_to_compare = ['description', 'cost', 'date', 'deleted_sw', 'creditor', 'debtor', 'debt', 'credit'] for sw_txn in sw_transactions: # If the transaction already exists in the dictionary if sw_txn.sw_id in transactions_dict: # If the values of sw_txn are different from the values of the existing transaction different_values = any(getattr(sw_txn, attr) != getattr(transactions_dict[sw_txn.sw_id], attr) for attr in attributes_to_compare) if different_values: # Copy the Splitwise transaction and add the YNAB ID updated_txn = sw_txn.copy() updated_txn.ynab_id = transactions_dict[sw_txn.sw_id].ynab_id # Add the updated transaction to the list of transactions to be updated in YNAB ynab_transactions_to_update.append(updated_txn) # If it's a new transaction else: if sw_txn.deleted == 'true': continue new_txn = sw_txn.copy() # Add the new transaction to the list of transactions to be created in YNAB ynab_transactions_to_create.append(new_txn) # Update transactions in YNAB for updated_txn in ynab_transactions_to_update.items(): try: ynab_id = updated_txn.ynab_id # Get filtered transactions from YNAB filtered_transactions = filter_transactions(ynab.transactions, check_valid_date, date_cutoff, sw.friend_name) # Get category IDs from YNAB transactions category_ids_from_ynab_transactions = get_category_ids_from_ynab_transactions(filtered_transactions) # In the 'Update transactions in YNAB' loop, get the category_id for the transaction being updated ynab_txn_category_id = category_ids_from_ynab_transactions[ynab_id] ynab_budget_id = ynab.budget_id if updated_txn.deleted_sw == 'true': ynab_txn_amount = 0 ynab_txn_memo = f"Removed from Splitwise - {updated_txn.description}" else: ynab_txn_amount = int(float(updated_txn.debt) * 1000) ynab_txn_memo = updated_txn.description ynab_txn_date = updated_txn.date ynab_txn_payee_name = updated_txn.creditor ynab_txn_payee_id = ynab.get_payee_id(ynab_txn_payee_name) # Create transaction object to be sent to YNAB ynab_txn = ynab.create_transaction(amount=ynab_txn_amount, date=ynab_txn_date, memo=ynab_txn_memo, payee_name=ynab_txn_payee_name, payee_id=ynab_txn_payee_id, category_id=ynab_txn_category_id) # API call to update the transaction in YNAB ynab.transactions.update_transaction(budget_id=ynab.budget_id, transaction_id=ynab_id, transaction=ynab_txn) # Update the transactions.json file with the updated transaction data transactions_dict[updated_txn.sw_id] = updated_txn # Save changes to transactions.json update_transactions_file(transactions_dict, transactions_path) except Exception as e: print(f"Error updating Splitwise transaction {updated_txn.sw_id}: {e}") # Create transactions in YNAB for new_txn in ynab_transactions_to_create: try: # Create YNAB transaction object budget_id = ynab.budget_id ynab_txn_payee_name = new_txn.creditor ynab_txn_payee_id = ynab.get_payee_id(ynab_txn_payee_name) ynab_txn_date = new_txn.date ynab_txn_amount = int(float(new_txn.debt) * 1000) ynab_txn_memo = new_txn.description new_ynab_txn = ynab.create_transaction(amount=ynab_txn_amount, date=ynab_txn_date, memo=ynab_txn_memo, payee_name=ynab_txn_payee_name, ynab_txn_payee_id=ynab_txn_payee_id) # API call to create the transaction in YNAB output_ynab_txn = ynab.transactions.create_transaction(budget_id=budget_id, transaction=new_ynab_txn) # Retrieve newly created YNAB transaction ID new_ynab_tnx_id = ynab.get_latest_tnx_id() new_txn["ynab_id"] = new_ynab_tnx_id # Add the created transactions to the transactions.json file transactions_dict[new_txn.sw_id] = sw_txn # Save changes to transactions.json update_transactions_file(transactions_dict, transactions_path) except Exception as e: print(f"Error creating Splitwise transaction {new_txn.sw_id}: {e}") if __name__ == "__main__": main()