|
#!/usr/bin/env python3
|
|
# /// script
|
|
# requires-python = ">=3.10"
|
|
# ///
|
|
#
|
|
# (PEP 723 inline metadata — `uv run csv_to_ynab_json.py` uses this Python bound;
|
|
# no third-party packages; stdlib only.)
|
|
"""
|
|
Convert YNAB CSV exports (Plan + Register) into JSON shaped like a full plan
|
|
export. Field meanings and enums follow the YNAB API (see OpenAPI spec).
|
|
|
|
Spec: https://api.ynab.com/papi/open_api_spec.yaml
|
|
|
|
Limitations (CSV does not contain):
|
|
- Server-issued UUIDs, goals, category notes, payee locations, splits,
|
|
scheduled transactions, real account balances, direct-import flags.
|
|
|
|
Dependencies: Python 3.10+ standard library only.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import re
|
|
import tempfile
|
|
import uuid
|
|
import zipfile
|
|
from collections import defaultdict
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
# Deterministic IDs across runs (same inputs -> same UUIDs)
|
|
_NS = uuid.UUID("6ba7b810-9dad-11d1-80b4-00c04fd430c8")
|
|
|
|
# YNAB export / API labels for the Ready to Assign inflow category (not user payees).
|
|
# See MonthSummaryBase.income in the OpenAPI spec.
|
|
_INFLOW_GROUP = "Inflow"
|
|
_RTA_CATEGORY = "Ready to Assign"
|
|
|
|
|
|
def _uid(kind: str, *parts: str) -> str:
|
|
return str(uuid.uuid5(_NS, f"{kind}|" + "|".join(parts)))
|
|
|
|
|
|
def parse_money(s: str) -> int:
|
|
"""CSV currency string -> YNAB milliunits (1000 = $1.00)."""
|
|
if not s or not str(s).strip():
|
|
return 0
|
|
s = str(s).strip().replace("$", "").replace(",", "")
|
|
neg = False
|
|
if s.startswith("-"):
|
|
neg = True
|
|
s = s[1:].strip()
|
|
if s.startswith("(") and s.endswith(")"):
|
|
neg = True
|
|
s = s[1:-1].strip()
|
|
try:
|
|
v = float(s)
|
|
except ValueError:
|
|
return 0
|
|
if neg:
|
|
v = -v
|
|
return int(round(v * 1000))
|
|
|
|
|
|
def parse_month_label(label: str) -> str | None:
|
|
"""'Mar 2024' -> '2024-03-01'."""
|
|
label = label.strip()
|
|
try:
|
|
dt = datetime.strptime(label, "%b %Y")
|
|
except ValueError:
|
|
try:
|
|
dt = datetime.strptime(label, "%B %Y")
|
|
except ValueError:
|
|
return None
|
|
return f"{dt.year:04d}-{dt.month:02d}-01"
|
|
|
|
|
|
def guess_account_type(name: str) -> str:
|
|
"""
|
|
Map account name to AccountType enum (OpenAPI components/schemas/AccountType).
|
|
Uses only generic English keywords, not institution or person names.
|
|
"""
|
|
n = name.lower()
|
|
|
|
if "mortgage" in n or "home loan" in n:
|
|
return "mortgage"
|
|
if "heloc" in n or "home equity" in n and "loan" not in n:
|
|
return "lineOfCredit"
|
|
if "line of credit" in n or re.search(r"\bloc\b", n):
|
|
return "lineOfCredit"
|
|
if "student loan" in n:
|
|
return "studentLoan"
|
|
if "auto loan" in n or "car loan" in n:
|
|
return "autoLoan"
|
|
if "medical" in n and "debt" in n:
|
|
return "medicalDebt"
|
|
if "credit card" in n:
|
|
return "creditCard"
|
|
if re.search(r"\b(visa|mastercard|amex|discover)\b", n):
|
|
return "creditCard"
|
|
if n.endswith(" card") or n.endswith(" card "):
|
|
return "creditCard"
|
|
if "store card" in n:
|
|
return "creditCard"
|
|
if "savings" in n:
|
|
return "savings"
|
|
if re.search(r"\bcash\b", n) or "wallet" in n:
|
|
return "cash"
|
|
if any(x in n for x in ("ira", "401k", "401(k)", "brokerage", "investment")):
|
|
return "otherAsset"
|
|
if any(x in n for x in ("equity", "retirement")) and "line" not in n:
|
|
return "otherAsset"
|
|
if any(x in n for x in ("crypto", "bitcoin", "precious metal")):
|
|
return "otherAsset"
|
|
if "loan" in n:
|
|
return "otherDebt"
|
|
if "liability" in n:
|
|
return "otherLiability"
|
|
return "checking"
|
|
|
|
|
|
def guess_on_budget(acct_type: str) -> bool:
|
|
"""Tracking / debt accounts are off-budget; everyday accounts on-budget."""
|
|
off_budget = frozenset(
|
|
{
|
|
"mortgage",
|
|
"autoLoan",
|
|
"studentLoan",
|
|
"personalLoan",
|
|
"medicalDebt",
|
|
"otherDebt",
|
|
"otherAsset",
|
|
"otherLiability",
|
|
}
|
|
)
|
|
return acct_type not in off_budget
|
|
|
|
|
|
def cleared_status(raw: str) -> str:
|
|
"""TransactionClearedStatus: cleared | uncleared | reconciled."""
|
|
m = raw.strip().lower()
|
|
if m == "reconciled":
|
|
return "reconciled"
|
|
if m == "uncleared":
|
|
return "uncleared"
|
|
return "cleared"
|
|
|
|
|
|
def transfer_target_name(payee: str) -> str | None:
|
|
"""
|
|
If payee is a transfer payee (\"Transfer : …\"), return the payee name with
|
|
the \"Transfer : \" prefix removed; otherwise None.
|
|
|
|
The remainder is typically the destination account display name as exported
|
|
by YNAB (e.g. matching another account's name).
|
|
"""
|
|
p = payee.strip()
|
|
prefix = "Transfer : "
|
|
if not p.startswith(prefix):
|
|
return None
|
|
rest = p[len(prefix) :].strip()
|
|
return rest if rest else None
|
|
|
|
|
|
def empty_category_template(
|
|
cid: str,
|
|
gid: str,
|
|
name: str,
|
|
hidden: bool,
|
|
budgeted: int,
|
|
activity: int,
|
|
balance: int,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"id": cid,
|
|
"category_group_id": gid,
|
|
"name": name,
|
|
"hidden": hidden,
|
|
"original_category_group_id": None,
|
|
"note": None,
|
|
"budgeted": budgeted,
|
|
"activity": activity,
|
|
"balance": balance,
|
|
"goal_type": None,
|
|
"goal_needs_whole_amount": None,
|
|
"goal_day": None,
|
|
"goal_cadence": None,
|
|
"goal_cadence_frequency": None,
|
|
"goal_creation_month": None,
|
|
"goal_target": 0,
|
|
"goal_target_month": None,
|
|
"goal_percentage_complete": None,
|
|
"goal_months_to_budget": None,
|
|
"goal_under_funded": None,
|
|
"goal_overall_funded": None,
|
|
"goal_overall_left": None,
|
|
"goal_snoozed_at": None,
|
|
"deleted": False,
|
|
}
|
|
|
|
|
|
def load_plan(path: str | Path) -> tuple[dict[tuple[str, str], str], dict[str, dict[tuple[str, str], dict[str, int]]]]:
|
|
by_month: dict[str, dict[tuple[str, str], dict[str, int]]] = defaultdict(
|
|
lambda: defaultdict(lambda: {"budgeted": 0, "activity": 0, "balance": 0})
|
|
)
|
|
month_for_key: dict[tuple[str, str], str] = {}
|
|
|
|
with open(path, newline="", encoding="utf-8-sig") as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
g = (row.get("Category Group") or "").strip()
|
|
c = (row.get("Category") or "").strip()
|
|
if not g and not c:
|
|
continue
|
|
mlabel = (row.get("Month") or "").strip()
|
|
month_iso = parse_month_label(mlabel)
|
|
if not month_iso:
|
|
continue
|
|
key = (g, c)
|
|
month_for_key[key] = month_iso
|
|
by_month[month_iso][key]["budgeted"] = parse_money(row.get("Assigned", ""))
|
|
by_month[month_iso][key]["activity"] = parse_money(row.get("Activity", ""))
|
|
by_month[month_iso][key]["balance"] = parse_money(row.get("Available", ""))
|
|
|
|
return month_for_key, dict(by_month)
|
|
|
|
|
|
def collect_category_keys_from_register(path: str | Path) -> set[tuple[str, str]]:
|
|
keys: set[tuple[str, str]] = set()
|
|
with open(path, newline="", encoding="utf-8-sig") as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
g = (row.get("Category Group") or "").strip()
|
|
c = (row.get("Category") or "").strip()
|
|
if g or c:
|
|
keys.add((g, c))
|
|
return keys
|
|
|
|
|
|
def _is_rta_row(row: dict[str, str]) -> bool:
|
|
return (
|
|
(row.get("Category Group") or "").strip() == _INFLOW_GROUP
|
|
and (row.get("Category") or "").strip() == _RTA_CATEGORY
|
|
)
|
|
|
|
|
|
def _register_row_amount(row: dict[str, str]) -> int:
|
|
return parse_money(row.get("Inflow", "")) - parse_money(row.get("Outflow", ""))
|
|
|
|
|
|
def _scan_register_metrics(path: str | Path) -> tuple[dict[str, int], dict[str, int]]:
|
|
"""
|
|
Per YNAB API MonthSummaryBase:
|
|
- income: total transaction amounts in 'Inflow: Ready to Assign' for the month
|
|
- activity_month: total transaction amounts excluding RTA inflow category
|
|
"""
|
|
income_by_ym: dict[str, int] = defaultdict(int)
|
|
activity_ex_rta_by_ym: dict[str, int] = defaultdict(int)
|
|
|
|
with open(path, newline="", encoding="utf-8-sig") as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
ds = (row.get("Date") or "").strip()
|
|
if len(ds) < 7:
|
|
continue
|
|
ym = ds[:7]
|
|
amt = _register_row_amount(row)
|
|
if _is_rta_row(row):
|
|
income_by_ym[ym] += amt
|
|
else:
|
|
activity_ex_rta_by_ym[ym] += amt
|
|
|
|
return dict(income_by_ym), dict(activity_ex_rta_by_ym)
|
|
|
|
|
|
def find_plan_and_register_csvs_in_zip(zip_path: Path) -> tuple[Path, Path]:
|
|
"""Locate Plan.csv and Register.csv inside an export zip (any folder)."""
|
|
with zipfile.ZipFile(zip_path) as zf:
|
|
names = [n for n in zf.namelist() if not n.endswith("/")]
|
|
|
|
def is_plan(n: str) -> bool:
|
|
return bool(re.search(r"Plan\.csv$", n, re.I))
|
|
|
|
def is_register(n: str) -> bool:
|
|
return bool(re.search(r"Register\.csv$", n, re.I))
|
|
|
|
plans = [n for n in names if is_plan(n)]
|
|
regs = [n for n in names if is_register(n)]
|
|
|
|
if not plans:
|
|
raise SystemExit(f"No *Plan.csv file found in {zip_path}")
|
|
if not regs:
|
|
raise SystemExit(f"No *Register.csv file found in {zip_path}")
|
|
|
|
if len(plans) == 1 and len(regs) == 1:
|
|
return Path(plans[0]), Path(regs[0])
|
|
|
|
def stem_base(path_str: str, kind: str) -> str:
|
|
base = Path(path_str).name
|
|
suf = f" - {kind}.csv"
|
|
if base.lower().endswith(suf.lower()):
|
|
return base[: -len(suf)].strip()
|
|
base = re.sub(r"(?i)\s*-\s*" + re.escape(kind) + r"\.csv\s*$", "", base)
|
|
return base.strip()
|
|
|
|
pb = {stem_base(p, "Plan"): p for p in plans}
|
|
rb = {stem_base(r, "Register"): r for r in regs}
|
|
common = set(pb.keys()) & set(rb.keys())
|
|
if len(common) == 1:
|
|
k = next(iter(common))
|
|
return Path(pb[k]), Path(rb[k])
|
|
|
|
raise SystemExit(
|
|
f"Could not pair Plan/Register CSVs in {zip_path}. Found plans={plans!r} registers={regs!r}"
|
|
)
|
|
|
|
|
|
def extract_zip(zip_path: Path) -> tuple[Path, Path, tempfile.TemporaryDirectory[str]]:
|
|
"""Extract zip to a temp directory; return (plan_path, register_path, tempdir)."""
|
|
tmp = tempfile.TemporaryDirectory(prefix="ynab_csv_export_")
|
|
root = Path(tmp.name)
|
|
plan_member, reg_member = find_plan_and_register_csvs_in_zip(zip_path)
|
|
with zipfile.ZipFile(zip_path) as zf:
|
|
zf.extractall(root)
|
|
plan_p = root / plan_member
|
|
reg_p = root / reg_member
|
|
if not plan_p.is_file():
|
|
raise SystemExit(f"Extracted Plan path missing: {plan_p}")
|
|
if not reg_p.is_file():
|
|
raise SystemExit(f"Extracted Register path missing: {reg_p}")
|
|
return plan_p, reg_p, tmp
|
|
|
|
|
|
def build_budget(
|
|
plan_path: str | Path,
|
|
register_path: str | Path,
|
|
budget_name: str,
|
|
) -> dict[str, Any]:
|
|
plan_path = Path(plan_path)
|
|
register_path = Path(register_path)
|
|
|
|
month_for_key, plan_by_month = load_plan(plan_path)
|
|
reg_cats = collect_category_keys_from_register(register_path)
|
|
|
|
all_keys: set[tuple[str, str]] = set(month_for_key.keys()) | reg_cats
|
|
all_keys = {(g, c) for g, c in all_keys if g and c}
|
|
|
|
group_names = sorted({g for g, _ in all_keys})
|
|
group_meta: dict[str, dict[str, Any]] = {}
|
|
for g in group_names:
|
|
gid = _uid("category_group", g)
|
|
group_meta[g] = {"id": gid, "name": g, "hidden": False, "deleted": False}
|
|
|
|
cat_ids: dict[tuple[str, str], str] = {}
|
|
for g, c in sorted(all_keys):
|
|
cat_ids[(g, c)] = _uid("category", g, c)
|
|
|
|
account_names: list[str] = []
|
|
seen_acc: set[str] = set()
|
|
with open(register_path, newline="", encoding="utf-8-sig") as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
name = (row.get("Account") or "").strip()
|
|
if name and name not in seen_acc:
|
|
seen_acc.add(name)
|
|
account_names.append(name)
|
|
|
|
accounts: list[dict[str, Any]] = []
|
|
account_by_name: dict[str, dict[str, Any]] = {}
|
|
for name in account_names:
|
|
aid = _uid("account", name)
|
|
tid = _uid("transfer_payee", name)
|
|
atype = guess_account_type(name)
|
|
ob = guess_on_budget(atype)
|
|
acct = {
|
|
"id": aid,
|
|
"name": name,
|
|
"type": atype,
|
|
"on_budget": ob,
|
|
"closed": False,
|
|
"note": None,
|
|
"balance": 0,
|
|
"cleared_balance": 0,
|
|
"uncleared_balance": 0,
|
|
"transfer_payee_id": tid,
|
|
"direct_import_linked": False,
|
|
"direct_import_in_error": False,
|
|
"last_reconciled_at": None,
|
|
"debt_original_balance": None,
|
|
"debt_interest_rates": {},
|
|
"debt_minimum_payments": {},
|
|
"debt_escrow_amounts": {},
|
|
"deleted": False,
|
|
}
|
|
accounts.append(acct)
|
|
account_by_name[name] = acct
|
|
|
|
payees: list[dict[str, Any]] = []
|
|
payee_by_name: dict[str, str] = {}
|
|
|
|
for name in account_names:
|
|
acct = account_by_name[name]
|
|
pid = acct["transfer_payee_id"]
|
|
payees.append(
|
|
{
|
|
"id": pid,
|
|
"name": f"Transfer : {name}",
|
|
"transfer_account_id": acct["id"],
|
|
"deleted": False,
|
|
}
|
|
)
|
|
payee_by_name[f"Transfer : {name}"] = pid
|
|
|
|
with open(register_path, newline="", encoding="utf-8-sig") as f:
|
|
reader = csv.DictReader(f)
|
|
payee_names: set[str] = set()
|
|
for row in reader:
|
|
p = (row.get("Payee") or "").strip()
|
|
if not p:
|
|
continue
|
|
if transfer_target_name(p) is not None:
|
|
continue
|
|
payee_names.add(p)
|
|
|
|
for pname in sorted(payee_names):
|
|
pid = _uid("payee", pname)
|
|
payees.append({"id": pid, "name": pname, "transfer_account_id": None, "deleted": False})
|
|
payee_by_name[pname] = pid
|
|
|
|
all_months = sorted(plan_by_month.keys())
|
|
if not all_months:
|
|
first_m = last_m = datetime.now(UTC).strftime("%Y-%m-01")
|
|
else:
|
|
first_m = all_months[0]
|
|
last_m = all_months[-1]
|
|
|
|
budget_id = _uid("budget", budget_name)
|
|
|
|
last_plan = plan_by_month.get(last_m, {})
|
|
root_categories: list[dict[str, Any]] = []
|
|
for (g, c), cid in sorted(cat_ids.items(), key=lambda x: (x[0][0], x[0][1])):
|
|
gid = group_meta[g]["id"]
|
|
cat_hidden = g == "Hidden Categories"
|
|
nums = last_plan.get((g, c), {"budgeted": 0, "activity": 0, "balance": 0})
|
|
root_categories.append(
|
|
empty_category_template(
|
|
cid, gid, c, cat_hidden, nums["budgeted"], nums["activity"], nums["balance"]
|
|
)
|
|
)
|
|
|
|
income_by_month, activity_register_by_month = _scan_register_metrics(register_path)
|
|
|
|
months_out: list[dict[str, Any]] = []
|
|
for month_iso in sorted(plan_by_month.keys(), reverse=True):
|
|
pdata = plan_by_month[month_iso]
|
|
cats_layer: list[dict[str, Any]] = []
|
|
sum_budgeted = 0
|
|
for (g, c), cid in sorted(cat_ids.items(), key=lambda x: (x[0][0], x[0][1])):
|
|
gid = group_meta[g]["id"]
|
|
cat_hidden = g == "Hidden Categories"
|
|
nums = pdata.get((g, c), {"budgeted": 0, "activity": 0, "balance": 0})
|
|
sum_budgeted += nums["budgeted"]
|
|
cats_layer.append(
|
|
empty_category_template(
|
|
cid, gid, c, cat_hidden, nums["budgeted"], nums["activity"], nums["balance"]
|
|
)
|
|
)
|
|
|
|
ym = month_iso[:7]
|
|
income_mu = income_by_month.get(ym, 0)
|
|
# API: month.activity = transaction totals excluding RTA inflow (not Plan rollups).
|
|
activity_month = activity_register_by_month.get(ym, 0)
|
|
|
|
months_out.append(
|
|
{
|
|
"month": month_iso,
|
|
"note": "",
|
|
"income": income_mu,
|
|
"budgeted": sum_budgeted,
|
|
"activity": activity_month,
|
|
"to_be_budgeted": 0,
|
|
"age_of_money": None,
|
|
"deleted": False,
|
|
"categories": cats_layer,
|
|
}
|
|
)
|
|
|
|
raw_rows: list[dict[str, Any]] = []
|
|
with open(register_path, newline="", encoding="utf-8-sig") as f:
|
|
reader = csv.DictReader(f)
|
|
for i, row in enumerate(reader):
|
|
raw_rows.append({"row_index": i, **row})
|
|
|
|
pending: dict[tuple[Any, ...], list[dict[str, Any]]] = defaultdict(list)
|
|
for row in raw_rows:
|
|
acc = (row.get("Account") or "").strip()
|
|
payee = (row.get("Payee") or "").strip()
|
|
tgt = transfer_target_name(payee)
|
|
if tgt is None:
|
|
continue
|
|
amt = parse_money(row.get("Outflow", "")) - parse_money(row.get("Inflow", ""))
|
|
if amt == 0:
|
|
continue
|
|
d = (row.get("Date") or "").strip()
|
|
key = (d, abs(amt), tuple(sorted([acc, tgt])))
|
|
pending[key].append(row)
|
|
|
|
pair_id: dict[int, tuple[str, str]] = {}
|
|
for key, rows in pending.items():
|
|
if len(rows) != 2:
|
|
continue
|
|
r1, r2 = rows
|
|
a1 = (r1.get("Account") or "").strip()
|
|
a2 = (r2.get("Account") or "").strip()
|
|
t1 = transfer_target_name((r1.get("Payee") or "").strip())
|
|
t2 = transfer_target_name((r2.get("Payee") or "").strip())
|
|
if t1 == a2 and t2 == a1:
|
|
base = _uid("xfer", key[0], str(key[1]), key[2][0], key[2][1])
|
|
d = key[0]
|
|
id_a = f"{base}_{d}"
|
|
id_b = f"{base}_t_{d}"
|
|
amt1 = parse_money(r1.get("Outflow", "")) - parse_money(r1.get("Inflow", ""))
|
|
if amt1 < 0:
|
|
pair_id[r1["row_index"]] = (id_a, id_b)
|
|
pair_id[r2["row_index"]] = (id_b, id_a)
|
|
else:
|
|
pair_id[r1["row_index"]] = (id_b, id_a)
|
|
pair_id[r2["row_index"]] = (id_a, id_b)
|
|
|
|
transactions: list[dict[str, Any]] = []
|
|
for row in raw_rows:
|
|
acc_name = (row.get("Account") or "").strip()
|
|
acct = account_by_name.get(acc_name)
|
|
if not acct:
|
|
continue
|
|
outf = parse_money(row.get("Outflow", ""))
|
|
inf = parse_money(row.get("Inflow", ""))
|
|
amount = inf - outf
|
|
payee = (row.get("Payee") or "").strip()
|
|
memo = (row.get("Memo") or "").strip() or None
|
|
date_s = (row.get("Date") or "").strip()
|
|
|
|
cat_id: str | None = None
|
|
g = (row.get("Category Group") or "").strip()
|
|
c = (row.get("Category") or "").strip()
|
|
if g and c and (g, c) in cat_ids:
|
|
cat_id = cat_ids[(g, c)]
|
|
|
|
xfer_account: str | None = None
|
|
payee_id: str | None = None
|
|
tt = transfer_target_name(payee)
|
|
if tt is not None:
|
|
target = account_by_name.get(tt)
|
|
if target:
|
|
payee_id = target["transfer_payee_id"]
|
|
xfer_account = target["id"]
|
|
else:
|
|
pname = payee
|
|
payee_id = payee_by_name.get(pname) or _uid("payee", pname)
|
|
if pname not in payee_by_name:
|
|
payees.append(
|
|
{"id": payee_id, "name": pname, "transfer_account_id": None, "deleted": False}
|
|
)
|
|
payee_by_name[pname] = payee_id
|
|
else:
|
|
payee_id = payee_by_name.get(payee)
|
|
if not payee_id:
|
|
payee_id = _uid("payee", payee)
|
|
payees.append({"id": payee_id, "name": payee, "transfer_account_id": None, "deleted": False})
|
|
payee_by_name[payee] = payee_id
|
|
|
|
if tt is not None:
|
|
cat_id = None
|
|
|
|
transfer_transaction_id: str | None = None
|
|
tx_id: str
|
|
if row["row_index"] in pair_id:
|
|
my_id, other_id = pair_id[row["row_index"]]
|
|
tx_id = my_id
|
|
transfer_transaction_id = other_id
|
|
else:
|
|
tx_id = f"{_uid('tx', acc_name, date_s, str(amount), str(row['row_index']))}_{date_s}"
|
|
|
|
flag = (row.get("Flag") or "").strip().lower() or None
|
|
if flag not in {"red", "orange", "yellow", "green", "blue", "purple"}:
|
|
flag = None
|
|
|
|
transactions.append(
|
|
{
|
|
"id": tx_id,
|
|
"date": date_s,
|
|
"amount": amount,
|
|
"memo": memo,
|
|
"cleared": cleared_status(row.get("Cleared", "")),
|
|
"approved": True,
|
|
"flag_color": flag,
|
|
"account_id": acct["id"],
|
|
"payee_id": payee_id,
|
|
"category_id": cat_id,
|
|
"transfer_account_id": xfer_account,
|
|
"transfer_transaction_id": transfer_transaction_id,
|
|
"matched_transaction_id": None,
|
|
"import_id": None,
|
|
"import_payee_name": None,
|
|
"import_payee_name_original": None,
|
|
"debt_transaction_type": None,
|
|
"deleted": False,
|
|
}
|
|
)
|
|
|
|
transactions.sort(key=lambda t: (t["date"], t["id"]), reverse=True)
|
|
|
|
now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
budget: dict[str, Any] = {
|
|
"id": budget_id,
|
|
"name": budget_name,
|
|
"last_modified_on": now,
|
|
"date_format": {"format": "YYYY-MM-DD"},
|
|
"currency_format": {
|
|
"iso_code": "USD",
|
|
"example_format": "123,456.78",
|
|
"decimal_digits": 2,
|
|
"decimal_separator": ".",
|
|
"symbol_first": True,
|
|
"group_separator": ",",
|
|
"currency_symbol": "$",
|
|
"display_symbol": True,
|
|
},
|
|
"first_month": first_m,
|
|
"last_month": last_m,
|
|
"accounts": accounts,
|
|
"payees": payees,
|
|
"payee_locations": [],
|
|
"category_groups": [group_meta[g] for g in group_names],
|
|
"categories": root_categories,
|
|
"months": months_out,
|
|
"transactions": transactions,
|
|
"subtransactions": [],
|
|
"scheduled_transactions": [],
|
|
"scheduled_subtransactions": [],
|
|
}
|
|
|
|
return {"data": {"budget": budget, "server_knowledge": 0}}
|
|
|
|
|
|
def _budget_name_from_plan_path(plan_path: Path) -> str:
|
|
base = plan_path.name
|
|
name = re.sub(r"\s+as of\s+.*$", "", base, flags=re.I)
|
|
name = re.sub(r"\s*-\s*Plan\.csv$", "", name, flags=re.I).strip()
|
|
return name or "Budget"
|
|
|
|
|
|
def default_output_path(*, plan_path: Path | None, zip_path: Path | None) -> Path:
|
|
"""
|
|
Default JSON path beside the input: same directory and basename pattern as
|
|
the zip, or the Plan/Register CSV with ' - Plan.csv' / ' - Register.csv' removed.
|
|
"""
|
|
if zip_path is not None:
|
|
return zip_path.with_suffix(".json")
|
|
if plan_path is None:
|
|
raise ValueError("default_output_path requires zip_path or plan_path")
|
|
stem = plan_path.name
|
|
stem = re.sub(r"(?i)\s*-\s*Plan\.csv\s*$", "", stem)
|
|
stem = re.sub(r"(?i)\s*-\s*Register\.csv\s*$", "", stem).strip()
|
|
if not stem:
|
|
stem = plan_path.stem
|
|
return plan_path.with_name(f"{stem}.json")
|
|
|
|
|
|
def _budget_name_from_zip_path(zip_path: Path) -> str:
|
|
base = zip_path.stem
|
|
base = re.sub(r"^\s*YNAB\s+Export\s*-\s*", "", base, flags=re.I).strip()
|
|
base = re.sub(r"\s+as of\s+.*$", "", base, flags=re.I).strip()
|
|
return base or "Budget"
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser(
|
|
description="Convert YNAB Plan.csv + Register.csv to JSON aligned with the YNAB API plan export shape."
|
|
)
|
|
ap.add_argument(
|
|
"-z",
|
|
"--zip",
|
|
type=Path,
|
|
help="Zip export containing *Plan.csv and *Register.csv (e.g. YNAB Export - ….zip)",
|
|
)
|
|
ap.add_argument(
|
|
"-p",
|
|
"--plan",
|
|
type=Path,
|
|
help="Path to Plan.csv (requires --register; mutually exclusive with --zip)",
|
|
)
|
|
ap.add_argument(
|
|
"-r",
|
|
"--register",
|
|
type=Path,
|
|
help="Path to Register.csv (requires --plan; mutually exclusive with --zip)",
|
|
)
|
|
ap.add_argument(
|
|
"-o",
|
|
"--output",
|
|
type=Path,
|
|
default=None,
|
|
help="Output .json path (default: derived from Plan filename or zip name)",
|
|
)
|
|
ap.add_argument(
|
|
"-n",
|
|
"--budget-name",
|
|
default=None,
|
|
help="Budget name in JSON (default: derived from Plan filename or zip name)",
|
|
)
|
|
args = ap.parse_args()
|
|
|
|
use_zip = args.zip is not None
|
|
use_csv = args.plan is not None or args.register is not None
|
|
if use_zip and use_csv:
|
|
ap.error("Use either --zip or --plan/--register together, not both.")
|
|
if not use_zip and not (args.plan is not None and args.register is not None):
|
|
ap.error("Provide --zip, or both --plan and --register.")
|
|
|
|
out_path = args.output
|
|
if out_path is None:
|
|
out_path = default_output_path(
|
|
plan_path=args.plan if not use_zip else None,
|
|
zip_path=args.zip if use_zip else None,
|
|
)
|
|
|
|
if use_zip:
|
|
if not args.zip.is_file():
|
|
ap.error(f"Not a file: {args.zip}")
|
|
plan_path, reg_path, tmp = extract_zip(args.zip)
|
|
try:
|
|
name = args.budget_name or _budget_name_from_zip_path(args.zip)
|
|
out = build_budget(plan_path, reg_path, name)
|
|
finally:
|
|
tmp.cleanup()
|
|
else:
|
|
if not args.plan.is_file():
|
|
ap.error(f"Not a file: {args.plan}")
|
|
if not args.register.is_file():
|
|
ap.error(f"Not a file: {args.register}")
|
|
name = args.budget_name or _budget_name_from_plan_path(args.plan)
|
|
out = build_budget(args.plan, args.register, name)
|
|
|
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(out_path, "w", encoding="utf-8") as f:
|
|
json.dump(out, f, indent=2, ensure_ascii=False)
|
|
f.write("\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
