alexklibisz · January 16, 2025 09:42 · Mar 21, 2021 · Mar 21, 2021 · Mar 21, 2021
diff --git a/fixtransfers.py b/fixtransfers.py
@@ -0,0 +1,133 @@
+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()