Buxfer->Firefly importer

firefly_import.py

#!/usr/bin/env python
# A little script that takes buxfer export csv for whole years and imports them into firefly.
token = "your_firefly_token_here"

headers = {
    "accept": "application/vnd.api+json",
    "Authorization": f"Bearer {token}",
    "Content-Type": "application/json",
}

count = 0

# Iterate through yearly buxfer exports.
for year in range(2016, 2024):
    with open(
        f"/path_to/buxfer_backups/{ year }_transactions.csv",
    ) as csvfile:
        transactions = csv.DictReader(csvfile)

        for transaction in transactions:
            # Ignore all transactions that aren't expenses/income
            if transaction["Type"] == "Income":
                transaction_type = "deposit"
            elif transaction["Type"] == "Expense":
                transaction_type = "withdrawal"
            else:
                continue

            payload = {
                "error_if_duplicate_hash": True,
                "transactions": [
                    {
                        "type": transaction_type,
                        "date": transaction["Date"] + "T12:00:00+02:00",
                        "amount": transaction["Amount"].replace("-", ""),
                        "description": transaction["Description"],
                        "category_name": transaction["Tags"],
                        "reconciled": False,
                    }
                ],
            }

            tid = transaction.get("ID")
            if tid is not None:
                payload["transactions"][0]["external_id"] = tid

            if transaction_type == "deposit":
                payload["transactions"][0]["destination_name"] = 'Apobank'
            elif transaction_type == "withdrawal":
                payload["transactions"][0]["source_name"] = 'Apobank'

            response = requests.post("http://localhost:8081/api/v1/transactions", headers=headers, json=payload)

            if response.status_code == 200:
                continue

            if response.status_code == 422:
                continue

            count += 1