#!/usr/bin/env python3 """ GnuCash to Firefly III Migration Script This script automates the migration of accounts and transactions from a GnuCash SQLite database to Firefly III using the Firefly III API. Usage: python gnucash_to_firefly.py --gnucash-db path/to/gnucash.gnucash --firefly-url http://localhost --firefly-token your-token Author: Migration Script Generator """ import argparse import logging import os import sys from typing import Optional import requests import piecash # Configure logging logging.basicConfig(level=logging.INFO, format="%(asctime)s|%(levelname)s|%(message)s") logger = logging.getLogger(__name__) ACCOUNT_TYPE_MAPPING = { "ASSET": "asset", "BANK": "asset", "CASH": "asset", "CHECKING": "asset", "SAVINGS": "asset", "MONEYMRKT": "asset", "INVESTMENT": "asset", "STOCK": "asset", "MUTUAL": "asset", "LIABILITY": "liability", "CREDIT": "liability", "CREDITCARD": "liability", "LOAN": "liability", "EXPENSE": "expense", "INCOME": "revenue", "EQUITY": "asset", "RECEIVABLE": "asset", "PAYABLE": "liability", } IGNORE_CURRENCIES = [ "VFV.TO", "TEC.TO", ] class GnucashClient: def __init__(self, db_path: str): self.db_path = db_path self.book = None def connect(self) -> bool: """Connect to GnuCash database using piecash""" try: self.book = piecash.open_book(self.db_path, readonly=True) logger.info(f"Connected to GnuCash database: {self.db_path}") return True except Exception as e: logger.error(f"Failed to connect to GnuCash database: {e}") return False def get_accounts(self) -> list: """Get all accounts from the GnuCash book""" if not self.book: logger.error("Not connected to GnuCash database") return [] accounts = [acc for acc in self.book.accounts if acc.type != "ROOT"] logger.info(f"Found {len(accounts)} accounts in GnuCash") return accounts def get_transactions(self, limit: int = 0) -> list: """Get all transactions from the GnuCash book""" if not self.book: logger.error("Not connected to GnuCash database") return [] # get all transactions, sorted by date all_transactions = sorted(self.book.transactions, key=lambda t: t.post_date) # apply limit if specified if limit: all_transactions = all_transactions[:limit] logger.info(f"Found {len(all_transactions)} transactions in GnuCash") return all_transactions class FireflyClient: def __init__(self, base_url: str, token: str, dry_run: bool = False): self.base_url = base_url.rstrip("/") self.headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Accept": "application/json", } self.dry_run = dry_run def test_connection(self) -> bool: """Test connection to Firefly III API""" try: response = requests.get(f"{self.base_url}/api/v1/about", headers=self.headers) if response.status_code == 200: logger.info("Successfully connected to Firefly III API") return True else: logger.error(f"Failed to connect to Firefly III:" f" {response.status_code}") return False except Exception as e: logger.error(f"Error connecting to Firefly III: {e}") return False def _paginated_get(self, endpoint: str, resource_name: str) -> list: """generic method to fetch all items from a paginated endpoint""" all_items = [] page = 1 try: while True: response = requests.get( f"{self.base_url}{endpoint}", headers=self.headers, params={"page": page, "limit": 50} ) if response.status_code != 200: logger.error(f"Failed to fetch {resource_name} page {page}:" f" {response.status_code} - {response.text}") break data = response.json() items_page = data["data"] all_items.extend(items_page) # check pagination info meta = data.get("meta", {}) pagination = meta.get("pagination", {}) current_page = pagination.get("current_page", 1) total_pages = pagination.get("total_pages", 1) logger.debug(f"Fetched {resource_name} page {current_page} of {total_pages}" f" ({len(items_page)} items)") if current_page >= total_pages: break page += 1 logger.info(f"Found {len(all_items)} {resource_name} in Firefly III") return all_items except Exception as e: logger.error(f"Error fetching {resource_name}: {e}") return all_items def get_currencies(self) -> list: """Get all currencies from Firefly III""" return self._paginated_get("/api/v1/currencies", "currencies") def get_accounts(self) -> list: """Get all accounts from Firefly III""" return self._paginated_get("/api/v1/accounts", "accounts") def get_piggy_banks(self) -> list: """Get all piggy banks from Firefly III""" return self._paginated_get("/api/v1/piggy-banks", "piggy banks") def enable_currency(self, currency_code: str) -> bool: """enable an existing but disabled currency""" if self.dry_run: logger.info(f"Dry run: would enable existing currency {currency_code}") return True try: response = requests.post( f"{self.base_url}/api/v1/currencies/{currency_code}/enable", headers=self.headers, ) if response.status_code in [200, 201]: logger.info(f"Enabled existing currency: {currency_code}") return True else: logger.warning( f"Could not enable currency {currency_code}: " f"{response.status_code} {response.text}" ) return False except Exception as e: logger.error(f"Error enabling currency {currency_code}: {e}") return False def create_currency(self, currency_code: str) -> bool: """create a new currency""" currency_data = { "code": currency_code, "name": currency_code, # use code as name fallback "symbol": currency_code, # use code as symbol fallback "enabled": True, } if self.dry_run: logger.info( f"Dry run: would create currency {currency_code} with data:" f" {currency_data}" ) return True try: response = requests.post( f"{self.base_url}/api/v1/currencies", headers=self.headers, json=currency_data, ) if response.status_code in [200, 201]: logger.info(f"Created currency: {currency_code}") return True else: logger.warning( f"Could not create currency {currency_code}: " f"{response.status_code} {response.text}" ) return False except Exception as e: logger.error(f"Error creating currency {currency_code}: {e}") return False def create_account(self, account_data: dict) -> Optional[dict]: """Create an account in Firefly III""" if self.dry_run: logger.info( f"Dry run: would create account {account_data.get('name')}" f" with data: {account_data}" ) return {"id": f"dry-run-{account_data.get('name')}", "data": account_data} try: response = requests.post( f"{self.base_url}/api/v1/accounts", headers=self.headers, json=account_data, ) if response.status_code in [200, 201]: firefly_account = response.json()["data"] logger.info( f"Created account: {firefly_account['attributes']['name']}" f" -> {firefly_account['id']}" ) return firefly_account else: logger.error( f"Failed to create account {account_data.get('name')}:" f" {response.status_code} - {response.text}" ) return None except Exception as e: logger.error(f"Error creating account: {e}") return None def create_transaction(self, transaction_data: dict) -> bool: """Create a transaction in Firefly III""" if self.dry_run: logger.info(f"Dry run: would create transaction with data:" f" {transaction_data}") return True try: response = requests.post( f"{self.base_url}/api/v1/transactions", headers=self.headers, json=transaction_data, ) if response.status_code in [200, 201]: logger.info( f"Created transaction:" f" {transaction_data['transactions'][0]['description']}" ) return True else: logger.error( f"Failed to create transaction:" f" {response.status_code} - {response.text}" ) return False except Exception as e: logger.error(f"Error creating transaction: {e}") return False def create_piggy_bank(self, piggy_bank_data: dict) -> Optional[dict]: """Create a piggy bank in Firefly III""" if self.dry_run: logger.info( f"Dry run: would create piggy bank {piggy_bank_data.get('name')}" f" with data: {piggy_bank_data}" ) return {"id": f"dry-run-piggy-{piggy_bank_data.get('name')}", "data": piggy_bank_data} try: response = requests.post( f"{self.base_url}/api/v1/piggy-banks", headers=self.headers, json=piggy_bank_data, ) if response.status_code in [200, 201]: firefly_piggy_bank = response.json()["data"] logger.info( f"Created piggy bank: {firefly_piggy_bank['attributes']['name']}" f" -> {firefly_piggy_bank['id']}" ) return firefly_piggy_bank else: logger.error( f"Failed to create piggy bank {piggy_bank_data.get('name')}:" f" {response.status_code} - {response.text}" ) return None except Exception as e: logger.error(f"Error creating piggy bank: {e}") return None class GnuCashToFireflyMigrator: def __init__(self, gnucash_client: GnucashClient, firefly_client: FireflyClient): self.gnucash_client = gnucash_client self.firefly_client = firefly_client self.accounts = {} # account full name -> Firefly III ID self._currencies_cache = {} # currency_code -> firefly currency attributes @property def currencies(self): """Cache for currencies to avoid multiple API calls""" if not self._currencies_cache: self.load_currencies() return self._currencies_cache def _clear_currencies_cache(self): """Clear the currencies cache""" self._currencies_cache = {} logger.debug("Cleared currencies cache") def load_currencies(self) -> bool: """fetch all currencies and populate the mapping cache""" try: currencies = self.firefly_client.get_currencies() for currency in currencies: code = currency["attributes"]["code"] self._currencies_cache[code] = currency["attributes"] self._currencies_cache[code]["id"] = currency["id"] logger.info(f"Initialized {len(self._currencies_cache)} currencies in cache") return True except Exception as e: logger.error(f"Error initializing currencies: {e}") return False def enable_currency(self, currency_code: str) -> bool: """Enable a currency in Firefly III""" # check if currency already exists and is enabled if currency_code in self.currencies: if self.currencies[currency_code]["enabled"]: return True # currency exists but is disabled - enable it success = self.firefly_client.enable_currency(currency_code) if success: self._clear_currencies_cache() return success # currency doesn't exist - create it success = self.firefly_client.create_currency(currency_code) if success: self._clear_currencies_cache() return success def create_firefly_account(self, account: piecash.Account) -> Optional[dict]: """Create an account in Firefly III using piecash account object""" # get currency code currency_code = account.commodity.mnemonic if account.commodity else "USD" firefly_type = ACCOUNT_TYPE_MAPPING.get(account.type) if not firefly_type: logger.warning( f"Unsupported account type {account.type} for account {account.fullname}" ) return None account_data = { "name": account.fullname, "type": firefly_type, "currency_code": currency_code, "active": not account.hidden, "include_net_worth": firefly_type in ["asset", "liability"], } # add account_role for asset accounts (required by Firefly III) if firefly_type == "asset": # determine appropriate role based on account type and name if account.type in ["INVESTMENT", "STOCK", "MUTUAL", "BANK"]: account_data["account_role"] = "savingAsset" elif account.type in ["CASH"]: account_data["account_role"] = "cashWalletAsset" elif account.type in ["CHECKING"]: account_data["account_role"] = "defaultAsset" elif account.type in ["SAVINGS"]: account_data["account_role"] = "savingAsset" else: account_data["account_role"] = "defaultAsset" # default fallback # add required fields for liability accounts elif firefly_type == "liability": # liability_type - mandatory when type is liability if "mortgage" in account.fullname.lower(): account_data["liability_type"] = "mortgage" elif account.type in ["LOAN"] or "loan" in account.name.lower(): account_data["liability_type"] = "loan" else: account_data["liability_type"] = "debt" # default for credit cards, etc. # liability_direction - required by API # "debit" means you owe the debt (most common case) account_data["liability_direction"] = "debit" # interest - mandatory when type is liability, defaults to "0" account_data["interest"] = "0" if account.code: account_data["account_number"] = account.code if account.description: account_data["notes"] = account.description return self.firefly_client.create_account(account_data) def create_firefly_piggy_bank(self, account: piecash.Account, parent_account_id: str) -> Optional[dict]: """Create a piggy bank in Firefly III using piecash account object""" # Get currency code from parent account currency_code = account.commodity.mnemonic if account.commodity else "USD" # Calculate current amount from account balance current_amount = abs(float(account.get_balance())) piggy_bank_data = { "name": account.fullname, # Use short name, not fullname for piggy banks "accounts": [{ "account_id": parent_account_id, }], "transaction_currency_code": currency_code, "target_amount": current_amount * 2, # TODO fetch from the description "current_amount": current_amount, } if account.description: piggy_bank_data["notes"] = account.description return self.firefly_client.create_piggy_bank(piggy_bank_data) def match_existing_accounts(self): """fetch existing firefly accounts and match with gnucash accounts""" matched_count = 0 ff_accounts = self.firefly_client.get_accounts() ff_piggy_banks = self.firefly_client.get_piggy_banks() ff_acc: dict for ff_acc in ff_accounts + ff_piggy_banks: ff_acc_data = ff_acc["attributes"] ff_acc_data["type"] = ff_acc["type"] ff_acc_data["id"] = ff_acc["id"] try: gc_acc = self.gnucash_client.book.accounts(fullname=ff_acc_data["name"]) except KeyError: continue matched_count += 1 self.accounts[gc_acc.fullname] = ff_acc_data["id"] logger.debug(f"Pre-matched account: {gc_acc.fullname}") logger.debug(f"Pre-matched {matched_count} existing accounts") def create_firefly_transaction(self, transaction: piecash.Transaction) -> bool: """Create a transaction in Firefly III using piecash transaction object""" splits = transaction.splits # skip transactions with only one split if len(splits) < 2: logger.warning( f"Skipping transaction with less than 2 splits:" f" {transaction.description}" ) return False # for simple two-split transactions if len(splits) == 2: return self.create_simple_transaction(transaction) else: return self.create_split_transaction(transaction) def create_simple_transaction(self, transaction: piecash.Transaction) -> bool: """Create a simple two-account transaction using piecash objects""" splits = transaction.splits # find source (negative amount) and destination (positive amount) source_split = None dest_split = None for split in splits: if float(split.value) < 0: source_split = split elif float(split.value) > 0: dest_split = split if not source_split or not dest_split: logger.warning( f"Could not determine source/destination for transaction:" f" {transaction.description}" ) return False source_account_id = self.accounts.get(source_split.account.fullname) dest_account_id = self.accounts.get(dest_split.account.fullname) if not source_account_id or not dest_account_id: logger.warning(f"Missing account mapping for transaction:" f" {transaction.description}") return False # get currency code currency_code = transaction.currency.mnemonic if transaction.currency else "USD" # format date formatted_date = transaction.post_date.strftime("%Y-%m-%d") transaction_data = { "transactions": [ { "type": "transfer", # let Firefly determine the actual type "date": formatted_date, "amount": str(abs(float(dest_split.value))), "description": transaction.description or "Imported from GnuCash", "source_id": str(source_account_id), "destination_id": str(dest_account_id), "currency_code": currency_code, } ] } # add memos if available memos = [] if source_split.memo: memos.append(f"Source: {source_split.memo}") if dest_split.memo: memos.append(f"Dest: {dest_split.memo}") if memos: transaction_data["transactions"][0]["notes"] = "; ".join(memos) return self.firefly_client.create_transaction(transaction_data) def create_split_transaction(self, transaction: piecash.Transaction) -> bool: """Create a multi-split transaction (not fully implemented - needs complex handling)""" logger.warning( f"Multi-split transactions not fully supported yet: {transaction.description}" ) return False def migrate_accounts(self) -> int: """Migrate all accounts from GnuCash to Firefly III""" logger.info("Starting account migration...") self.match_existing_accounts() gnucash_accounts = self.gnucash_client.get_accounts() # enable currencies first unique_currencies = set() for account in gnucash_accounts: if account.commodity: # skip ignored currencies if account.commodity.mnemonic in IGNORE_CURRENCIES: logger.debug(f"Skipping ignored currency:" f" {account.commodity.mnemonic}") continue # add to unique currencies set unique_currencies.add(account.commodity.mnemonic) for currency in unique_currencies: self.enable_currency(currency) # process remaining accounts that weren't pre-matched created_count = 0 piggy_bank_count = 0 matched_count = len(self.accounts) gnucash_accounts = sorted(gnucash_accounts, key=lambda x: x.fullname) for account in gnucash_accounts: # skip if already matched during initialization if account.fullname in self.accounts: continue if account.placeholder: logger.debug(f"Skipping placeholder account: {account.fullname}") continue # skip accounts with ignored currencies if account.commodity and account.commodity.mnemonic in IGNORE_CURRENCIES: logger.debug(f"Skipping account with ignored currency:" f" {account.fullname} ({account.commodity.mnemonic})") continue # accounts with non-placeholder parents are piggy-banks if (account.parent and account.parent.type != "ROOT" and not account.parent.placeholder): logger.debug(f"Found piggy-bank account: {account.fullname}") # Get parent account ID from mapping parent_account_id = self.accounts.get(account.parent.fullname) if parent_account_id: firefly_piggy_bank = self.create_firefly_piggy_bank(account, parent_account_id) if firefly_piggy_bank: piggy_bank_count += 1 else: logger.warning(f"Parent account not found for piggy bank: {account.fullname}") continue # no existing match, create new account firefly_account = self.create_firefly_account(account) if firefly_account: self.accounts[account.fullname] = firefly_account["id"] created_count += 1 logger.info(f"Account migration completed:" f" {matched_count} matched," f" {created_count} created," f" {piggy_bank_count} piggy banks created") return matched_count + created_count + piggy_bank_count def migrate_transactions(self, limit: int = 0) -> int: """Migrate transactions from GnuCash to Firefly III""" logger.info("Starting transaction migration...") transactions = self.gnucash_client.get_transactions(limit) success_count = 0 for transaction in transactions: if self.create_firefly_transaction(transaction): success_count += 1 logger.info(f"Successfully created {success_count} out of {len(transactions)} transactions") return success_count def run_migration(self, transactions_limit: int = 0, accounts_only=False) -> bool: """Run migration process""" logger.info("Starting GnuCash to Firefly III migration...") # test connections if not self.gnucash_client.connect(): return False if not self.firefly_client.test_connection(): self.gnucash_client.book.close() return False # initialize currencies cache if not self.load_currencies(): logger.warning("Failed to initialize currencies cache," " will fetch individually") # handle accounts if not self.migrate_accounts(): logger.error("Account migration failed!") return False if accounts_only: logger.info("Accounts-only migration completed!") return True # handle transactions (unless accounts-only) transaction_count = self.migrate_transactions(transactions_limit) logger.info( f"Migration completed! Migrated {len(self.accounts)}" f" accounts and {transaction_count} transactions" ) return True def main(): parser = argparse.ArgumentParser(description="Migrate data from GnuCash to Firefly III") parser.add_argument("--gnucash-db", required=True, help="Path to GnuCash SQLite database file") parser.add_argument( "--firefly-url", required=True, help="Firefly III base URL (e.g., http://localhost)", ) parser.add_argument("--firefly-token", help="Firefly III API access token") parser.add_argument( "--transactions-limit", type=int, help="Limit number of transactions to migrate (0 for all)", default=0, ) parser.add_argument( "--accounts-only", action="store_true", help="Only migrate accounts, skip transactions", ) parser.add_argument( "--dry-run", action="store_true", help="Perform a dry run without making any changes", ) args = parser.parse_args() # get token from argument or environment variable firefly_token = args.firefly_token or os.getenv("FIREFLY_TOKEN") if not firefly_token: logger.error( "No token provided. Use --firefly-token or" " set FIREFLY_TOKEN environment variable" ) sys.exit(1) # create client instances gnucash_client = GnucashClient(args.gnucash_db) firefly_client = FireflyClient(args.firefly_url, firefly_token, dry_run=args.dry_run) migrator = GnuCashToFireflyMigrator(gnucash_client, firefly_client) success = migrator.run_migration(args.transactions_limit, args.accounts_only) if success: logger.info("Migration completed successfully!") sys.exit(0) else: logger.error("Migration failed!") sys.exit(1) if __name__ == "__main__": main()