757 lines
28 KiB
Python
757 lines
28 KiB
Python
#!/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()
|