Import GNUCash gnca exports in FireflyIII

gncaToJson.py

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)

jsonToFireflyIII.py

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)