#!/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.currency_mapping = {} # currency_code -> enabled status 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 get_currencies(self) -> list: """Get all currencies from Firefly III""" try: response = requests.get(f"{self.base_url}/api/v1/currencies", headers=self.headers) if response.status_code == 200: currencies = response.json()["data"] logger.info(f"Found {len(currencies)} currencies in Firefly III") return currencies else: logger.error( f"Failed to fetch currencies: {response.status_code} - {response.text}" ) return [] except Exception as e: logger.error(f"Error fetching currencies: {e}") return [] def initialize_currencies(self) -> bool: """fetch all currencies and populate the mapping cache""" try: currencies = self.get_currencies() for currency in currencies: code = currency["attributes"]["code"] enabled = currency["attributes"]["enabled"] self.currency_mapping[code] = enabled logger.info(f"Initialized {len(self.currency_mapping)} currencies in cache") return True except Exception as e: logger.error(f"Error initializing currencies: {e}") return False def get_accounts(self) -> list: """Get all accounts from Firefly III""" try: response = requests.get(f"{self.base_url}/api/v1/accounts", headers=self.headers) if response.status_code == 200: accounts = response.json()["data"] logger.info(f"Found {len(accounts)} accounts in Firefly III") return accounts else: logger.error(f"Failed to fetch accounts:" f" {response.status_code} - {response.text}") return [] except Exception as e: logger.error(f"Error fetching accounts: {e}") return [] 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.currency_mapping: if self.currency_mapping[currency_code]: return True # currency exists but is disabled - enable it return self._enable_existing_currency(currency_code) # currency doesn't exist - create it return self._create_new_currency(currency_code) def _enable_existing_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}") self.currency_mapping[currency_code] = True 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]: self.currency_mapping[currency_code] = True 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_new_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}" ) self.currency_mapping[currency_code] = True 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]: self.currency_mapping[currency_code] = True 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 class GnuCashToFireflyMigrator: def __init__(self, gnucash_client: GnucashClient, firefly_client: FireflyClient): self.gnucash_client = gnucash_client self.firefly_client = firefly_client self.account_mapping = {} # Account object -> Firefly III ID 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 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 match_existing_accounts(self, gnucash_accounts: list, firefly_accounts: list) -> bool: """fetch existing firefly accounts and match with gnucash accounts""" try: # create name lookup for firefly accounts firefly_by_name = {} for account in firefly_accounts: account_name = account["attributes"]["name"] firefly_by_name[account_name] = account["id"] # match gnucash accounts with existing firefly accounts matched_count = 0 for gnucash_account in gnucash_accounts: if gnucash_account.fullname in firefly_by_name: firefly_id = firefly_by_name[gnucash_account.fullname] self.account_mapping[gnucash_account.fullname] = firefly_id matched_count += 1 logger.info(f"Pre-matched account:" f" {gnucash_account.fullname} -> {firefly_id}") logger.info(f"Pre-matched {matched_count} existing accounts") return True except Exception as e: logger.error(f"Error initializing existing accounts: {e}") return False 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.account_mapping.get(source_split.account.fullname) dest_account_id = self.account_mapping.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...") gnucash_accounts = self.gnucash_client.get_accounts() firefly_accounts = self.firefly_client.get_accounts() # initialize existing firefly accounts for matching if not self.match_existing_accounts(gnucash_accounts, firefly_accounts): logger.warning("Failed to initialize existing accounts," " may create duplicates") # 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.firefly_client.enable_currency(currency) # process remaining accounts that weren't pre-matched created_count = 0 matched_count = len(self.account_mapping) for account in gnucash_accounts: # skip if already matched during initialization if account.fullname in self.account_mapping: 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.info(f"Skipping account with ignored currency:" f" {account.fullname} ({account.commodity.mnemonic})") continue # no existing match, create new account firefly_account = self.create_firefly_account(account) if firefly_account: self.account_mapping[account.fullname] = firefly_account["id"] created_count += 1 logger.info(f"Account migration completed:" f"{matched_count} matched, {created_count} created") return created_count + matched_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.firefly_client.initialize_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( f"Accounts-only migration completed!" f" Migrated {len(self.account_mapping)} accounts" ) return True # handle transactions (unless accounts-only) transaction_count = self.migrate_transactions(transactions_limit) logger.info( f"Migration completed! Migrated {len(self.account_mapping)}" 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()