CCMA/src/ccma/services/gnucash_import.py

from __future__ import annotations

import gzip
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal, InvalidOperation
from pathlib import Path
from xml.etree import ElementTree as ET

NS = {
    "gnc": "http://www.gnucash.org/XML/gnc",
    "act": "http://www.gnucash.org/XML/act",
    "trn": "http://www.gnucash.org/XML/trn",
    "split": "http://www.gnucash.org/XML/split",
    "ts": "http://www.gnucash.org/XML/ts",
}

CENT = Decimal("0.01")


class GnuCashImportError(RuntimeError):
    pass


@dataclass(frozen=True, slots=True)
class GnuCashAccount:
    guid: str
    name: str
    full_name: str
    account_type: str


@dataclass(frozen=True, slots=True)
class GnuCashTransaction:
    guid: str
    date: date
    description: str
    memo: str
    amount: Decimal


def _read_root(path: Path) -> ET.Element:
    try:
        raw = path.read_bytes()
    except OSError as exc:
        raise GnuCashImportError(f"GnuCash-Datei konnte nicht gelesen werden: {exc}") from exc
    if raw[:2] == b"\x1f\x8b":
        try:
            raw = gzip.decompress(raw)
        except OSError as exc:
            raise GnuCashImportError(f"GnuCash-Datei konnte nicht entpackt werden: {exc}") from exc
    try:
        return ET.fromstring(raw)
    except ET.ParseError as exc:
        raise GnuCashImportError(f"GnuCash-Datei ist kein gültiges XML: {exc}") from exc


def _book(root: ET.Element) -> ET.Element:
    book = root.find("gnc:book", NS)
    if book is None:
        raise GnuCashImportError("Keine GnuCash-Buchführung (gnc:book) in der Datei gefunden.")
    return book


def _text(element: ET.Element, path: str) -> str:
    return (element.findtext(path, default="", namespaces=NS) or "").strip()


def list_accounts(path: str | Path) -> list[GnuCashAccount]:
    root = _read_root(Path(path))
    book = _book(root)
    by_guid: dict[str, ET.Element] = {}
    parent_of: dict[str, str] = {}
    for account_el in book.findall("gnc:account", NS):
        guid = _text(account_el, "act:id")
        if not guid:
            continue
        by_guid[guid] = account_el
        parent_of[guid] = _text(account_el, "act:parent")
    if not by_guid:
        raise GnuCashImportError("Die Datei enthält keine Konten.")

    def full_name(guid: str, seen: frozenset[str]) -> str:
        account_el = by_guid.get(guid)
        if account_el is None or guid in seen or _text(account_el, "act:type") == "ROOT":
            return ""
        name = _text(account_el, "act:name")
        parent = parent_of.get(guid, "")
        if not parent:
            return name
        parent_name = full_name(parent, seen | {guid})
        return f"{parent_name}:{name}" if parent_name else name

    accounts = []
    for guid, account_el in by_guid.items():
        account_type = _text(account_el, "act:type")
        if account_type in {"ROOT", ""}:
            continue
        accounts.append(
            GnuCashAccount(
                guid=guid,
                name=_text(account_el, "act:name"),
                full_name=full_name(guid, frozenset()) or _text(account_el, "act:name"),
                account_type=account_type,
            )
        )
    accounts.sort(key=lambda item: item.full_name.casefold())
    return accounts


def _parse_fraction(text: str) -> Decimal | None:
    text = text.strip()
    if "/" not in text:
        return None
    numerator, _, denominator = text.partition("/")
    try:
        denom = Decimal(denominator)
        if denom == 0:
            return None
        return (Decimal(numerator) / denom).quantize(CENT)
    except InvalidOperation:
        return None


def _parse_posted_date(text: str) -> date | None:
    text = text.strip()
    if not text:
        return None
    for fmt in ("%Y-%m-%d %H:%M:%S %z", "%Y-%m-%d"):
        try:
            return datetime.strptime(text, fmt).date()
        except ValueError:
            continue
    return None


def list_transactions(path: str | Path, account_guid: str) -> list[GnuCashTransaction]:
    """Return every booking that has a split on the given account, one row per
    matching split (a transaction with two splits on the same account, e.g. a
    transfer to itself, yields two rows -- rare in practice, harmless here)."""
    root = _read_root(Path(path))
    book = _book(root)
    results: list[GnuCashTransaction] = []
    for transaction_el in book.findall("gnc:transaction", NS):
        guid = _text(transaction_el, "trn:id")
        posted_date = _parse_posted_date(_text(transaction_el, "trn:date-posted/ts:date"))
        if not guid or posted_date is None:
            continue
        description = _text(transaction_el, "trn:description")
        splits_el = transaction_el.find("trn:splits", NS)
        if splits_el is None:
            continue
        for split_el in splits_el.findall("trn:split", NS):
            if _text(split_el, "split:account") != account_guid:
                continue
            amount = _parse_fraction(_text(split_el, "split:value"))
            if amount is None:
                continue
            results.append(
                GnuCashTransaction(
                    guid=guid,
                    date=posted_date,
                    description=description,
                    memo=_text(split_el, "split:memo"),
                    amount=amount,
                )
            )
    results.sort(key=lambda item: (item.date, item.description.casefold()))
    return results