J2897 icon

GnuCash Accounts CSV Book Builder

J2897 | PRO | 02/11/26 12:44:23 PM UTC | 0 ⭐ | 264 👁️ | Never ⏰ | []
Python |

33.4 KB

|

None

|

0 👍

/

0 👎

#!/usr/bin/env python3
"""
GnuCash Accounts CSV → Book Builder
 
Build a new GnuCash XML book (*.gnucash) from a “chart of accounts” (account tree) CSV.
 
Outputs
- A new book containing your full account hierarchy (a ROOT account is created automatically).
- All commodities/securities referenced by the CSV (currencies, crypto-assets, precious metals, etc.).
- Optional account attributes stored as slots: Placeholder, Hidden, Tax related, Notes, and Colour.
 
Deterministic by default
- Account GUIDs are stable: UUIDv5 derived from each account’s full path, so rebuilding produces the
  same IDs and git-friendly diffs.
- Use --random-guids to generate new GUIDs on each run.
 
Usage
  build  <accounts.csv> --out <book.gnucash> [options]
  verify <book.gnucash> <accounts.csv> [options]
 
CSV format (header must match exactly)
  Type, Full Account Name, Account Name, Account Code, Description, Account Colour, Notes,
  Symbol, Namespace, Hidden, Tax Info, Placeholder
 
Commodity rules enforced
- ISO 4217 currencies (GBP, USD, EUR, …) must use Namespace=CURRENCY.
- ISO precious metal codes are treated as securities (not currencies) to avoid currency-style behaviour:
  XAG/XAU/XPT/XPD → METALS:SILVER/GOLD/PLATINUM/PALLADIUM
- Defaults are provided for common crypto-assets and metals (names and display fractions). Override any
  non-currency fraction with:
    --commodity-fraction NAMESPACE:SYMBOL=FRACTION
 
GnuCash compatibility notes
- The internal template commodity (template:template) is created by default; disable with
  --no-template-commodity.
- Hidden and Tax Info are stored via account slots and round-tripped so strict verify can compare results.
 
Getting started
- Begin with a small CSV: your top-level accounts plus at least one CURRENCY row (e.g. GBP/CURRENCY),
  then run:
    python gnucash_accounts_csv_book_builder.py build accounts.csv --out new_book.gnucash
"""
 
from __future__ import annotations
 
import argparse
import sys
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Any
 
import pandas as pd
import pycountry
import xml.etree.ElementTree as ET
 
 
EXPECTED_COLUMNS: list[str] = [
    "Type",
    "Full Account Name",
    "Account Name",
    "Account Code",
    "Description",
    "Account Colour",
    "Notes",
    "Symbol",
    "Namespace",
    "Hidden",
    "Tax Info",
    "Placeholder",
]
 
 
GN_ACCOUNT_TYPES: set[str] = {
    "ASSET", "BANK", "CASH", "STOCK", "MUTUAL",
    "LIABILITY", "CREDIT", "EQUITY",
    "INCOME", "EXPENSE",
    "RECEIVABLE", "PAYABLE",
    "TRADING",
}
 
 
# Defaults for common securities (crypto + precious metals) precisions
DEFAULT_SECURITY_FRACTIONS: dict[str, int] = {
    # Crypto
    "BTC": 100_000_000,
    "ETH": 100_000_000,
    "BNB": 100_000_000,
    "ADA": 1_000_000,
    "USDT": 1_000_000,
    "USDC": 1_000_000,
 
    # Precious metals (treat as normal securities, units = ounces)
    "SILVER": 1_000,
    "GOLD": 1_000,
    "PLATINUM": 1_000,
    "PALLADIUM": 1_000,
}
 
DEFAULT_SECURITY_NAMES: dict[str, str] = {
    "BTC": "Bitcoin",
    "ETH": "Ethereum",
    "BNB": "BNB",
    "ADA": "Cardano",
    "USDT": "Tether",
    "USDC": "USD Coin",
 
    "SILVER": "Silver",
    "GOLD": "Gold",
    "PLATINUM": "Platinum",
    "PALLADIUM": "Palladium",
}
 
 
# Precious metals: avoid treating ISO metal codes (XAG/XAU/XPT/XPD) as CURRENCY,
# because GnuCash handles CURRENCY commodities as currencies/exchange rates, which
# makes bullion "buy 1 oz for £44" style entries awkward. We normalise these to
# METALS:<NAME> securities instead.
PRECIOUS_METALS_NAMESPACE = "METALS"
PRECIOUS_METAL_ISO_TO_SYMBOL: dict[str, str] = {
    "XAG": "SILVER",
    "XAU": "GOLD",
    "XPT": "PLATINUM",
    "XPD": "PALLADIUM",
}
 
 
 
def die(msg: str, *, code: int = 2) -> "NoReturn":
    print(f"ERROR: {msg}", file=sys.stderr)
    raise SystemExit(code)
 
 
def load_iso_currency_codes() -> set[str]:
    codes: set[str] = set()
    for c in pycountry.currencies:
        a3 = getattr(c, "alpha_3", None)
        if a3:
            codes.add(str(a3).upper())
    return codes
 
 
ISO_CURRENCY_CODES = load_iso_currency_codes()
 
 
def _norm_bool_tf(val: Any) -> str:
    s = str(val).strip().upper()
    if s in {"T", "TRUE", "1", "YES", "Y"}:
        return "T"
    if s in {"F", "FALSE", "0", "NO", "N", ""}:
        return "F"
    die(f"Invalid boolean flag value: {val!r} (expected T/F)")
 
 
def normalize_account_type(t: str) -> str:
    t = str(t).strip().upper()
    return {
        "CREDITCARD": "CREDIT",
        "CREDIT_CARD": "CREDIT",
        "A/RECEIVABLE": "RECEIVABLE",
        "A/RECEIVABLES": "RECEIVABLE",
        "RECEIVABLES": "RECEIVABLE",
        "A/PAYABLE": "PAYABLE",
        "A/PAYABLES": "PAYABLE",
        "PAYABLES": "PAYABLE",
    }.get(t, t)
 
 
@dataclass(frozen=True)
class CommodityKey:
    namespace: str
    symbol: str
 
    def __post_init__(self) -> None:
        ns = str(self.namespace).strip()
        sym = str(self.symbol).strip()
 
        # v3: preserve template commodity EXACTLY as lowercase
        if ns.lower() == "template" and sym.lower() == "template":
            ns_out, sym_out = "template", "template"
        else:
            ns_out, sym_out = ns.upper(), sym.upper()
 
        object.__setattr__(self, "namespace", ns_out)
        object.__setattr__(self, "symbol", sym_out)
 
 
def read_accounts_csv(path: Path) -> pd.DataFrame:
    df = pd.read_csv(path, dtype=str, keep_default_na=False).fillna("")
    missing = [c for c in EXPECTED_COLUMNS if c not in df.columns]
    if missing:
        die(f"CSV is missing required columns: {missing}")
 
    df = df[EXPECTED_COLUMNS].copy()
    df["Type"] = df["Type"].map(normalize_account_type)
 
    for col in [
        "Full Account Name", "Account Name", "Account Code",
        "Description", "Account Colour", "Notes", "Symbol", "Namespace",
    ]:
        df[col] = df[col].astype(str).str.strip()
 
    df["Hidden"] = df["Hidden"].map(_norm_bool_tf)
    df["Tax Info"] = df["Tax Info"].map(_norm_bool_tf)
    df["Placeholder"] = df["Placeholder"].map(_norm_bool_tf)
 
    # Normalise user-specified commodities (template commodity is internal, not from CSV)
    df["Namespace"] = df["Namespace"].str.upper()
    df["Symbol"] = df["Symbol"].str.upper()
 
    # Precious metals: normalise XAG/XAU/XPT/XPD (and any accidental CURRENCY:SILVER etc)
    # to METALS:<NAME> securities.
    pm_iso = df["Symbol"].isin(PRECIOUS_METAL_ISO_TO_SYMBOL.keys())
    if bool(pm_iso.any()):
        df.loc[pm_iso, "Symbol"] = df.loc[pm_iso, "Symbol"].map(PRECIOUS_METAL_ISO_TO_SYMBOL)
        df.loc[pm_iso, "Namespace"] = PRECIOUS_METALS_NAMESPACE
 
    pm_named = df["Namespace"].eq("CURRENCY") & df["Symbol"].isin(PRECIOUS_METAL_ISO_TO_SYMBOL.values())
    if bool(pm_named.any()):
        df.loc[pm_named, "Namespace"] = PRECIOUS_METALS_NAMESPACE
 
    return df
 
 
def sanity_check(df: pd.DataFrame) -> CommodityKey:
    if df["Full Account Name"].duplicated().any():
        dups = df[df["Full Account Name"].duplicated()]["Full Account Name"].tolist()[:20]
        die(f"Duplicate Full Account Name values found (first 20): {dups}")
 
    bad = df[df.apply(lambda r: str(r["Full Account Name"]).split(":")[-1] != str(r["Account Name"]), axis=1)]
    if not bad.empty:
        sample = bad[["Full Account Name", "Account Name"]].head(10).to_dict(orient="records")
        die(f"Account Name must match the last component of Full Account Name. Examples: {sample}")
 
    bad_types = sorted(set(df["Type"]) - GN_ACCOUNT_TYPES)
    if bad_types:
        die(f"Unsupported account Type(s): {bad_types}. Allowed: {sorted(GN_ACCOUNT_TYPES)}")
 
    full_set = set(df["Full Account Name"])
    parent_map: dict[str, str | None] = {}
    for full in df["Full Account Name"]:
        parts = full.split(":")
        parent = ":".join(parts[:-1]) if len(parts) > 1 else None
        parent_map[full] = parent
        if parent is not None and parent not in full_set:
            die(f"Missing parent account '{parent}' needed for '{full}'")
 
    siblings: dict[str | None, set[str]] = {}
    for full, name in zip(df["Full Account Name"], df["Account Name"], strict=True):
        parent = parent_map[full]
        siblings.setdefault(parent, set())
        if name in siblings[parent]:
            die(f"Duplicate sibling account name '{name}' under parent '{parent or '<ROOT>'}'")
        siblings[parent].add(name)
 
    commodities = {CommodityKey(ns, sym) for ns, sym in zip(df["Namespace"], df["Symbol"], strict=True)}
 
    non_iso_in_currency = sorted([c.symbol for c in commodities if c.namespace == "CURRENCY" and c.symbol not in ISO_CURRENCY_CODES])
    if non_iso_in_currency:
        die(f"Namespace=CURRENCY used with non-ISO currency code(s): {non_iso_in_currency[:50]}")
 
    iso_wrong_ns = sorted([f"{c.namespace}:{c.symbol}" for c in commodities if c.symbol in ISO_CURRENCY_CODES and c.namespace != "CURRENCY"])
    if iso_wrong_ns:
        die("ISO currency codes must use Namespace=CURRENCY. Conflicts found (namespace:symbol): "
            f"{iso_wrong_ns[:50]}")
 
    by_symbol: dict[str, set[str]] = {}
    for c in commodities:
        by_symbol.setdefault(c.symbol, set()).add(c.namespace)
    clashes = sorted([sym for sym, nss in by_symbol.items() if sym in ISO_CURRENCY_CODES and len(nss) > 1])
    if clashes:
        die(f"Currency/stock namespace conflict for ISO currency code(s): {clashes[:50]}")
 
    cur_rows = df[df["Namespace"] == "CURRENCY"]
    if cur_rows.empty:
        die("No CURRENCY commodities found in CSV. At least one currency (e.g. GBP/CURRENCY) is required.")
    home_symbol = cur_rows["Symbol"].mode().iat[0]
    return CommodityKey("CURRENCY", home_symbol)
 
 
def qn(uri: str, local: str) -> str:
    return f"{{{uri}}}{local}"
 
 
NS = {
    "gnc": "http://www.gnucash.org/XML/gnc",
    "act": "http://www.gnucash.org/XML/act",
    "book": "http://www.gnucash.org/XML/book",
    "cd": "http://www.gnucash.org/XML/cd",
    "cmdty": "http://www.gnucash.org/XML/cmdty",
    "slot": "http://www.gnucash.org/XML/slot",
}
 
 
def _indent(elem: ET.Element, level: int = 0) -> None:
    i = "\n" + level * "  "
    if len(elem):
        if not elem.text or not elem.text.strip():
            elem.text = i + "  "
        for child in elem:
            _indent(child, level + 1)
        if not elem[-1].tail or not elem[-1].tail.strip():
            elem[-1].tail = i
    if level and (not elem.tail or not elem.tail.strip()):
        elem.tail = i
 
 
def make_book_slots(parent: ET.Element) -> None:
    slots_el = ET.SubElement(parent, qn(NS["book"], "slots"))
 
    def slot_frame(key: str, inner: list[tuple[str, str, str]]) -> None:
        s = ET.SubElement(slots_el, "slot")
        ET.SubElement(s, qn(NS["slot"], "key")).text = key
        v = ET.SubElement(s, qn(NS["slot"], "value"))
        v.set("type", "frame")
        for ikey, itype, itext in inner:
            inner_s = ET.SubElement(v, "slot")
            ET.SubElement(inner_s, qn(NS["slot"], "key")).text = ikey
            iv = ET.SubElement(inner_s, qn(NS["slot"], "value"))
            iv.set("type", itype)
            iv.text = itext
 
    def slot_string(key: str, value: str) -> None:
        s = ET.SubElement(slots_el, "slot")
        ET.SubElement(s, qn(NS["slot"], "key")).text = key
        v = ET.SubElement(s, qn(NS["slot"], "value"))
        v.set("type", "string")
        v.text = value
 
    counters_keys = ["gncBill", "gncCustomer", "gncEmployee", "gncExpVoucher", "gncInvoice", "gncJob", "gncOrder", "gncVendor"]
    slot_frame("counter_formats", [(k, "string", "") for k in counters_keys])
    slot_frame("counters", [(k, "integer", "0") for k in counters_keys])
    slot_string("remove-trading-splits", "true")
    slot_string("trading-accts", "false")
 
 
def build_gnucash_xml(
    df: pd.DataFrame,
    out_path: Path,
    *,
    book_currency: CommodityKey,
    default_currency_scu: int = 100,
    default_security_fraction: int = 100_000_000,
    commodity_fractions: dict[CommodityKey, int] | None = None,
    deterministic_guids: bool = True,
    include_template_commodity: bool = True,
) -> None:
    for prefix, uri in NS.items():
        ET.register_namespace(prefix, uri)
 
    root = ET.Element("gnc-v2")
 
    count_book = ET.SubElement(root, qn(NS["gnc"], "count-data"))
    count_book.set(qn(NS["cd"], "type"), "book")
    count_book.text = "1"
 
    book = ET.SubElement(root, qn(NS["gnc"], "book"))
    book.set("version", "2.0.0")
 
    bid = ET.SubElement(book, qn(NS["book"], "id"))
    bid.set("type", "guid")
    bid.text = uuid.uuid4().hex
 
    make_book_slots(book)
 
    commodity_fractions = commodity_fractions or {}
 
    def commodity_fraction(key: CommodityKey) -> int | None:
        if key.namespace == "template" and key.symbol == "template":
            return 1
        if key.namespace == "CURRENCY":
            return None
        if key in commodity_fractions:
            return commodity_fractions[key]
        if key.symbol in DEFAULT_SECURITY_FRACTIONS:
            return DEFAULT_SECURITY_FRACTIONS[key.symbol]
        return int(default_security_fraction)
 
    def account_scu(key: CommodityKey) -> int:
        if key.namespace == "CURRENCY":
            return int(default_currency_scu)
        return int(commodity_fraction(key) or default_security_fraction)
 
    needed = {CommodityKey(ns, sym) for ns, sym in zip(df["Namespace"], df["Symbol"], strict=True)}
    if include_template_commodity:
        needed.add(CommodityKey("template", "template"))
 
    count_commodity = ET.SubElement(book, qn(NS["gnc"], "count-data"))
    count_commodity.set(qn(NS["cd"], "type"), "commodity")
    count_commodity.text = str(len(needed))
 
    count_account = ET.SubElement(book, qn(NS["gnc"], "count-data"))
    count_account.set(qn(NS["cd"], "type"), "account")
    count_account.text = str(len(df) + 1)
 
    def emit_commodity(key: CommodityKey) -> None:
        c = ET.SubElement(book, qn(NS["gnc"], "commodity"))
        c.set("version", "2.0.0")
        ET.SubElement(c, qn(NS["cmdty"], "space")).text = key.namespace
        ET.SubElement(c, qn(NS["cmdty"], "id")).text = key.symbol
 
        frac = commodity_fraction(key)
        if key.namespace == "template" and key.symbol == "template":
            ET.SubElement(c, qn(NS["cmdty"], "name")).text = "template"
            ET.SubElement(c, qn(NS["cmdty"], "fraction")).text = "1"
            return
 
        if key.namespace != "CURRENCY":
            ET.SubElement(c, qn(NS["cmdty"], "name")).text = DEFAULT_SECURITY_NAMES.get(key.symbol, key.symbol)
        if frac is not None:
            ET.SubElement(c, qn(NS["cmdty"], "fraction")).text = str(frac)
 
    for key in sorted([k for k in needed if k.namespace == "CURRENCY"], key=lambda k: k.symbol):
        emit_commodity(key)
    for key in sorted([k for k in needed if k.namespace not in {"CURRENCY", "template"}], key=lambda k: (k.namespace, k.symbol)):
        emit_commodity(key)
    if include_template_commodity:
        emit_commodity(CommodityKey("template", "template"))
 
    def guid_for(seed: str) -> str:
        if deterministic_guids:
            return uuid.uuid5(uuid.NAMESPACE_URL, seed).hex
        return uuid.uuid4().hex
 
    root_guid = guid_for("gnucash-account:__ROOT__")
 
    root_acc = ET.SubElement(book, qn(NS["gnc"], "account"))
    root_acc.set("version", "2.0.0")
    ET.SubElement(root_acc, qn(NS["act"], "name")).text = "Root Account"
    rid = ET.SubElement(root_acc, qn(NS["act"], "id"))
    rid.set("type", "guid")
    rid.text = root_guid
    ET.SubElement(root_acc, qn(NS["act"], "type")).text = "ROOT"
    root_cmd = ET.SubElement(root_acc, qn(NS["act"], "commodity"))
    ET.SubElement(root_cmd, qn(NS["cmdty"], "space")).text = book_currency.namespace
    ET.SubElement(root_cmd, qn(NS["cmdty"], "id")).text = book_currency.symbol
    ET.SubElement(root_acc, qn(NS["act"], "commodity-scu")).text = str(account_scu(book_currency))
 
    df2 = df.copy()
    df2["depth"] = df2["Full Account Name"].map(lambda s: str(s).count(":"))
    df2 = df2.sort_values(["depth", "Full Account Name"]).reset_index(drop=True)
 
    guid_map: dict[str, str] = {
        row["Full Account Name"]: guid_for(f"gnucash-account:{row['Full Account Name']}")
        for _, row in df2.iterrows()
    }
 
    def emit_account(row: pd.Series) -> None:
        full = str(row["Full Account Name"])
        name = str(row["Account Name"])
        typ = str(row["Type"])
        desc = str(row["Description"]).strip()
        code = str(row["Account Code"]).strip()
        notes = str(row["Notes"]).strip()
        color = str(row["Account Colour"]).strip()
        placeholder_tf = str(row["Placeholder"]).strip().upper()
 
        hidden_tf = str(row["Hidden"]).strip().upper()
        tax_info_tf = str(row["Tax Info"]).strip().upper()
 
        ckey = CommodityKey(str(row["Namespace"]), str(row["Symbol"]))
 
        a = ET.SubElement(book, qn(NS["gnc"], "account"))
        a.set("version", "2.0.0")
 
        ET.SubElement(a, qn(NS["act"], "name")).text = name
        aid = ET.SubElement(a, qn(NS["act"], "id"))
        aid.set("type", "guid")
        aid.text = guid_map[full]
 
        ET.SubElement(a, qn(NS["act"], "type")).text = typ
 
        acmd = ET.SubElement(a, qn(NS["act"], "commodity"))
        ET.SubElement(acmd, qn(NS["cmdty"], "space")).text = ckey.namespace
        ET.SubElement(acmd, qn(NS["cmdty"], "id")).text = ckey.symbol
        ET.SubElement(a, qn(NS["act"], "commodity-scu")).text = str(account_scu(ckey))
 
        if code:
            ET.SubElement(a, qn(NS["act"], "code")).text = code
        if desc:
            ET.SubElement(a, qn(NS["act"], "description")).text = desc
 
        slots_to_write: list[tuple[str, str, str | None]] = []
        if placeholder_tf == "T":
            slots_to_write.append(("placeholder", "string", "true"))
        if hidden_tf == "T":
            slots_to_write.append(("hidden", "string", "true"))
        if tax_info_tf == "T":
            slots_to_write.append(("tax-related", "string", "true"))
        if notes:
            slots_to_write.append(("notes", "string", notes))
        if color:
            slots_to_write.append(("color", "string", color))
        if typ in {"STOCK", "MUTUAL"}:
            slots_to_write.append(("balance-limit", "frame", None))
 
        if slots_to_write:
            slots_el = ET.SubElement(a, qn(NS["act"], "slots"))
            for key, vtype, vtext in slots_to_write:
                s = ET.SubElement(slots_el, "slot")
                ET.SubElement(s, qn(NS["slot"], "key")).text = key
                v = ET.SubElement(s, qn(NS["slot"], "value"))
                v.set("type", vtype)
                if vtype != "frame":
                    v.text = vtext or ""
 
        parent_full = ":".join(full.split(":")[:-1]) if ":" in full else None
        parent_guid = guid_map[parent_full] if parent_full else root_guid
        ap = ET.SubElement(a, qn(NS["act"], "parent"))
        ap.set("type", "guid")
        ap.text = parent_guid
 
    for _, row in df2.iterrows():
        emit_account(row)
 
    _indent(root)
    out_path.write_bytes(ET.tostring(root, encoding="utf-8", xml_declaration=True))
 
 
def parse_gnucash_accounts_for_verify(gnucash_path: Path) -> pd.DataFrame:
    tree = ET.parse(gnucash_path)
    root = tree.getroot()
 
    ns = {"gnc": NS["gnc"], "act": NS["act"], "cmdty": NS["cmdty"], "slot": NS["slot"]}
    book = root.find("gnc:book", ns)
    if book is None:
        die("Invalid .gnucash file: missing gnc:book")
 
    accs = book.findall("gnc:account", ns)
 
    data: dict[str, dict[str, Any]] = {}
    for a in accs:
        aid = a.find("act:id", ns).text  # type: ignore[union-attr]
        name = (a.find("act:name", ns).text or "").strip()  # type: ignore[union-attr]
        atype = (a.find("act:type", ns).text or "").strip()  # type: ignore[union-attr]
        parent_el = a.find("act:parent", ns)
        parent = parent_el.text.strip() if parent_el is not None and parent_el.text else None
 
        desc_el = a.find("act:description", ns)
        desc = (desc_el.text or "").strip() if desc_el is not None else ""
 
        code_el = a.find("act:code", ns)
        code = (code_el.text or "").strip() if code_el is not None else ""
 
        com = a.find("act:commodity", ns)
        symbol = ""
        namespace = ""
        if com is not None:
            space = com.find("cmdty:space", ns)
            cid = com.find("cmdty:id", ns)
            namespace = (space.text or "").strip() if space is not None else ""  # type: ignore[union-attr]
            symbol = (cid.text or "").strip() if cid is not None else ""  # type: ignore[union-attr]
 
        slots = a.find("act:slots", ns)
        slot_map: dict[str, tuple[str, str]] = {}
        if slots is not None:
            for s in slots.findall("slot", ns):
                key = s.find("slot:key", ns)
                val = s.find("slot:value", ns)
                if key is not None and val is not None:
                    slot_map[(key.text or "").strip()] = (val.get("type") or "", (val.text or "").strip())
 
        def _slot_truthy(key: str) -> bool:
            vtype, vtext = slot_map.get(key, ("", ""))
            vt = (vtype or "").strip().lower()
            vx = (vtext or "").strip().lower()
            if vx in {"true", "t", "1", "yes", "y"}:
                return True
            # Tax related info is often stored as a frame when TXF metadata is set.
            if vt == "frame":
                return True
            return False
 
        placeholder = "T" if _slot_truthy("placeholder") else "F"
        hidden = "T" if _slot_truthy("hidden") else "F"
        tax_info = "T" if _slot_truthy("tax-related") else "F"
        notes = slot_map.get("notes", ("", ""))[1]
        color = slot_map.get("color", ("", ""))[1]
 
        data[aid] = {
            "id": aid,
            "name": name,
            "type": atype,
            "parent": parent,
            "description": desc,
            "code": code,
            "namespace": namespace,
            "symbol": symbol,
            "placeholder": placeholder,
            "hidden": hidden,
            "tax_info": tax_info,
            "notes": notes,
            "color": color,
        }
 
    def full_name(aid: str) -> str:
        parts: list[str] = []
        cur = aid
        while cur and cur in data:
            t = data[cur]["type"]
            if t == "ROOT":
                break
            parts.append(data[cur]["name"])
            cur = data[cur]["parent"]
        return ":".join(reversed(parts))
 
    rows: list[dict[str, Any]] = []
    for aid, info in data.items():
        if info["type"] == "ROOT":
            continue
        rows.append({
            "Type": info["type"],
            "Full Account Name": full_name(aid),
            "Account Name": info["name"],
            "Account Code": info["code"],
            "Description": info["description"],
            "Account Colour": info["color"],
            "Notes": info["notes"],
            "Symbol": str(info["symbol"]).upper(),
            "Namespace": str(info["namespace"]).upper(),
            "Hidden": info["hidden"],
            "Tax Info": info["tax_info"],
            "Placeholder": info["placeholder"],
        })
 
    out = pd.DataFrame(rows).fillna("")
    out = out[EXPECTED_COLUMNS].sort_values(["Full Account Name"]).reset_index(drop=True)
    return out
 
 
def parse_gnucash_commodities(gnucash_path: Path) -> dict[CommodityKey, dict[str, str]]:
    tree = ET.parse(gnucash_path)
    root = tree.getroot()
    ns = {"gnc": NS["gnc"], "cmdty": NS["cmdty"]}
    book = root.find("{" + NS["gnc"] + "}book")
    if book is None:
        die("Invalid .gnucash file: missing gnc:book")
 
    out: dict[CommodityKey, dict[str, str]] = {}
    for c in book.findall("{" + NS["gnc"] + "}commodity"):
        space_el = c.find("{" + NS["cmdty"] + "}space")
        id_el = c.find("{" + NS["cmdty"] + "}id")
        name_el = c.find("{" + NS["cmdty"] + "}name")
        frac_el = c.find("{" + NS["cmdty"] + "}fraction")
        space = (space_el.text or "").strip() if space_el is not None else ""
        cid = (id_el.text or "").strip() if id_el is not None else ""
        key = CommodityKey(space, cid)
        out[key] = {
            "space": space,
            "id": cid,
            "name": (name_el.text or "").strip() if name_el is not None else "",
            "fraction": (frac_el.text or "").strip() if frac_el is not None else "",
        }
    return out
 
 
def verify(
    gnucash_path: Path,
    csv_path: Path,
    *,
    allow_missing: bool,
    allow_extra: bool,
    max_list: int,
    check_commodities: bool,
) -> None:
    df_csv = read_accounts_csv(csv_path).copy()
    df_gc = parse_gnucash_accounts_for_verify(gnucash_path)
 
    for df in (df_csv, df_gc):
        for c in EXPECTED_COLUMNS:
            df[c] = df[c].fillna("").astype(str).str.strip()
 
    set_csv = set(df_csv["Full Account Name"])
    set_gc = set(df_gc["Full Account Name"])
 
    missing_in_gc = sorted(set_csv - set_gc)
    extra_in_gc = sorted(set_gc - set_csv)
 
    if missing_in_gc:
        print("Missing in GnuCash (present in CSV):", file=sys.stderr)
        for name in missing_in_gc[:max_list]:
            print(f"  - {name}", file=sys.stderr)
        if len(missing_in_gc) > max_list:
            print(f"  ... and {len(missing_in_gc) - max_list} more", file=sys.stderr)
 
    if extra_in_gc:
        print("Extra in GnuCash (not in CSV):", file=sys.stderr)
        for name in extra_in_gc[:max_list]:
            print(f"  + {name}", file=sys.stderr)
        if len(extra_in_gc) > max_list:
            print(f"  ... and {len(extra_in_gc) - max_list} more", file=sys.stderr)
 
    if (missing_in_gc and not allow_missing) or (extra_in_gc and not allow_extra):
        die(
            f"Account set mismatch: {len(missing_in_gc)} missing account(s) in GnuCash; "
            f"{len(extra_in_gc)} extra account(s) in GnuCash."
        )
 
    # Compare intersection if non-strict
    common = sorted(set_csv & set_gc)
    df_csv2 = df_csv[df_csv["Full Account Name"].isin(common)].sort_values(["Full Account Name"]).reset_index(drop=True)
    df_gc2 = df_gc[df_gc["Full Account Name"].isin(common)].sort_values(["Full Account Name"]).reset_index(drop=True)
 
    diffs: list[str] = []
    for c in EXPECTED_COLUMNS:
        if not df_csv2[c].equals(df_gc2[c]):
            diffs.append(c)
 
    if diffs:
        for c in diffs:
            neq = df_csv2[c] != df_gc2[c]
            if bool(neq.any()):
                i = int(neq[neq].index[0])
                print(f"First difference in column '{c}':", file=sys.stderr)
                print(f"  Account: {df_csv2.at[i, 'Full Account Name']}", file=sys.stderr)
                print(f"  CSV:     {df_csv2.at[i, c]!r}", file=sys.stderr)
                print(f"  GnuCash: {df_gc2.at[i, c]!r}", file=sys.stderr)
                break
        die(f"Verification failed. Columns differing: {diffs}")
 
    if check_commodities:
        comms = parse_gnucash_commodities(gnucash_path)
 
        # 1) template commodity must remain lowercase (if present)
        tmpl_key_any = None
        for k in comms.keys():
            if k.namespace.lower() == "template" and k.symbol.lower() == "template":
                tmpl_key_any = k
                break
        if tmpl_key_any is not None:
            # Must be EXACTLY template:template with fraction 1 and name template
            if not (tmpl_key_any.namespace == "template" and tmpl_key_any.symbol == "template"):
                die(f"Template commodity regression: found {tmpl_key_any.namespace}:{tmpl_key_any.symbol} (expected template:template)")
            meta = comms[tmpl_key_any]
            if meta.get("fraction", "") not in {"1", ""}:
                die(f"Template commodity fraction must be 1 (found {meta.get('fraction')!r})")
            if meta.get("name", "") not in {"template", ""}:
                die(f"Template commodity name must be 'template' (found {meta.get('name')!r})")
 
        # 2) required securities (non-currency) must exist
        required = {CommodityKey(ns, sym) for ns, sym in zip(df_csv["Namespace"], df_csv["Symbol"], strict=True)}
        required = {k for k in required if k.namespace != "CURRENCY"}
 
        missing = sorted([f"{k.namespace}:{k.symbol}" for k in required if k not in comms])
        if missing:
            die(f"Missing commodity definition(s) in book: {missing[:50]}")
 
        # 3) for known crypto, check expected fraction if present
        bad_frac = []
        for k in required:
            if k.namespace != "CRYPTO":
                continue
            exp = DEFAULT_SECURITY_FRACTIONS.get(k.symbol)
            if exp is None:
                continue
            meta = comms.get(k, {})
            frac_txt = meta.get("fraction", "")
            if frac_txt and frac_txt.isdigit() and int(frac_txt) != exp:
                bad_frac.append(f"{k.namespace}:{k.symbol} fraction={frac_txt} expected={exp}")
        if bad_frac:
            die(f"Commodity fraction mismatch(es): {bad_frac[:50]}")
 
    if allow_missing or allow_extra:
        print(
            f"OK: {gnucash_path.name} matches {csv_path.name} ({len(df_gc2)} account(s)). "
            f"Compared {len(common)} common account(s); missing {len(missing_in_gc)}; extra {len(extra_in_gc)}."
        )
    else:
        print(f"OK: {gnucash_path.name} matches {csv_path.name} ({len(df_gc2)} account(s)).")
 
 
def parse_commodity_fraction_args(items: list[str]) -> dict[CommodityKey, int]:
    out: dict[CommodityKey, int] = {}
    for item in items:
        try:
            key_part, frac_part = item.split("=", 1)
            ns_part, sym_part = key_part.split(":", 1)
            frac = int(frac_part.strip())
            if frac <= 0:
                die(f"Invalid fraction (must be >0): {item!r}")
            out[CommodityKey(ns_part, sym_part)] = frac
        except ValueError:
            die(f"Invalid --commodity-fraction value {item!r}. Expected NAMESPACE:SYMBOL=FRACTION")
    return out
 
 
def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(description="Generate a GnuCash XML book (*.gnucash) from an accounts tree CSV.")
    sub = p.add_subparsers(dest="cmd", required=True)
 
    b = sub.add_parser("build", help="Build a new .gnucash XML file from a CSV")
    b.add_argument("csv", type=Path, help="Input accounts tree CSV")
    b.add_argument("--out", type=Path, required=True, help="Output .gnucash file path")
    b.add_argument("--default-currency-scu", type=int, default=100, help="Default currency SCU (GBP uses 100)")
    b.add_argument("--default-security-fraction", type=int, default=100_000_000, help="Default fraction/SCU for non-currency commodities")
    b.add_argument("--commodity-fraction", action="append", default=[], help="Override fraction: NAMESPACE:SYMBOL=FRACTION (repeatable)")
    b.add_argument("--random-guids", action="store_true", help="Use random GUIDs (default is deterministic)")
    b.add_argument("--home-currency", type=str, default="", help="Override book currency (e.g. GBP). Default: most common currency in CSV")
    b.add_argument("--no-template-commodity", action="store_true", help="Do not emit the internal template:template commodity")
 
    v = sub.add_parser("verify", help="Verify an existing .gnucash matches a CSV (same schema)")
    v.add_argument("gnucash", type=Path, help="Existing .gnucash XML file")
    v.add_argument("csv", type=Path, help="CSV to compare against")
    v.add_argument("--allow-missing", action="store_true", help="Allow accounts present in CSV but missing in GnuCash")
    v.add_argument("--allow-extra", action="store_true", help="Allow accounts present in GnuCash but missing in CSV")
    v.add_argument("--max-list", type=int, default=50, help="Max missing/extra accounts to list")
    v.add_argument("--no-check-commodities", action="store_true", help="Skip commodity checks (template + securities)")
 
    args = p.parse_args(argv)
 
    if args.cmd == "verify":
        verify(
            args.gnucash,
            args.csv,
            allow_missing=args.allow_missing,
            allow_extra=args.allow_extra,
            max_list=args.max_list,
            check_commodities=not args.no_check_commodities,
        )
        return 0
 
    if args.cmd == "build":
        df = read_accounts_csv(args.csv)
        book_currency = sanity_check(df)
 
        if args.home_currency:
            sym = args.home_currency.strip().upper()
            if sym not in ISO_CURRENCY_CODES:
                die(f"--home-currency {sym!r} is not an ISO 4217 currency code")
            book_currency = CommodityKey("CURRENCY", sym)
 
        overrides = parse_commodity_fraction_args(args.commodity_fraction)
 
        build_gnucash_xml(
            df,
            args.out,
            book_currency=book_currency,
            default_currency_scu=args.default_currency_scu,
            default_security_fraction=args.default_security_fraction,
            commodity_fractions=overrides,
            deterministic_guids=not args.random_guids,
            include_template_commodity=not args.no_template_commodity,
        )
        print(f"Wrote: {args.out}")
        return 0
 
    die("Unknown command")
    return 2
 
 
if __name__ == "__main__":
    raise SystemExit(main())
 

Comments