Update Lunch Money transactions with Mint data on payees, categories, etc.

reconcile_mint.py

#!/usr/bin/env python

################################################################################
#                                reconcile-mint                                #
#                 This script doesn't recreate Mint categories                 #
#     Parsing transactions with multiple tags should work, but is untested     #
#    You must manually split transactions for this script to recognize them    #
#  Mint and Lunch Money account names must match (different display names ok)  #
#                               (C) 2021 Mischif                               #
#       Released under version 3.0 of the Non-Profit Open Source License       #
################################################################################

from argparse import ArgumentParser
from csv import DictReader
from datetime import datetime, timedelta
from decimal import Decimal
from itertools import islice
from json import dumps, load
from operator import itemgetter
from pathlib import Path
from urllib.request import Request, urlopen

ISO8601 = "%Y-%m-%d"
RFC3339 = "%Y-%m-%dT%X.%fZ"

class LMClient:
	API_ROOT = "https://dev.lunchmoney.app"

	def __init__(self, token, version="v1"):
		self.api_root = f"{LMClient.API_ROOT}/{version}"
		self.req_headers = {"Authorization": f"Bearer {token}",
							"Content-Type": "application/json"}
		self.accounts = {acct["id"]: acct for acct in self.accounts()}
		self.categories = {cat["cat_id"]: cat for cat in self.categories()}
		self.tags = {tag["id"]: tag for tag in self.tags()}

	def get(self, endpoint):
		result = None
		url = f"{self.api_root}/{endpoint}"

		req = Request(url, headers=self.req_headers)
		try:
			with urlopen(req) as res:
				result = load(res)
		except Exception as e:
			raise

		return result

	def put(self, endpoint, data):
		result = None
		url = f"{self.api_root}/{endpoint}"

		if isinstance(data, dict):
			data = dumps(data).encode("utf-8")

		req = Request(url, headers=self.req_headers, data=data, method="PUT")
		try:
			with urlopen(req) as res:
				result = load(res)
		except Exception as e:
			raise

		return result

	def accounts(self):
		result = []

		data = self.get("plaid_accounts")
		if data is not None:
			result = [{
				"active": True if acct["status"] == "active" else False,
				"bal": Decimal(acct["balance"]).normalize(),
				"cur": acct["currency"],
				"id": acct["id"],
				# "imported": datetime.strptime(acct["last_import"], RFC3339)
				"institution": acct["institution_name"],
				"limit": acct["limit"],
				"linked": datetime.strptime(acct["date_linked"],
											ISO8601).date(),
				"mask": acct["mask"],
				"name": acct["name"],
				"subtype": acct["subtype"],
				"type": acct["type"],
				"updated": datetime.strptime(acct["balance_last_update"],
											 RFC3339),
				} for acct in data["plaid_accounts"]]

		return result

	def categories(self):
		result = []

		data = self.get("categories")
		if data is not None:
			result = [{
				"cat_id": cat["id"],
				"created": datetime.strptime(cat["created_at"], RFC3339),
				"desc": cat["description"],
				"grp_id": cat["group_id"],
				"is_grp": cat["is_group"],
				"in_totals": not cat["exclude_from_totals"],
				"in_budgets": not cat["exclude_from_budget"],
				"income": cat["is_income"],
				"name": cat["name"],
				"updated": datetime.strptime(cat["updated_at"], RFC3339),
				} for cat in data.get("categories", [])]

		return result

	def tags(self):
		result = []

		data = self.get("tags")
		if data is not None:
			result = [{
				"desc": txn["description"],
				"id": txn["id"],
				"name": txn["name"],
				} for txn in data]

		return result

	def transactions(self, start=None, end=None):
		result = []

		if start is not None and end is not None:
			query = f"transactions?start_date={start}&end_date={end}"
		else:
			query = "transactions"

		data = self.get(query)
		if data is not None:
			result = [{
				"acct_id": txn["plaid_account_id"],
				"amt": Decimal(txn["amount"]).normalize(),
				"asset_id": txn["asset_id"],
				"cat_id": txn["category_id"],
				"cur": txn["currency"],
				"date": datetime.strptime(txn["date"], ISO8601).date(),
				"desc": txn["payee"],
				"ext_id": txn["external_id"],
				"grp_id": txn["group_id"],
				"is_grp": txn["is_group"],
				"notes": txn["notes"],
				"par_id": txn["parent_id"],
				"rcr_id": txn["recurring_id"],
				"status": txn["status"],
				"tags": txn["tags"],
				"txn_id": txn["id"],
				} for txn in data.get("transactions", [])]

			for txn in result:
				if txn["acct_id"] is not None:
					txn["acct"] = self.accounts[txn["acct_id"]]["name"]

				if txn["cat_id"] is not None:
					txn["cat"] = self.categories[txn["cat_id"]]["name"]

		return result

	def update_txns(self, txns):
		for txn in txns:
			txn.pop("cat")
			txn_id = txn.pop("id")
			data = self.put(f"transactions/{txn_id}", {"transaction": txn})
			if not data.get("updated", False):
				print(f"Could not update txn {txn_id} with data {txn}")

def mint_transactions(path):
	result = []

	with open(path) as f:
		csv = DictReader(f)

		result = [{
			"acct": txn["Account Name"],
			"amt": Decimal(txn["Amount"]).normalize(),
			"cat": txn["Category"],
			"date": datetime.strptime(txn["Date"], "%m/%d/%Y").date(),
			"desc": txn["Description"],
			"desc_orig": txn["Original Description"],
			"lbls": txn["Labels"],
			"notes": txn["Notes"],
			"type": txn["Transaction Type"],
			} for txn in csv]

	return result

def match_txns(lm, mint):
	result = []

	def _build_change_indices(l):
		result = {}
		cur = None

		for (i, txn) in enumerate(l):
			if txn["date"] != cur:
				result[txn["date"]] = i
				cur = txn["date"]

		return result

	def _txn_slice(txn_list, date_indices, tgt_date):
		result = ()

		start_date = tgt_date - timedelta(days=1)
		if start_date not in date_indices:
			if tgt_date in date_indices:
				start_date = tgt_date
			else:
				start_date = None

		if start_date:
			if max(date_indices) == tgt_date:
				end_date = None
			else:
				end_date = next(filter(lambda d: d > tgt_date, date_indices))

			start_idx = date_indices[start_date]
			end_idx = None if end_date is None else date_indices[end_date]
			result = islice(txn_list, start_idx, end_idx)

		return result

	lm = sorted(lm, key=itemgetter("date"))
	lm_indices = _build_change_indices(lm)

	mint = sorted(mint, key=itemgetter("date"))
	mint_indices = _build_change_indices(mint)

	for lm_txn in lm:
		for mint_txn in _txn_slice(mint, mint_indices, lm_txn["date"]):
			if (lm_txn["acct"] == mint_txn["acct"]
				and lm_txn["amt"].compare_total_mag(mint_txn["amt"]).is_zero()
				and (lm_txn["amt"] < 0 and mint_txn["type"] == "credit"
					 or lm_txn["amt"] > 0 and mint_txn["type"] == "debit")):
				result.append((lm_txn, mint_txn))

	return result

def unify_txns(matched_txns, categories, tags):
	result = []

	for (lm, mint) in matched_txns:
		def _cat_filter(cat):
			return cat["name"].lower() == mint["cat"].lower()

		try:
			cat_id = next(filter(_cat_filter, categories))["cat_id"]
		except StopIteration:
			cat_id = lm["cat_id"]

		tag_ids = set(lm["tags"] or [])
		for lbl in filter(None, mint["lbls"].split(",")):
			def _tag_filter(tag):
				return tag["name"].lower() == lbl.lower()

			try:
				tag_id = next(filter(_tag_filter, tags))["id"]
			except StopIteration:
				tag_ids.add(lbl)


		unified_txn = { 
			"cat": next(filter(lambda c: c["cat_id"] == cat_id, categories))["name"],
			"id": lm["txn_id"],
			"category_id": cat_id,
			"payee": mint["desc"][:140],
			}

		if mint["notes"]:
			unified_txn["notes"] = mint["notes"][:350]
		elif lm["notes"]:
			unified_txn["notes"] = lm["notes"]

		if tag_ids:
			unified_txn["tags"] = list(tag_ids)

		result.append(unified_txn)

	return result

def parse_args():
	parser = ArgumentParser(
		prog="reconcile-mint",
		epilog="Released under NPOSL-3.0, (C) 2021 Mischif",
		)

	parser.add_argument(
		"-t", "--token",
		required=True,
		type=str,
		help="Lunch Money access token",
		)

	parser.add_argument(
		"--csv",
		required=True,
		type=Path,
		dest="csv_path",
		help="Path to Mint transaction CSV",
		)

	return parser.parse_args()

def main():
	args = parse_args()

	mint_txns = mint_transactions(args.csv_path)
	start_date = str(min(mint_txns, key=itemgetter("date"))["date"])
	end_date = str(max(mint_txns, key=itemgetter("date"))["date"])
	print(f"Fetching transactions between {start_date} and {end_date}")

	c = LMClient(args.token)
	lm_txns = c.transactions(start_date, end_date)

	matched_txns = match_txns(lm_txns, mint_txns)
	unified_txns = unify_txns(matched_txns, c.categories.values(), c.tags.values())
	print(f"{len(unified_txns)} transactions to be reconciled")
	c.update_txns(unified_txns)

if __name__ == "__main__": main()