diff --git a/src/ccma/services/gnucash_import.py b/src/ccma/services/gnucash_import.py new file mode 100644 index 0000000..6d26881 --- /dev/null +++ b/src/ccma/services/gnucash_import.py @@ -0,0 +1,169 @@ +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 diff --git a/src/ccma/ui/gnucash_import_dialog.py b/src/ccma/ui/gnucash_import_dialog.py new file mode 100644 index 0000000..74754fd --- /dev/null +++ b/src/ccma/ui/gnucash_import_dialog.py @@ -0,0 +1,349 @@ +from __future__ import annotations + +import tkinter as tk +from collections.abc import Callable +from decimal import Decimal +from pathlib import Path +from tkinter import filedialog, messagebox, ttk + +from ccma.config import AppConfig +from ccma.domain.contributions import money_text +from ccma.domain.dates import ( + DateValidationError, + date_input_hint, + format_date_for_display, + normalize_date_input, +) +from ccma.services.gnucash_import import ( + GnuCashAccount, + GnuCashImportError, + GnuCashTransaction, + list_accounts, + list_transactions, +) +from ccma.storage.repository import MemberRepository, RepositoryError + + +class GnuCashImportDialog(tk.Toplevel): + """Lets the board point at a GnuCash file, pick one account, narrow its bookings + down with filters, and tick which ones to import as payments for this member. + + Only bookings with a positive amount on the selected account are offered (money + coming in), and any booking whose date + amount already match an existing payment + for this member is flagged and cannot be checked -- avoiding accidental double + imports if the same statement is imported twice.""" + + def __init__( + self, + master: tk.Misc, + repository: MemberRepository, + member_id: str, + config: AppConfig, + on_imported: Callable[[], None], + ): + super().__init__(master) + self.repository = repository + self.member_id = member_id + self.config = config + self.on_imported = on_imported + + self.accounts: list[GnuCashAccount] = [] + self.account_by_label: dict[str, GnuCashAccount] = {} + self.transactions: list[GnuCashTransaction] = [] + self.selected_guids: set[str] = set() + self.existing_payment_keys: set[tuple[str, str]] = set() + + self.title("Zahlungen aus GnuCash importieren") + self.transient(master.winfo_toplevel()) + self.geometry("960x640") + self.minsize(780, 480) + self.protocol("WM_DELETE_WINDOW", self.destroy) + self.bind("", lambda _event: self.destroy()) + self._build_ui() + self._load_existing_payment_keys() + self.after_idle(self._activate) + if self.file_var.get().strip(): + self._load_file() + + def _activate(self) -> None: + try: + self.deiconify() + self.lift() + self.focus_force() + self.grab_set() + except tk.TclError: + return + + def _build_ui(self) -> None: + self.columnconfigure(0, weight=1) + self.rowconfigure(2, weight=1) + + file_frame = ttk.Frame(self, padding=(16, 16, 16, 8)) + file_frame.grid(row=0, column=0, sticky="ew") + file_frame.columnconfigure(1, weight=1) + self.file_var = tk.StringVar(value=self.config.gnucash_path) + ttk.Label(file_frame, text="GnuCash-Datei").grid(row=0, column=0, sticky="w", padx=(0, 12)) + ttk.Entry(file_frame, textvariable=self.file_var).grid(row=0, column=1, sticky="ew") + ttk.Button(file_frame, text="Durchsuchen…", command=self._browse_file).grid( + row=0, column=2, padx=(8, 0) + ) + ttk.Label(file_frame, text="Konto").grid(row=1, column=0, sticky="w", padx=(0, 12), pady=(8, 0)) + self.account_var = tk.StringVar() + self.account_combo = ttk.Combobox( + file_frame, textvariable=self.account_var, state="readonly" + ) + self.account_combo.grid(row=1, column=1, sticky="ew", pady=(8, 0)) + self.account_combo.bind("<>", lambda _event: self._account_selected()) + ttk.Button(file_frame, text="Datei laden", command=self._load_file).grid( + row=1, column=2, padx=(8, 0), pady=(8, 0) + ) + + filter_frame = ttk.Frame(self, padding=(16, 0, 16, 8)) + filter_frame.grid(row=1, column=0, sticky="ew") + ttk.Label(filter_frame, text="Beschreibung enthält").pack(side="left") + self.description_filter_var = tk.StringVar() + ttk.Entry(filter_frame, textvariable=self.description_filter_var, width=26).pack( + side="left", padx=(8, 16) + ) + self.description_filter_var.trace_add("write", lambda *_args: self._render_transactions()) + ttk.Label(filter_frame, text=f"Von ({date_input_hint()})").pack(side="left") + self.date_from_var = tk.StringVar() + ttk.Entry(filter_frame, textvariable=self.date_from_var, width=12).pack(side="left", padx=(8, 16)) + self.date_from_var.trace_add("write", lambda *_args: self._render_transactions()) + ttk.Label(filter_frame, text=f"Bis ({date_input_hint()})").pack(side="left") + self.date_to_var = tk.StringVar() + ttk.Entry(filter_frame, textvariable=self.date_to_var, width=12).pack(side="left", padx=(8, 0)) + self.date_to_var.trace_add("write", lambda *_args: self._render_transactions()) + + tree_frame = ttk.Frame(self, padding=(16, 0, 16, 8)) + tree_frame.grid(row=2, column=0, sticky="nsew") + tree_frame.columnconfigure(0, weight=1) + tree_frame.rowconfigure(0, weight=1) + self.tree = ttk.Treeview( + tree_frame, + columns=("selected", "date", "description", "memo", "amount", "status"), + show="headings", + selectmode="browse", + ) + for key, title, width in ( + ("selected", "Import", 60), + ("date", "Datum", 100), + ("description", "Beschreibung", 280), + ("memo", "Memo", 180), + ("amount", "Betrag", 100), + ("status", "Status", 220), + ): + self.tree.heading(key, text=title) + self.tree.column(key, width=width, anchor="w") + self.tree.column("selected", anchor="center", stretch=False) + self.tree.grid(row=0, column=0, sticky="nsew") + scrollbar = ttk.Scrollbar(tree_frame, orient="vertical", command=self.tree.yview) + scrollbar.grid(row=0, column=1, sticky="ns") + self.tree.configure(yscrollcommand=scrollbar.set) + self.tree.tag_configure("duplicate", background="#7a2323", foreground="#ffffff") + self.tree.bind("", self._on_tree_click) + + ttk.Label( + self, + text=( + "Es werden nur Buchungen mit positivem Betrag auf dem gewählten Konto angezeigt " + "(eingehende Zahlungen). Rot markierte Buchungen haben Datum und Betrag einer bereits " + "vorhandenen Zahlung dieses Mitglieds und können nicht erneut importiert werden." + ), + style="Mono.TLabel", + wraplength=900, + ).grid(row=3, column=0, sticky="w", padx=16) + + summary_frame = ttk.Frame(self, padding=(16, 8, 16, 8)) + summary_frame.grid(row=4, column=0, sticky="ew") + self.summary_var = tk.StringVar(value="Keine Datei geladen.") + ttk.Label(summary_frame, textvariable=self.summary_var, style="Mono.TLabel").pack(side="left") + + buttons = ttk.Frame(self, padding=(16, 0, 16, 16)) + buttons.grid(row=5, column=0, sticky="e") + ttk.Button(buttons, text="Schließen", command=self.destroy).pack(side="left", padx=(0, 8)) + self.import_button = ttk.Button( + buttons, + text="Ausgewählte importieren", + style="Accent.TButton", + command=self._import_selected, + ) + self.import_button.pack(side="left") + + def _load_existing_payment_keys(self) -> None: + data = self.repository.get_contributions(self.member_id) + self.existing_payment_keys = { + (str(payment.get("date", "")), money_text(payment.get("amount", "0"))) + for payment in data.payments + } + + def _browse_file(self) -> None: + current_text = self.file_var.get().strip() + current = Path(current_text).expanduser() if current_text else Path.home() + initial_dir = current.parent if current.is_file() else current + selected = filedialog.askopenfilename( + parent=self, + title="GnuCash-Datei auswählen", + initialdir=str(initial_dir) if initial_dir.is_dir() else str(Path.home()), + filetypes=[("GnuCash-Dateien", "*.gnucash *.xac"), ("Alle Dateien", "*.*")], + ) + if selected: + self.file_var.set(selected) + self._load_file() + + def _load_file(self) -> None: + path_text = self.file_var.get().strip() + if not path_text: + messagebox.showinfo("GnuCash-Datei", "Bitte zuerst eine GnuCash-Datei auswählen.", parent=self) + return + path = Path(path_text).expanduser() + if not path.is_file(): + messagebox.showerror("GnuCash-Datei", "Die ausgewählte Datei existiert nicht.", parent=self) + return + try: + self.accounts = list_accounts(path) + except GnuCashImportError as exc: + messagebox.showerror("GnuCash-Datei konnte nicht gelesen werden", str(exc), parent=self) + return + self.account_by_label = { + f"{account.full_name} ({account.account_type})": account for account in self.accounts + } + self.account_combo.configure(values=list(self.account_by_label)) + if self.account_by_label and self.account_var.get() not in self.account_by_label: + self.account_var.set(next(iter(self.account_by_label))) + self.selected_guids.clear() + if self.account_var.get(): + self._account_selected() + else: + self.transactions = [] + self._render_transactions() + if path_text != self.config.gnucash_path: + self.config.gnucash_path = path_text + self.config.save() + + def _account_selected(self) -> None: + account = self.account_by_label.get(self.account_var.get()) + if account is None: + return + path = Path(self.file_var.get().strip()).expanduser() + try: + all_transactions = list_transactions(path, account.guid) + except GnuCashImportError as exc: + messagebox.showerror("GnuCash-Datei konnte nicht gelesen werden", str(exc), parent=self) + self.transactions = [] + self._render_transactions() + return + self.transactions = [item for item in all_transactions if item.amount > 0] + self.selected_guids.clear() + self._render_transactions() + + def _filtered_transactions(self) -> list[GnuCashTransaction]: + needle = self.description_filter_var.get().strip().casefold() + try: + date_from = normalize_date_input(self.date_from_var.get(), "Von") or None + except DateValidationError: + date_from = None + try: + date_to = normalize_date_input(self.date_to_var.get(), "Bis") or None + except DateValidationError: + date_to = None + result = [] + for item in self.transactions: + if needle and needle not in item.description.casefold() and needle not in item.memo.casefold(): + continue + iso = item.date.isoformat() + if date_from and iso < date_from: + continue + if date_to and iso > date_to: + continue + result.append(item) + return result + + def _is_duplicate(self, item: GnuCashTransaction) -> bool: + return (item.date.isoformat(), money_text(item.amount)) in self.existing_payment_keys + + def _render_transactions(self) -> None: + self.tree.delete(*self.tree.get_children()) + for item in self._filtered_transactions(): + duplicate = self._is_duplicate(item) + if duplicate: + self.selected_guids.discard(item.guid) + checked = item.guid in self.selected_guids + self.tree.insert( + "", + "end", + iid=item.guid, + values=( + "☑" if checked else "☐", + format_date_for_display(item.date.isoformat()), + item.description, + item.memo, + f"{money_text(item.amount)} EUR", + "Bereits vorhanden" if duplicate else "", + ), + tags=("duplicate",) if duplicate else (), + ) + self._update_summary() + + def _on_tree_click(self, event: tk.Event) -> None: + row_id = self.tree.identify_row(event.y) + column = self.tree.identify_column(event.x) + if not row_id or column != "#1": + return + item = next((entry for entry in self.transactions if entry.guid == row_id), None) + if item is None or self._is_duplicate(item): + return + if row_id in self.selected_guids: + self.selected_guids.discard(row_id) + else: + self.selected_guids.add(row_id) + self._render_transactions() + + def _update_summary(self) -> None: + filtered = self._filtered_transactions() + selected_items = [item for item in filtered if item.guid in self.selected_guids] + total = sum((item.amount for item in selected_items), Decimal("0")) + self.summary_var.set( + f"{len(filtered)} Buchungen gefunden · {len(selected_items)} ausgewählt · " + f"{money_text(total)} EUR" + ) + + def _import_selected(self) -> None: + selected_items = [ + item + for item in self.transactions + if item.guid in self.selected_guids and not self._is_duplicate(item) + ] + if not selected_items: + messagebox.showinfo("Import", "Bitte mindestens eine Buchung auswählen.", parent=self) + return + imported = 0 + errors: list[str] = [] + for item in selected_items: + try: + self.repository.create_payment( + self.member_id, + payment_date=item.date.isoformat(), + amount=money_text(item.amount), + reference=item.description, + gnucash_transaction_id=item.guid, + ) + imported += 1 + self.selected_guids.discard(item.guid) + except RepositoryError as exc: + errors.append( + f"{format_date_for_display(item.date.isoformat())} · {money_text(item.amount)} EUR: {exc}" + ) + if imported: + self._load_existing_payment_keys() + self._render_transactions() + if errors: + messagebox.showwarning( + "Import teilweise fehlgeschlagen", + f"{imported} Zahlung(en) importiert.\n\nNicht importiert:\n" + "\n".join(errors), + parent=self, + ) + else: + messagebox.showinfo("Import abgeschlossen", f"{imported} Zahlung(en) importiert.", parent=self) + if imported: + self.on_imported() diff --git a/src/ccma/ui/main_window.py b/src/ccma/ui/main_window.py index edb591f..caef85e 100644 --- a/src/ccma/ui/main_window.py +++ b/src/ccma/ui/main_window.py @@ -295,6 +295,7 @@ class MainWindow(ttk.Frame): self.notebook, self.repository, member_id, + self.config, on_close=lambda: self.tabs.close(key), on_changed=self.refresh_overview, on_open_claim=self.open_claim, diff --git a/src/ccma/ui/member_tab.py b/src/ccma/ui/member_tab.py index 128207f..12321b7 100644 --- a/src/ccma/ui/member_tab.py +++ b/src/ccma/ui/member_tab.py @@ -7,6 +7,7 @@ from decimal import Decimal from pathlib import Path from tkinter import messagebox, ttk +from ccma.config import AppConfig from ccma.domain.contributions import ( CLAIM_STATUS_LABELS, DONATION_STATUS_LABELS, @@ -30,6 +31,7 @@ from ccma.ui.donation_dialog import ( DonationPaymentDialog, ) from ccma.ui.file_open import open_path +from ccma.ui.gnucash_import_dialog import GnuCashImportDialog from ccma.ui.labels import display_label, storage_key from ccma.ui.messages import MessageAction, MessageBannerList, TabMessage from ccma.ui.payment_dialog import PaymentCreateDialog, PaymentEditDialog @@ -62,6 +64,7 @@ class MemberTab(ttk.Frame): master: tk.Misc, repository: MemberRepository, member_id: str, + config: AppConfig, on_close: Callable[[], None], on_changed: Callable[[], None], on_open_claim: Callable[[str, str], None], @@ -72,6 +75,7 @@ class MemberTab(ttk.Frame): super().__init__(master, padding=12) self.repository = repository self.member_id = member_id + self.config = config self.on_close = on_close self.on_changed = on_changed self.on_open_claim = on_open_claim @@ -325,6 +329,10 @@ class MemberTab(ttk.Frame): ttk.Button(payment_actions, text="Zahlung löschen", command=self._delete_selected_payment).pack( side="left" ) + ttk.Separator(payment_actions, orient="vertical").pack(side="left", fill="y", padx=10) + ttk.Button( + payment_actions, text="Zahlungen importieren", command=self._import_payments + ).pack(side="left") donations_tab.columnconfigure(0, weight=1) donations_tab.rowconfigure(1, weight=1) @@ -733,6 +741,11 @@ class MemberTab(ttk.Frame): def _create_payment(self) -> None: PaymentCreateDialog(self, self.repository, self.member_id, self._payment_changed) + def _import_payments(self) -> None: + GnuCashImportDialog( + self, self.repository, self.member_id, self.config, self._payment_changed + ) + def _payment_changed(self) -> None: self.refresh() self.on_changed() diff --git a/tests/test_gnucash_import.py b/tests/test_gnucash_import.py new file mode 100644 index 0000000..1fd0de3 --- /dev/null +++ b/tests/test_gnucash_import.py @@ -0,0 +1,156 @@ +import gzip +from decimal import Decimal + +import pytest + +from ccma.services.gnucash_import import GnuCashImportError, list_accounts, list_transactions + +SAMPLE_GNUCASH_XML = """ + + + book-1 + + Root Account + root-guid + ROOT + + + Aktiva + assets-guid + ASSET + root-guid + + + Girokonto + bank-guid + BANK + assets-guid + + + Beitraege + income-guid + INCOME + root-guid + + + tx-1 + 2026-06-21 10:59:00 +0200 + Mitgliedsbeitrag Max Mustermann + + + split-1a + Beitrag Juni + 15000/100 + 15000/100 + bank-guid + + + split-1b + + -15000/100 + -15000/100 + income-guid + + + + + tx-2 + 2026-07-01 08:30:00 +0200 + Bankgebuehr + + + split-2a + + -500/100 + -500/100 + bank-guid + + + split-2b + + 500/100 + 500/100 + income-guid + + + + + +""" + + +def _write_sample(tmp_path, *, gzipped: bool = False, suffix: str = ".gnucash"): + path = tmp_path / f"test{suffix}" + data = SAMPLE_GNUCASH_XML.encode("utf-8") + if gzipped: + data = gzip.compress(data) + path.write_bytes(data) + return path + + +def test_list_accounts_builds_hierarchical_full_names(tmp_path) -> None: + path = _write_sample(tmp_path) + accounts = list_accounts(path) + by_guid = {account.guid: account for account in accounts} + + assert "root-guid" not in by_guid + assert by_guid["bank-guid"].full_name == "Aktiva:Girokonto" + assert by_guid["bank-guid"].account_type == "BANK" + assert by_guid["assets-guid"].full_name == "Aktiva" + assert by_guid["income-guid"].full_name == "Beitraege" + + +def test_list_transactions_returns_splits_for_selected_account(tmp_path) -> None: + path = _write_sample(tmp_path) + transactions = list_transactions(path, "bank-guid") + + assert [item.guid for item in transactions] == ["tx-1", "tx-2"] + first = transactions[0] + assert first.date.isoformat() == "2026-06-21" + assert first.description == "Mitgliedsbeitrag Max Mustermann" + assert first.memo == "Beitrag Juni" + assert first.amount == Decimal("150.00") + second = transactions[1] + assert second.amount == Decimal("-5.00") + + +def test_list_transactions_only_includes_splits_on_the_requested_account(tmp_path) -> None: + path = _write_sample(tmp_path) + transactions = list_transactions(path, "income-guid") + + assert [item.guid for item in transactions] == ["tx-1", "tx-2"] + assert transactions[0].amount == Decimal("-150.00") + assert transactions[1].amount == Decimal("5.00") + + +def test_gzip_compressed_gnucash_file_is_read_transparently(tmp_path) -> None: + path = _write_sample(tmp_path, gzipped=True, suffix=".gnucash") + accounts = list_accounts(path) + assert any(account.guid == "bank-guid" for account in accounts) + transactions = list_transactions(path, "bank-guid") + assert len(transactions) == 2 + + +def test_invalid_xml_raises_gnucash_import_error(tmp_path) -> None: + path = tmp_path / "broken.gnucash" + path.write_text("not xml at all <<<", encoding="utf-8") + with pytest.raises(GnuCashImportError): + list_accounts(path) + + +def test_missing_file_raises_gnucash_import_error(tmp_path) -> None: + with pytest.raises(GnuCashImportError): + list_accounts(tmp_path / "does-not-exist.gnucash") + + +def test_file_without_book_raises_gnucash_import_error(tmp_path) -> None: + path = tmp_path / "empty.gnucash" + path.write_text('', encoding="utf-8") + with pytest.raises(GnuCashImportError, match="gnc:book"): + list_accounts(path)