gnucash2firefly/g2f.py

#!/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 _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 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"""
        return self._paginated_get("/api/v1/accounts", "accounts")

    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

        # add required fields for liability accounts
        elif firefly_type == "liability":
            # liability_type - mandatory when type is liability
            if account.type in ["LOAN"] or "loan" in account.name.lower():
                account_data["liability_type"] = "loan"
            elif "mortgage" in account.fullname.lower():
                account_data["liability_type"] = "mortgage"
            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 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()