#!/usr/bin/env python3 """ import_firefly_rules.py Imports rules exported from a Firefly III instance (via the artisan command: docker exec firefly_iii_app php artisan firefly-iii:export-data \\ --export-rules --export_directory=./storage/export ) into a NEW Firefly III instance, using the REST API. -------------------------------------------------------------------------- KNOWN EXPORT QUIRK (as of Firefly III's export-data command, checked against a real export in Aug 2026): For "rule" summary rows, the CSV writer accidentally writes the RULE GROUP's title into the `title` column, which pushes every subsequent value one column to the right for that row only ("trigger"/"action" rows are fine). Header says: ... group_id, title, description, order, active, stop_processing, strict, trigger_type, ... Rule row actually holds: ... group_id, , , , , , , , (empty), ... This script detects `row_contains == "rule"` and reads the shifted positions directly (by index) rather than trusting the header labels for those rows. Trigger/action rows are unaffected and are read normally. If a future export fixes this bug, re-run with --no-shift-fix and the script will read rule rows using the header labels as-is. -------------------------------------------------------------------------- USAGE: export FIREFLY_URL="https://firefly.example.com" export FIREFLY_TOKEN="eyJ0eXAiOiJKV1..." # Personal Access Token (new instance) python3 import_firefly_rules.py --csv rules.csv # dry run by default? no, see --dry-run python3 import_firefly_rules.py --csv rules.csv --dry-run # preview only, no API calls python3 import_firefly_rules.py --csv rules.csv --apply # actually create rule groups + rules python3 import_firefly_rules.py --csv rules.csv --apply --skip-existing A Personal Access Token can be generated on the NEW instance under: Options -> Profile -> OAuth -> Personal Access Tokens """ import argparse import csv import json import os import sys import time from collections import defaultdict, OrderedDict try: import requests except ImportError: sys.exit("This script needs the 'requests' package: pip install requests") TRUE_VALUES = {"1", "true", "yes", "y"} def as_bool(value): return str(value).strip().lower() in TRUE_VALUES def as_int(value, default=None): try: return int(str(value).strip()) except (ValueError, TypeError): return default # -------------------------------------------------------------------------- # Parsing # -------------------------------------------------------------------------- def parse_csv(path, apply_shift_fix=True): """ Returns an OrderedDict: rule_id -> rule_dict, where rule_dict has: group_title, title, description, order, active, stop_processing, strict, trigger (store-journal/update-journal), triggers (list), actions (list) Rules are returned in the order first encountered in the file. """ rules = OrderedDict() with open(path, newline="", encoding="utf-8-sig") as f: reader = csv.reader(f) header = next(reader) idx = {name: i for i, name in enumerate(header)} for row in reader: if not row or len(row) < 3: continue rule_id = row[idx["rule_id"]] row_type = row[idx["row_contains"]] if rule_id not in rules: rules[rule_id] = { "group_title": "Default rules", "title": f"Imported rule {rule_id}", "description": "", "order": None, "active": True, "stop_processing": False, "strict": True, "trigger": "store-journal", "triggers": [], "actions": [], } r = rules[rule_id] if row_type == "rule": if apply_shift_fix: # Positions are shifted right by one starting at # group_id+1 (see module docstring). group_id itself # (idx["group_id"]) is NOT shifted. group_id_pos = idx["group_id"] group_title = row[group_id_pos + 1] if len(row) > group_id_pos + 1 else "" title = row[group_id_pos + 2] if len(row) > group_id_pos + 2 else "" description = row[group_id_pos + 3] if len(row) > group_id_pos + 3 else "" order = row[group_id_pos + 4] if len(row) > group_id_pos + 4 else "" active = row[group_id_pos + 5] if len(row) > group_id_pos + 5 else "" stop_processing = row[group_id_pos + 6] if len(row) > group_id_pos + 6 else "" strict = row[group_id_pos + 7] if len(row) > group_id_pos + 7 else "" else: group_title = row[idx["title"]] title = row[idx["description"]] description = "" order = row[idx["order"]] active = row[idx["active"]] stop_processing = row[idx["stop_processing"]] strict = row[idx["strict"]] r["group_title"] = group_title.strip() or "Default rules" r["title"] = title.strip() or f"Imported rule {rule_id}" r["description"] = description.strip() r["order"] = as_int(order) r["active"] = as_bool(active) r["stop_processing"] = as_bool(stop_processing) r["strict"] = as_bool(strict) elif row_type == "trigger": t_type = row[idx["trigger_type"]].strip() t_value = row[idx["trigger_value"]].strip() t_order = as_int(row[idx["trigger_order"]], default=len(r["triggers"]) + 1) t_active = as_bool(row[idx["trigger_active"]]) if row[idx["trigger_active"]] != "" else True t_stop = as_bool(row[idx["trigger_stop_processing"]]) if t_type == "user_action": # This is the rule's overall trigger context, not a # member of the "triggers" array. r["trigger"] = t_value or "store-journal" else: r["triggers"].append({ "type": t_type, "value": t_value, "order": t_order, "active": t_active, "stop_processing": t_stop, }) elif row_type == "action": a_type = row[idx["action_type"]].strip() a_value = row[idx["action_value"]].strip() a_order = as_int(row[idx["action_order"]], default=len(r["actions"]) + 1) a_active = as_bool(row[idx["action_active"]]) if row[idx["action_active"]] != "" else True a_stop = as_bool(row[idx["action_stop_processing"]]) r["actions"].append({ "type": a_type, "value": a_value, "order": a_order, "active": a_active, "stop_processing": a_stop, }) # sort triggers/actions by their order field for determinism for r in rules.values(): r["triggers"].sort(key=lambda t: t["order"]) r["actions"].sort(key=lambda a: a["order"]) return rules # -------------------------------------------------------------------------- # Firefly III API client # -------------------------------------------------------------------------- class FireflyClient: def __init__(self, base_url, token, timeout=30): base_url = (base_url or "").strip().strip('"').strip("'") token = (token or "").strip().strip('"').strip("'") # A newline, carriage return, or other control character in the # token (very common when it's loaded from a file or .env with a # trailing newline) produces exactly the kind of opaque # "invalid header value" 400 error this client guards against here. stripped_token = "".join(ch for ch in token if ch.isprintable()) if stripped_token != token: print("WARNING: your Firefly token contained non-printable characters " "(likely a trailing newline). They have been stripped automatically, " "but double-check how the token was saved/exported.", file=sys.stderr) token = stripped_token if not token: sys.exit("No Firefly API token provided (empty after cleanup). " "Check --token / FIREFLY_TOKEN.") self.base_url = base_url.rstrip("/") self.session = requests.Session() self.session.headers.update({ "Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json", }) self.timeout = timeout def _diagnose_and_raise(self, resp): """Raise a more informative error than requests' default one-liner.""" try: body = resp.text[:1000] except Exception: body = "" raise RuntimeError( f"{resp.status_code} {resp.reason} for {resp.request.method} {resp.url}\n" f"Response body: {body}" ) def _url(self, path): return f"{self.base_url}/api/v1{path}" def list_rule_groups(self): groups = [] url = self._url("/rule-groups") while url: resp = self.session.get(url, timeout=self.timeout) if not resp.ok: self._diagnose_and_raise(resp) payload = resp.json() groups.extend(payload.get("data", [])) url = payload.get("links", {}).get("next") return groups def create_rule_group(self, title, description="", active=True): body = {"title": title, "description": description, "active": active} resp = self.session.post(self._url("/rule-groups"), json=body, timeout=self.timeout) if not resp.ok: raise RuntimeError(f"Failed to create rule group '{title}': " f"{resp.status_code} {resp.text}") return resp.json()["data"] def list_rules(self): rules = [] url = self._url("/rules") while url: resp = self.session.get(url, timeout=self.timeout) if not resp.ok: self._diagnose_and_raise(resp) payload = resp.json() rules.extend(payload.get("data", [])) url = payload.get("links", {}).get("next") return rules def list_budgets(self): budgets = [] url = self._url("/budgets") while url: resp = self.session.get(url, timeout=self.timeout) if not resp.ok: self._diagnose_and_raise(resp) payload = resp.json() budgets.extend(payload.get("data", [])) url = payload.get("links", {}).get("next") return budgets def create_budget(self, name): resp = self.session.post(self._url("/budgets"), json={"name": name}, timeout=self.timeout) if not resp.ok: self._diagnose_and_raise(resp) return resp.json()["data"] def list_bills(self): bills = [] url = self._url("/bills") while url: resp = self.session.get(url, timeout=self.timeout) if not resp.ok: self._diagnose_and_raise(resp) payload = resp.json() bills.extend(payload.get("data", [])) url = payload.get("links", {}).get("next") return bills def list_accounts(self): accounts = [] url = self._url("/accounts") while url: resp = self.session.get(url, timeout=self.timeout) if not resp.ok: self._diagnose_and_raise(resp) payload = resp.json() accounts.extend(payload.get("data", [])) url = payload.get("links", {}).get("next") return accounts def create_rule(self, body): resp = self.session.post(self._url("/rules"), json=body, timeout=self.timeout) if not resp.ok: raise RuntimeError(f"{resp.status_code} {resp.text}") return resp.json()["data"] # -------------------------------------------------------------------------- # Main import logic # -------------------------------------------------------------------------- def collect_action_dependencies(rules): """Returns dict: action_type -> sorted set of referenced values, for the action types that reference other Firefly objects by name.""" tracked_types = {"set_budget", "link_to_bill", "set_destination_account", "set_source_account"} deps = defaultdict(set) for rule in rules.values(): for a in rule["actions"]: if a["type"] in tracked_types and a["value"]: deps[a["type"]].add(a["value"]) return deps def ensure_budgets_exist(client, budget_names, delay): """Auto-creates any missing budgets (safe: a budget only needs a name). Returns the set of budget names that exist after this call.""" existing = {b["attributes"]["name"].strip().lower() for b in client.list_budgets()} for name in sorted(budget_names): if name.strip().lower() in existing: continue try: client.create_budget(name) print(f" Created missing budget '{name}'") existing.add(name.strip().lower()) except Exception as e: print(f" [FAIL] could not create budget '{name}': {e}") time.sleep(delay) return existing def warn_about_missing_bills_and_accounts(client, deps): """Bills need real min/max amount + due date + repeat frequency, and accounts need a type (asset/expense/revenue/liability) - none of which is present in the rules CSV. Rather than fabricate that data, just tell the user exactly what to create by hand before re-running the import.""" missing_bills = set() if "link_to_bill" in deps: existing_bills = {b["attributes"]["name"].strip().lower() for b in client.list_bills()} missing_bills = {v for v in deps["link_to_bill"] if v.strip().lower() not in existing_bills} missing_accounts = set() account_values = deps.get("set_destination_account", set()) | deps.get("set_source_account", set()) if account_values: existing_accounts = {a["attributes"]["name"].strip().lower() for a in client.list_accounts()} missing_accounts = {v for v in account_values if v.strip().lower() not in existing_accounts} if missing_bills or missing_accounts: print("\nThe following need to be created MANUALLY before their rules will succeed") print("(bills need min/max amount + due date + repeat frequency, accounts need a") print("type — none of that is in the rules CSV, so this script won't guess):\n") if missing_bills: print(" Missing bills (create under Bills in the Firefly UI):") for b in sorted(missing_bills): print(f" - {b}") if missing_accounts: print(" Missing accounts (create under Accounts in the Firefly UI):") for a in sorted(missing_accounts): print(f" - {a}") print("\n After creating these, re-run this script with --apply --skip-existing") print(" to import only the rules that failed the first time.\n") return missing_bills, missing_accounts def build_rule_body(rule, rule_group_id): return { "title": rule["title"], "description": rule["description"] or None, "rule_group_id": str(rule_group_id), "trigger": rule["trigger"], "strict": rule["strict"], "stop_processing": rule["stop_processing"], "active": rule["active"], "order": rule["order"], "triggers": [ { "type": t["type"], "value": t["value"], "order": t["order"], "active": t["active"], "stop_processing": t["stop_processing"], } for t in rule["triggers"] ], "actions": [ { "type": a["type"], "value": a["value"], "order": a["order"], "active": a["active"], "stop_processing": a["stop_processing"], } for a in rule["actions"] ], } def main(): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--csv", required=True, help="Path to the exported rules CSV") parser.add_argument("--url", default=os.environ.get("FIREFLY_URL"), help="Base URL of the DESTINATION Firefly III instance " "(or set FIREFLY_URL env var)") parser.add_argument("--token", default=os.environ.get("FIREFLY_TOKEN"), help="Personal Access Token for the DESTINATION instance " "(or set FIREFLY_TOKEN env var)") parser.add_argument("--apply", action="store_true", help="Actually call the API and create rule groups/rules. " "Without this flag, the script only prints what it would do.") parser.add_argument("--skip-existing", action="store_true", help="Skip creating a rule if a rule with the same title " "already exists on the destination instance.") parser.add_argument("--no-shift-fix", dest="shift_fix", action="store_false", help="Disable the workaround for the title/description/order " "column-shift bug in Firefly's export-data command " "(use this only if your CSV doesn't exhibit that bug).") parser.add_argument("--delay", type=float, default=0.25, help="Seconds to sleep between API calls (default: 0.25)") args = parser.parse_args() if not os.path.exists(args.csv): sys.exit(f"CSV not found: {args.csv}") rules = parse_csv(args.csv, apply_shift_fix=args.shift_fix) print(f"Parsed {len(rules)} rules from {args.csv}\n") # Group rules by their rule group title, preserving order of first appearance groups = OrderedDict() for rule_id, rule in rules.items(): groups.setdefault(rule["group_title"], []).append((rule_id, rule)) print(f"Found {len(groups)} rule group(s): {', '.join(groups.keys())}\n") if not args.apply: print("[DRY RUN] No --apply flag given, so nothing will be sent to the API.") print("Preview of what would be created:\n") for group_title, group_rules in groups.items(): print(f"Rule group: {group_title}") for rule_id, rule in group_rules: print(f" - [{rule_id}] {rule['title']!r} " f"(trigger={rule['trigger']}, " f"{len(rule['triggers'])} triggers, " f"{len(rule['actions'])} actions, " f"active={rule['active']}, stop_processing={rule['stop_processing']}, " f"strict={rule['strict']})") print("\nRe-run with --apply (and --url/--token or FIREFLY_URL/FIREFLY_TOKEN) " "to actually create these.") return if not args.url or not args.token: sys.exit("For --apply you must provide --url and --token " "(or set FIREFLY_URL / FIREFLY_TOKEN env vars).") client = FireflyClient(args.url, args.token) print(f"Connecting to {client.base_url} ...") try: about_resp = client.session.get(client._url("/about"), timeout=client.timeout) if not about_resp.ok: client._diagnose_and_raise(about_resp) about = about_resp.json().get("data", {}) print(f" Connected OK. Firefly III version: {about.get('version', 'unknown')}") # A Personal Access Token authenticates as whoever generated it - # there is no way to override which user a request runs as. This # confirms exactly which account the token belongs to, so you can # verify it's the one that should own the imported rules. user_resp = client.session.get(client._url("/about/user"), timeout=client.timeout) if user_resp.ok: user_attrs = user_resp.json().get("data", {}).get("attributes", {}) user_id = user_resp.json().get("data", {}).get("id") print(f" Token authenticates as: id={user_id}, " f"email={user_attrs.get('email', 'unknown')}\n") else: print(" (Could not fetch /about/user to confirm token identity)\n") except Exception as e: sys.exit( f"Could not connect to the Firefly III API at {client.base_url}/api/v1/about\n" f"{e}\n\n" "Common causes:\n" " - Wrong --url (should be the base site URL, no trailing /api/...)\n" " - Token has a trailing newline/whitespace or quotes baked in\n" " - Token is expired or was generated on a different instance\n" " - A reverse proxy / tunnel (e.g. duckdns/cloudflare) in front of the " "instance is rejecting the request headers" ) # --- resolve / create rule groups ------------------------------------------------- print("Fetching existing rule groups from destination instance...") existing_groups = client.list_rule_groups() title_to_group_id = {g["attributes"]["title"].strip().lower(): g["id"] for g in existing_groups} group_title_to_id = {} for group_title in groups: key = group_title.strip().lower() if key in title_to_group_id: group_title_to_id[group_title] = title_to_group_id[key] print(f" Using existing rule group '{group_title}' (id={title_to_group_id[key]})") else: created = client.create_rule_group(group_title) group_title_to_id[group_title] = created["id"] print(f" Created rule group '{group_title}' (id={created['id']})") time.sleep(args.delay) # --- resolve existing rule titles (for --skip-existing) --------------------------- existing_rule_titles = set() if args.skip_existing: print("\nFetching existing rules from destination instance...") for r in client.list_rules(): existing_rule_titles.add(r["attributes"]["title"].strip().lower()) # --- resolve dependencies: budgets (auto-create), bills/accounts (warn only) ----- deps = collect_action_dependencies(rules) if "set_budget" in deps: print("\nChecking/creating budgets referenced by rule actions...") ensure_budgets_exist(client, deps["set_budget"], args.delay) warn_about_missing_bills_and_accounts(client, deps) # --- create rules ------------------------------------------------------------------- print("\nCreating rules...") created, skipped, failed = 0, 0, 0 failures = [] for group_title, group_rules in groups.items(): rule_group_id = group_title_to_id[group_title] for rule_id, rule in group_rules: if args.skip_existing and rule["title"].strip().lower() in existing_rule_titles: print(f" [skip] '{rule['title']}' already exists") skipped += 1 continue body = build_rule_body(rule, rule_group_id) try: result = client.create_rule(body) print(f" [ok] '{rule['title']}' -> new id {result['id']}") created += 1 except Exception as e: print(f" [FAIL] '{rule['title']}' (old id {rule_id}): {e}") failed += 1 failures.append((rule_id, rule["title"], str(e))) time.sleep(args.delay) print(f"\nDone. Created={created}, Skipped={skipped}, Failed={failed}") if failures: log_path = "firefly_import_failures.json" with open(log_path, "w") as f: json.dump( [{"rule_id": rid, "title": t, "error": err} for rid, t, err in failures], f, indent=2, ) print(f"Failure details written to {log_path}") if __name__ == "__main__": main()