initial load of categories

AGENTS.md

@@ -144,3 +144,7 @@ uv run pytest --cov=src --cov-report=term-missing
 ```
 
 **Note**: Some comparisons between `date | None` types require explicit guards or type assertions to satisfy mypy, even when the logic is correct. See `src/domain_utils.py` for examples.
+
+## Code Style
+
+- Don't be too verbose in docstrings and comments

scripts/initial_load/categories/categories.txt

@@ -0,0 +1,35 @@
+Girokonto:Beruf / Ausbildung
+Girokonto:Beruf / Ausbildung:Gehalt
+Girokonto:Beruf / Ausbildung:Reisekosten
+Girokonto:Freizeit
+Girokonto:Freizeit:Abo
+Girokonto:Freizeit:Elektronik&Hardware
+Girokonto:Freizeit:Foto
+Girokonto:Freizeit:Medien
+Girokonto:Freizeit:Musikintrumente
+Girokonto:Freizeit:Urlaub&Ausflüge
+Girokonto:Freizeit:sonst.
+Girokonto:Lebenshaltung
+Girokonto:Lebenshaltung:Auto
+Girokonto:Lebenshaltung:Bekleidung
+Girokonto:Lebenshaltung:Fahrrad
+Girokonto:Lebenshaltung:Gesundheit
+Girokonto:Lebenshaltung:Haushalt sonst.
+Girokonto:Lebenshaltung:Heimwerken
+Girokonto:Lebenshaltung:Kontoführung
+Girokonto:Lebenshaltung:Körperpflege
+Girokonto:Lebenshaltung:Lebensmittel
+Girokonto:Lebenshaltung:Telekommunikation
+Girokonto:Lebenshaltung:Versicherungen
+Girokonto:Privat
+Girokonto:Privat:Geschenke
+Girokonto:Privat:Mitgliedsbeiträge
+Girokonto:Privat:Steuerausgleich Ehetine
+Girokonto:Sonstiges
+Girokonto:Sparen
+Girokonto:Sparen:Depot
+Girokonto:Sparen:Tagesgeld
+Girokonto:unbestimmte Verwendung:Bargeld
+Girokonto:unbestimmte Verwendung:Kreditkarte
+Haushaltskonto:Beiträge
+Umbuchung

scripts/initial_load/categories/extract_categories.py

@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+"""Extract unique categories from b4w CSV export."""
+
+import argparse
+import csv
+from pathlib import Path
+
+
+def extract_unique_categories(csv_path: str, output_path: str) -> list[str]:
+    """Merge CSV categories with the existing output file, sorted, one per line."""
+
+    unique_categories: set[str] = set()
+
+    with open(csv_path, "r", encoding="utf-8") as f:
+        reader = csv.DictReader(f)
+        for row in reader:
+            category = (row.get("Category") or "").strip()
+            if category:
+                unique_categories.add(category)
+
+    output_file_path = Path(output_path)
+    if output_file_path.exists():
+        with open(output_file_path, "r", encoding="utf-8") as f:
+            for line in f:
+                if category := line.strip():
+                    unique_categories.add(category)
+
+    sorted_categories = sorted(unique_categories)
+
+    with open(output_file_path, "w", encoding="utf-8") as f:
+        f.writelines(f"{cat}\n" for cat in sorted_categories)
+
+    print(f"Extracted {len(sorted_categories)} unique categories")
+    print(f"Output written to: {output_path}")
+    return sorted_categories
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Extract unique categories from a b4w CSV export."
+    )
+    parser.add_argument("--csv", required=True, help="Pfad der b4w-Exportdatei (CSV)")
+    parser.add_argument(
+        "--output",
+        default="scripts/initial_load/categories/categories.txt",
+        help="Pfad der Ausgabedatei (Default: %(default)s)",
+    )
+    args = parser.parse_args()
+
+    extract_unique_categories(args.csv, args.output)
+
+
+if __name__ == "__main__":
+    main()

tests/test_extract_categories.py

@@ -0,0 +1,125 @@
+"""Tests for extract_categories.py - extracting unique categories from b4w CSV exports."""
+
+import csv
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))
+
+from scripts.initial_load.categories.extract_categories import extract_unique_categories
+
+
+def write_csv(path: Path, categories: list[str]) -> None:
+    with open(path, "w", encoding="utf-8") as f:
+        writer = csv.DictWriter(f, fieldnames=["Id", "AcctId", "Category"])
+        writer.writeheader()
+        for i, category in enumerate(categories, 1):
+            writer.writerow({"Id": str(i), "AcctId": "587713594", "Category": category})
+
+
+def read_lines(path: Path) -> list[str]:
+    return path.read_text(encoding="utf-8").splitlines()
+
+
+def test_extract_unique_categories_basic(tmp_path):
+    csv_file = tmp_path / "input.csv"
+    output_file = tmp_path / "categories.txt"
+    write_csv(csv_file, ["Girokonto:Lebenshaltung", "Girokonto:Sparen"])
+
+    result = extract_unique_categories(str(csv_file), str(output_file))
+
+    assert result == ["Girokonto:Lebenshaltung", "Girokonto:Sparen"]
+    assert read_lines(output_file) == ["Girokonto:Lebenshaltung", "Girokonto:Sparen"]
+
+
+def test_extract_unique_categories_no_duplicates(tmp_path):
+    csv_file = tmp_path / "input.csv"
+    output_file = tmp_path / "categories.txt"
+    write_csv(
+        csv_file,
+        ["Girokonto:Lebenshaltung", "Girokonto:Sparen", "Girokonto:Lebenshaltung"],
+    )
+
+    extract_unique_categories(str(csv_file), str(output_file))
+
+    assert len(read_lines(output_file)) == 2
+
+
+def test_extract_unique_categories_output_format(tmp_path):
+    csv_file = tmp_path / "input.csv"
+    output_file = tmp_path / "categories.txt"
+    write_csv(
+        csv_file,
+        ["Girokonto:Lebenshaltung", "Girokonto:Sparen", "Girokonto:Privat"],
+    )
+
+    extract_unique_categories(str(csv_file), str(output_file))
+
+    lines = read_lines(output_file)
+    assert len(lines) == 3
+    # Plain list: no numbering prefix, no blank lines.
+    assert all(line == line.strip() and not line[0].isdigit() for line in lines)
+
+
+def test_extract_unique_categories_german_special_chars(tmp_path):
+    csv_file = tmp_path / "input.csv"
+    output_file = tmp_path / "categories.txt"
+    write_csv(
+        csv_file,
+        ["Girokonto:Freizeit:Urlaub&Ausflüge", "Girokonto:Privat:Mitgliedsbeiträge"],
+    )
+
+    extract_unique_categories(str(csv_file), str(output_file))
+
+    assert read_lines(output_file) == [
+        "Girokonto:Freizeit:Urlaub&Ausflüge",
+        "Girokonto:Privat:Mitgliedsbeiträge",
+    ]
+
+
+def test_extract_unique_categories_handles_empty_csv(tmp_path):
+    csv_file = tmp_path / "input.csv"
+    output_file = tmp_path / "categories.txt"
+    csv_file.write_text("Id,AcctId,Category\n", encoding="utf-8")
+
+    extract_unique_categories(str(csv_file), str(output_file))
+
+    assert output_file.exists()
+    assert read_lines(output_file) == []
+
+
+def test_extract_unique_categories_handles_csv_with_all_empty_categories(tmp_path):
+    csv_file = tmp_path / "input.csv"
+    output_file = tmp_path / "categories.txt"
+    csv_file.write_text(
+        "Id,AcctId,Category\n1,587713594,\n2,587713594,\n3,587713594,\n",
+        encoding="utf-8",
+    )
+
+    extract_unique_categories(str(csv_file), str(output_file))
+
+    assert read_lines(output_file) == []
+
+
+def test_extract_unique_categories_rerun_is_idempotent(tmp_path):
+    csv_file = tmp_path / "input.csv"
+    output_file = tmp_path / "categories.txt"
+    write_csv(csv_file, ["Girokonto:Lebenshaltung", "Girokonto:Sparen"])
+
+    extract_unique_categories(str(csv_file), str(output_file))
+    first_run = read_lines(output_file)
+
+    # A new category appears in the CSV on the second run.
+    write_csv(
+        csv_file, ["Girokonto:Lebenshaltung", "Girokonto:Sparen", "Girokonto:Privat"]
+    )
+    extract_unique_categories(str(csv_file), str(output_file))
+    second_run = read_lines(output_file)
+
+    # Existing entries are merged, not duplicated; list stays sorted and plain.
+    assert second_run == [
+        "Girokonto:Lebenshaltung",
+        "Girokonto:Privat",
+        "Girokonto:Sparen",
+    ]
+    assert first_run == ["Girokonto:Lebenshaltung", "Girokonto:Sparen"]