Created
October 27, 2022 04:22
-
-
Save adyanth/20c004869baf33458e416d4396ca40a8 to your computer and use it in GitHub Desktop.
Import GNUCash gnca exports in FireflyIII
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from lxml import etree | |
from datetime import datetime as dt | |
def parseTwo(node, n, key1, key2, tup=False): | |
ns = {"namespaces": node.nsmap} | |
_key1 = node.find(f"{n}:{key1}", **ns).text | |
_key2 = node.find(f"{n}:{key2}", **ns).text | |
return {_key1: _key2} if not tup else (_key1, _key2) | |
def parseCommodity(node, tup=False): | |
return parseTwo(node, "cmdty", "id", "space", tup) | |
def parseSlot(node, tup=False): | |
return parseTwo(node, "slot", "key", "value", tup) | |
def parseDateTime(node): | |
tsdate = node.find("ts:date", namespaces=node.nsmap).text | |
return dt.strptime(tsdate, "%Y-%m-%d %H:%M:%S %z") | |
def parsePrices(node): | |
ns = {"namespaces": node.nsmap} | |
src = parseCommodity(node.find("price:commodity", **ns), tup=True)[0] | |
dst = parseCommodity(node.find("price:currency", **ns), tup=True)[0] | |
val = node.find("price:value", **ns).text | |
time = parseDateTime(node.find("price:time", **ns)) | |
return { | |
"src": src, | |
"dst": dst, | |
"val": val, | |
"time": time, | |
} | |
def parseAccount(node): | |
ns = {"namespaces": node.nsmap} | |
currency = parseCommodity(node.find("act:commodity", **ns), tup=True)[0] | |
desc = node.find("act:description", **ns) | |
desc = desc.text if desc is not None else "" | |
slots = [] | |
for slot in node.find("act:slots", **ns): | |
slots.append(parseSlot(slot)) | |
_type = node.find("act:type", **ns).text | |
parent = None | |
if _type != "ROOT": | |
parent = node.find("act:parent", **ns).text | |
return { | |
"name": node.find("act:name", **ns).text, | |
"id": node.find("act:id", **ns).text, | |
"type": _type, | |
"currency": currency, | |
"description": desc, | |
"slots": slots, | |
"parent": parent, | |
} | |
def parseSplit(node): | |
ns = {"namespaces": node.nsmap} | |
return { | |
"id": node.find("split:id", **ns).text, | |
"reconciled": node.find("split:reconciled-state", **ns).text, | |
"value": node.find("split:value", **ns).text, | |
"quantity": node.find("split:quantity", **ns).text, | |
"account": node.find("split:account", **ns).text | |
} | |
def parseTranscation(node): | |
ns = {"namespaces": node.nsmap} | |
currency = parseCommodity(node.find("trn:currency", **ns), tup=True)[0] | |
datePosted = parseDateTime(node.find("trn:date-posted", **ns)) | |
dateEntered = parseDateTime(node.find("trn:date-entered", **ns)) | |
slots = [] | |
for slot in node.find("trn:slots", **ns): | |
slots.append(parseSlot(slot)) | |
splits = [] | |
for split in node.find("trn:splits", **ns): | |
splits.append(parseSplit(split)) | |
return { | |
"id": node.find("trn:id", **ns).text, | |
"currency": currency, | |
"datePosted": datePosted, | |
"dateEntered": dateEntered, | |
"description": node.find("trn:description", **ns).text, | |
"slots": slots, | |
"splits": splits, | |
} | |
def translate(file): | |
tree = etree.parse(file) | |
root = tree.getroot() | |
ns = {"namespaces": root.nsmap} | |
nBooks = int(root.find("gnc:count-data", **ns).text) | |
if not nBooks: | |
print("No books found.") | |
book = root.find("gnc:book", **ns) | |
book = etree.ElementTree(book).getroot() | |
# Commodity is Currency types | |
nCommodity = int(book.find("gnc:count-data[@cd:type='commodity']", **ns).text) | |
# Accounts are liabilities, assets, expenses | |
nAccount = int(book.find("gnc:count-data[@cd:type='account']", **ns).text) | |
# Transactions are transactions between accounts | |
nTransaction = int(book.find("gnc:count-data[@cd:type='transaction']", **ns).text) | |
# Prices are exchange rates | |
nPrice = int(book.find("gnc:count-data[@cd:type='price']", **ns).text) | |
commodities = {} | |
for cm in book.findall("gnc:commodity", **ns): | |
commodities.update(parseCommodity(cm)) | |
prices = [] | |
for pr in book.findall("gnc:pricedb/price", **ns): | |
prices.append(parsePrices(pr)) | |
accounts = [] | |
for act in book.findall("gnc:account", **ns): | |
accounts.append(parseAccount(act)) | |
transactions = [] | |
for trn in book.findall("gnc:transaction", **ns): | |
transactions.append(parseTranscation(trn)) | |
return { | |
"commodities": commodities, | |
"prices": prices, | |
"accounts": accounts, | |
"transactions": transactions, | |
} | |
if __name__ == "__main__": | |
import json | |
book = translate("sample.gnca") | |
def serialize(obj): | |
if isinstance(obj, dt): | |
return obj.isoformat() | |
raise TypeError ("Type %s not serializable" % type(obj)) | |
with open("sample.json", "w") as f: | |
json.dump(book, f, indent=4, default=serialize) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import requests | |
baseUrl = "https://firefly.domain.com" | |
token = "Personal Access Token from FireflyIII->Profile->OAuth" | |
auth = {"Authorization": f"Bearer {token}"} | |
def callApi(path, method="POST", params={}, body={}, fail=True): | |
headers = auth | |
res = requests.request( | |
method, | |
f"{baseUrl}/api/v1/{path}", | |
headers=headers, | |
params=params, | |
json=body, | |
) | |
if fail: | |
res.raise_for_status() | |
return res | |
accountMap = {} | |
def enableCurrencies(currencies): | |
for cur in currencies: | |
res = callApi(f"currencies/{cur}/enable") | |
if res.status_code == 204: | |
print(f"Enabled currency: {cur}") | |
print(f"Currency {cur} already enabled") | |
def deleteAllAccounts(): | |
i = 1 | |
res = callApi("accounts", method="GET", params={"page": i}) | |
res = res.json() | |
while res["data"]: | |
for act in res["data"]: | |
callApi(f"accounts/{act['id']}", method="DELETE") | |
print(f"Deleted account: {act['attributes']['name']}") | |
res = callApi("accounts", method="GET", params={"page": i}) | |
res = res.json() | |
def addAccounts(accounts): | |
for act in accounts: | |
oldid = act["oldid"] | |
del act["oldid"] | |
res = callApi("accounts", body=act) | |
try: | |
res = res.json() | |
accountMap[oldid]["id"] = res["data"]["id"] | |
accountMap[oldid]["new"] = res["data"]["attributes"] | |
except requests.exceptions.JSONDecodeError: | |
print(f"Account {act['name']} already present, please delete and try again") | |
break | |
print(f"Added account: {act['name']}") | |
def addTransactions(txns): | |
for txn in txns: | |
body = { | |
"error_if_duplicate_hash": True, | |
"group_title": txn["description"], | |
"transactions": [txn] | |
} | |
try: | |
callApi("transactions", method="POST", body=body).json() | |
except Exception as e: | |
print(f"Transaction {txn['description']} errored, body: {body}") | |
raise | |
print(f"Added Transaction: {txn['description']}") | |
def translateAccount(act): | |
typeMap = { | |
"ASSET": "asset", | |
"CASH": "asset", | |
"BANK": "asset", | |
"EXPENSE": "expense", | |
"INCOME": "revenue", | |
"LIABILITY": "liability", | |
"CREDIT": "asset", | |
} | |
roleMap = { | |
"ASSET": "defaultAsset", | |
"CASH": "cashWalletAsset", | |
"BANK": "defaultAsset", | |
"EXPENSE": "expense", | |
"INCOME": "revenue", | |
"LIABILITY": "liability", | |
"CREDIT": "ccAsset", | |
} | |
newAct = { | |
"oldid": act["id"], | |
"name": act["name"], | |
"type": typeMap[act["type"]], | |
"currency_code": act["currency"], | |
"include_net_worth": True, | |
"notes": act["description"], | |
} | |
if newAct["type"] == "asset": | |
newAct.update({ | |
"account_role": roleMap[act["type"]], | |
}) | |
if newAct["type"] == "liability": | |
newAct.update({ | |
"liability_type": "loan", | |
"liability_direction": "credit", | |
"interest": "5.3", | |
"interest_period": "monthly", | |
}) | |
if newAct.get("account_role") == "ccAsset": | |
newAct.update({ | |
"credit_card_type": "monthlyFull", | |
"monthly_payment_date": "1900-01-18", | |
}) | |
accountMap[act["id"]] = {"old": act, "new": newAct} | |
return newAct | |
def evaluateTransaction(src, dst): | |
# I use eval, sue me :) | |
src["value"] = eval(src["value"]) | |
src["quantity"] = eval(src["quantity"]) | |
dst["value"] = eval(dst["value"]) | |
dst["quantity"] = eval(dst["quantity"]) | |
if dst["value"] < 0: | |
src, dst = dst, src | |
sa = accountMap[src["account"]] | |
da = accountMap[dst["account"]] | |
cc = da["old"]["currency"] | |
ca = dst["quantity"] | |
ttype = "deposit" if da["new"]["type"] == "asset" else "transfer" | |
if sa["new"]["type"] == "asset": | |
ttype = "transfer" if da["new"]["type"] == "asset" else "withdrawal" | |
elif da["new"]["type"] == "asset": | |
ttype = "transfer" if sa["new"]["type"] == "revenue" else "deposit" | |
dant = da["new"]["type"] | |
sant = sa["new"]["type"] | |
if dant == sant: | |
ttype = "transfer" | |
elif dant == "asset": | |
ttype = "deposit" | |
elif dant == "revenue": | |
raise ValueError("Destination cannot be a revenue account!") | |
elif sant == "asset": | |
ttype = "withdrawal" | |
else: | |
raise ValueError("Neither source nor destination is asset!") | |
out = { | |
"type": ttype, | |
"amount": f"{ca:.2f}", | |
"currency_code": cc, | |
"source_id": sa["id"], | |
"destination_id": da["id"], | |
} | |
# Firefly: | |
# deposit -> source is foreign | |
# withdrawal/transfer -> destination is foreign | |
# GNUCash Source is foreign currency | |
if abs(src["value"] - src["quantity"]) > 0.001: | |
if ttype == "deposit": | |
fc = sa["old"]["currency"] | |
fa = abs(src["quantity"]) | |
cc = da["old"]["currency"] | |
ca = dst["quantity"] | |
else: | |
fc = da["old"]["currency"] | |
fa = dst["quantity"] | |
cc = sa["old"]["currency"] | |
ca = abs(src["quantity"]) | |
out.update({ | |
"amount": f"{ca:.2f}", | |
"currency_code": cc, | |
"foreign_currency_code": fc, | |
"foreign_amount": fa, | |
}) | |
# GNUCash Destination is foreign currrency | |
if abs(dst["value"] - dst["quantity"]) > 0.001: | |
if out.get("foreign_amount"): | |
raise ValueError("Both source and destination accounts cannot have foreign currencies") | |
if ttype == "deposit": | |
fc = sa["old"]["currency"] | |
fa = abs(src["quantity"]) | |
cc = da["old"]["currency"] | |
ca = dst["quantity"] | |
else: | |
fc = da["old"]["currency"] | |
fa = dst["quantity"] | |
cc = sa["old"]["currency"] | |
ca = abs(src["quantity"]) | |
out.update({ | |
"amount": f"{ca:.2f}", | |
"currency_code": cc, | |
"foreign_currency_code": fc, | |
"foreign_amount": fa, | |
}) | |
return out | |
def translateTransaction(txn): | |
newTxn = { | |
"date": txn["datePosted"], | |
"description": txn["description"], | |
"reconciled": False, | |
} | |
try: | |
newTxn.update(evaluateTransaction(*txn["splits"])) | |
except ValueError as e: | |
print(f"Failed to evaulate transaction {txn['description']}: {e}") | |
raise | |
if len(txn["slots"]) > 0: | |
try: | |
newTxn.update({ | |
"notes": txn["slots"][0]["notes"] | |
}) | |
except Exception: | |
pass | |
return newTxn | |
def translate(data): | |
print("Enabling currencies...") | |
enableCurrencies(list(data["commodities"].keys())) | |
print("Done.") | |
print() | |
print("Adding accounts...") | |
accounts = [] | |
for act in data.get("accounts", []): | |
if act["type"] == "ROOT": | |
continue | |
accounts.append(translateAccount(act)) | |
addAccounts(accounts) | |
print("Done.") | |
print() | |
print("Adding transactions...") | |
txns = [] | |
for txn in data.get("transactions", []): | |
txns.append(translateTransaction(txn)) | |
addTransactions(txns) | |
print("Done.") | |
if __name__ == "__main__": | |
import json | |
with open("sample.json") as f: | |
data = json.load(f) | |
print("Continuing will wipe your firefly instance...") | |
input() | |
deleteAllAccounts() | |
print("Done.") | |
print() | |
translate(data) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment