nabeel/Firefly-III-scripts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Upload files to "/"

Browse Source 
Script to import rules exported from a firefly instance using the CLI --export-rules command
This commit is contained in:
nabeel
2026-08-18 00:00:18 +00:00
commit dca4e6dc14
1 changed files with 575 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

575
import_firefly_rules.py Normal file
View File

@@ -0,0 +1,575 @@
#!/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, <group title>, <real title>, <real description>, <real order>, <real active>, <real stop_processing>, <real strict>, (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 = "<no 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()