Python script for the Firefly-III personal finance app. Finds and merges equivalent Withdrawal/Deposit pairs into a single Transfer. See also: https://www.reddit.com/r/FireflyIII/comments/lwuzia/hide_certain_categories/

fixtransfers.py

import os
import sys
import requests
from pprint import pprint
from datetime import datetime
from dataclasses import dataclass
from time import time

@dataclass
class Transaction:
	transaction_id: int
	amount: float
	date: datetime
	source_id: int
	source_name: str
	destination_id: int
	destination_name: str
	typ: str
	category_id: int
	description: str

def make_transaction_list(json_response):
	transactions = []
	for t in json_response['data']:
		for z in t['attributes']['transactions']:
			transactions.append(
				Transaction(
					transaction_id = t['id'],
					amount = z['amount'],
					date = datetime.strptime(z['date'], '%Y-%m-%dT%H:%M:%S+00:00'),
					source_id = z['source_id'],
					source_name = z['source_name'],
					destination_id = z['destination_id'],
					destination_name = z['destination_name'],
					typ = z['type'],
					category_id = z['category_id'],
					description = z['description']
				)
			)
	return transactions


def make_pairs(transactions):
	ws = [t for t in transactions if t.typ == 'withdrawal']
	ds = [t for t in transactions if t.typ == 'deposit']
	pairs = []
	for w in ws:
		same_amount = [d for d in ds if w.amount == d.amount]
		with_day_delta = [(d, (d.date - w.date).days) for d in same_amount]
		within_10 = [(d, delta) for (d, delta) in with_day_delta if abs(delta) <= 10]
		if len(within_10) > 0:
			(closest, _) = min(within_10, key=lambda x: abs(x[1]))
			pairs.append((w, closest))
		else:
			print(f"warning: no pair for {w}")
	return pairs

def make_transfer(w, d, tag):
	return {
      "type": "transfer",
      "date": datetime.strftime(min(w.date, d.date), '%Y-%m-%dT%H:%M:%SZ'),
      "amount": w.amount,
      "description": f"Transfer from {w.source_name} to {d.destination_name}",
      "currency_code": "USD",
      "category_id": w.category_id,
      "source_id": w.source_id,
      "destination_id": d.destination_id,
      "tags": ["fix-transfers", tag]
    }

def main():
	assert len(sys.argv) == 3, "usage: <script> <server URL> <category ID>"
	
	url = sys.argv[1]
	cat = sys.argv[2]
	token = os.environ.get('FF3_TOKEN')
	assert token is not None, "Must set tokean as environment variable FF3_TOKEN"
	headers = {'Authorization': f"Bearer {token}"}	

	def get(path):
		res = requests.get(f"{url}/{path}", headers=headers)
		res.raise_for_status()
		return res

	def post(path, json):
		res = requests.post(f"{url}/{path}", json=json, headers=headers)
		res.raise_for_status()
		return res

	def delete(path):
		res = requests.delete(f"{url}/{path}", headers=headers)
		res.raise_for_status()
		return res

	print("Requesting info at /api/v1/about")
	res = get(f"api/v1/about")
	print(res.json())

	print(f"Requesting category {cat}")
	res = get(f"api/v1/categories/{cat}")
	cat_name = res.json()['data']['attributes']['name']
	print(f"Category {cat} is: {cat_name}")

	print(f"Requesting transactions for category {cat_name}")
	res = get(f"api/v1/categories/{cat}/transactions")
	transactions = make_transaction_list(res.json())
	while res.json()['meta']['pagination']['current_page'] != res.json()['meta']['pagination']['total_pages']:
		res = get(f"api/v1/categories/{cat}/transactions?page={res.json()['meta']['pagination']['current_page'] + 1}")
		transactions += make_transaction_list(res.json())		

	print(f"Found {len(transactions)} transactions")

	print("Finding pairs to reconcile")
	pairs = make_pairs(transactions)
	print(f"Found {len(pairs)} withdrawal/deposit pairs")

	tag = f"fix-transfers-{int(time())}"
	print(f"Merging pairs into transfers with tag {tag}")

	for i, (w, d) in enumerate(pairs):
		print(f"{i + 1} of {len(pairs)}\n{w}\n{d}")
		body = {
		  "error_if_duplicate_hash": False,
		  "apply_rules": False,
		  "group_title": None,
		  "transactions": [ make_transfer(w, d, tag) ]
		}
		post("api/v1/transactions", body)
		delete(f"api/v1/transactions/{w.transaction_id}")
		delete(f"api/v1/transactions/{d.transaction_id}")

if __name__ == "__main__":
	main()