From 6920328ebca8a2f7f4ff3de369b44170b96acef9 Mon Sep 17 00:00:00 2001 From: Maks Snegov Date: Fri, 15 Aug 2025 22:24:00 -0700 Subject: [PATCH] Support piggy-banks --- g2f.py | 241 ++++++++++++++++++++++++++++++++++++++------------------- 1 file changed, 162 insertions(+), 79 deletions(-) diff --git a/g2f.py b/g2f.py index 2a68d19..b811f44 100644 --- a/g2f.py +++ b/g2f.py @@ -23,7 +23,7 @@ import piecash # Configure logging logging.basicConfig(level=logging.INFO, - format="%(asctime)s - %(levelname)s - %(message)s") + format="%(asctime)s|%(levelname)s|%(message)s") logger = logging.getLogger(__name__) ACCOUNT_TYPE_MAPPING = { @@ -103,7 +103,6 @@ class FireflyClient: "Content-Type": "application/json", "Accept": "application/json", } - self.currency_mapping = {} # currency_code -> enabled status self.dry_run = dry_run def test_connection(self) -> bool: @@ -169,42 +168,18 @@ class FireflyClient: """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 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 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: @@ -213,7 +188,6 @@ class FireflyClient: 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: @@ -226,7 +200,7 @@ class FireflyClient: logger.error(f"Error enabling currency {currency_code}: {e}") return False - def _create_new_currency(self, currency_code: str) -> bool: + def create_currency(self, currency_code: str) -> bool: """create a new currency""" currency_data = { "code": currency_code, @@ -240,7 +214,6 @@ class FireflyClient: f"Dry run: would create currency {currency_code} with data:" f" {currency_data}" ) - self.currency_mapping[currency_code] = True return True try: @@ -251,7 +224,6 @@ class FireflyClient: ) if response.status_code in [200, 201]: - self.currency_mapping[currency_code] = True logger.info(f"Created currency: {currency_code}") return True else: @@ -330,12 +302,92 @@ class FireflyClient: 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.account_mapping = {} # Account object -> Firefly III ID + 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""" @@ -396,30 +448,49 @@ class GnuCashToFireflyMigrator: return self.firefly_client.create_account(account_data) - def match_existing_accounts(self, gnucash_accounts: list, firefly_accounts: list) -> bool: + 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""" - 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"] + matched_count = 0 + ff_accounts = self.firefly_client.get_accounts() + ff_piggy_banks = self.firefly_client.get_piggy_banks() - # 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}") + 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.info(f"Pre-matched {matched_count} existing accounts") - return True - except Exception as e: - logger.error(f"Error initializing existing accounts: {e}") - return False + 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""" @@ -460,8 +531,8 @@ class GnuCashToFireflyMigrator: ) return False - source_account_id = self.account_mapping.get(source_split.account.fullname) - dest_account_id = self.account_mapping.get(dest_split.account.fullname) + 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:" @@ -510,14 +581,9 @@ class GnuCashToFireflyMigrator: """Migrate all accounts from GnuCash to Firefly III""" logger.info("Starting account migration...") + self.match_existing_accounts() + 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: @@ -531,15 +597,17 @@ class GnuCashToFireflyMigrator: unique_currencies.add(account.commodity.mnemonic) for currency in unique_currencies: - self.firefly_client.enable_currency(currency) + self.enable_currency(currency) # process remaining accounts that weren't pre-matched created_count = 0 - matched_count = len(self.account_mapping) + 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.account_mapping: + if account.fullname in self.accounts: continue if account.placeholder: @@ -547,19 +615,37 @@ class GnuCashToFireflyMigrator: 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})") + 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.account_mapping[account.fullname] = firefly_account["id"] + self.accounts[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 + 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""" @@ -588,7 +674,7 @@ class GnuCashToFireflyMigrator: return False # initialize currencies cache - if not self.firefly_client.initialize_currencies(): + if not self.load_currencies(): logger.warning("Failed to initialize currencies cache," " will fetch individually") @@ -598,16 +684,13 @@ class GnuCashToFireflyMigrator: return False if accounts_only: - logger.info( - f"Accounts-only migration completed!" - f" Migrated {len(self.account_mapping)} accounts" - ) + 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.account_mapping)}" + f"Migration completed! Migrated {len(self.accounts)}" f" accounts and {transaction_count} transactions" ) return True