CSV-Export: bestehende Ausgaben und Originaldateien schützen

.github/workflows/source-platform-smoke.yml

@@ -47,8 +47,11 @@ jobs:
             libxcb-keysyms1 libxcb-randr0 libxcb-render-util0 \
             libxcb-xinerama0 libxcb-xfixes0
 
-      - name: Install PySide6
-        run: pip install PySide6>=6.5.0
+      - name: Install PySide6 and pytest
+        run: python -m pip install "PySide6>=6.5.0" pytest
+
+      - name: Run CSV publication regressions
+        run: python -m pytest -q tests/test_csv_export_transaction.py
 
       - name: Run source platform smoke
         env:

CHANGELOG.md

@@ -23,6 +23,13 @@ Alle wichtigen Änderungen an diesem Projekt werden in dieser Datei dokumentiert
 
 ## [Unreleased]
 
+### Fixed
+- CSV-Exporte aus GUI und CLI werden vollständig in einer eigenen temporären Datei
+  vorbereitet. Format-, Schreib- und Ersetzungsfehler erhalten vorhandene Exporte.
+  Rechnungsoriginale sowie Konfiguration und Datenbank sind einschließlich
+  Datei-Aliassen als Exportziel geschützt; auch ausgeschlossene Rechnungen zählen.
+  UTF-8-BOM, Semikolon, Spalten, Auswahl und Filter bleiben erhalten.
+
 ### CLI Resilience, Amount Normalization & Error Handling Härtung (2026-09-30)
 - **Headless CLI Robustness & Fehlerbehebung (`cli.py`)**:
   - `_format_display_amount()`: Robuste Betragsformatierung für `--list-invoices` implementiert; verhindert unhandled `ValueError: Unknown format code 'f' for object of type 'str'` bei Rechnungsbeträgen im String- oder Währungsformat (`42.50`, `19,99 €`).

UniversalInvoiceMail.py

@@ -4003,7 +4003,7 @@ def setup_ui(self):
         )
         btn_refresh.setToolTip("Rechnungstabelle mit Ordnerinhalt synchronisieren")
         btn_export_csv = QPushButton("CSV Export")
-        btn_export_csv.clicked.connect(self.export_invoices_csv)
+        btn_export_csv.clicked.connect(lambda _checked=False: self.export_invoices_csv())
         btn_export_csv.setObjectName("export_invoices_csv_button")
         btn_export_csv.setAccessibleName("Rechnungsliste als CSV exportieren")
         btn_export_csv.setAccessibleDescription(
@@ -5381,11 +5381,14 @@ def export_invoices_csv(self, filepath: Optional[Union[str, Path]] = None) -> Op
             return None
 
         # Ausgewaehlte Rechnungen ermitteln (falls Checkboxen in Spalte 0 aktiv sind)
+        original_invoices = list(self.invoices)
+        protected_paths = [CONFIG_FILE, INVOICES_DB]
+        protected_paths.extend(inv.path for inv in original_invoices if inv.path)
         selected_paths = self._get_selected_invoice_paths()
         if selected_paths:
-            source_invoices = [inv for inv in self.invoices if inv.path in selected_paths]
+            source_invoices = [inv for inv in original_invoices if inv.path in selected_paths]
         else:
-            source_invoices = list(self.invoices)
+            source_invoices = original_invoices
 
         if not source_invoices:
             if interactive:
@@ -5406,9 +5409,11 @@ def export_invoices_csv(self, filepath: Optional[Union[str, Path]] = None) -> Op
 
         try:
             import csv
+            from csv_export import atomic_csv_output
             target_path = Path(filepath)
-            target_path.parent.mkdir(parents=True, exist_ok=True)
-            with open(target_path, 'w', newline='', encoding='utf-8-sig') as f:
+            protected_paths.extend([CONFIG_FILE, INVOICES_DB])
+            protected_paths.extend(inv.path for inv in self.invoices if inv.path)
+            with atomic_csv_output(target_path, protected_paths) as f:
                 writer = csv.writer(f, delimiter=';')
                 # Header mit erweiterten Feldern
                 writer.writerow([

cli.py

@@ -30,7 +30,9 @@
 import sys
 from datetime import datetime
 from pathlib import Path
-from typing import Any, List, Optional, Sequence, Tuple
+from typing import Any, Iterable, List, Optional, Sequence, Tuple
+
+from csv_export import atomic_csv_output
 
 # Version and identity constants
 APP_NAME = "UniversalInvoiceMail"
@@ -180,19 +182,20 @@ def export_invoices_to_csv(
     output_path: Path,
     profile_filter: Optional[str] = None,
     status_filter: Optional[str] = None,
+    protected_paths: Iterable[Path] = (),
 ) -> int:
     """Exports invoices to a semicolon-separated CSV file with UTF-8 BOM."""
-    output_path.parent.mkdir(parents=True, exist_ok=True)
-
-    filtered = invoices
+    originals = list(invoices)
+    protected = [*protected_paths, *(i.get("path") for i in originals if i.get("path"))]
+    filtered = originals
     if profile_filter:
         p_low = profile_filter.strip().lower()
         filtered = [i for i in filtered if p_low in _normalize_filter_profile(i.get("profile_name"))]
     if status_filter:
         s_low = status_filter.strip().lower()
         filtered = [i for i in filtered if s_low == _normalize_filter_status(i.get("review_status"))]
 
-    with open(output_path, "w", newline="", encoding="utf-8-sig") as f:
+    with atomic_csv_output(output_path, protected) as f:
         writer = csv.writer(f, delimiter=";")
         writer.writerow([
             "Datum", "Profil/Shop", "Absender", "Betreff",
@@ -447,15 +450,17 @@ def run_cli(argv: Optional[Sequence[str]] = None) -> int:
         target = args.export_csv
         if target == "DEFAULT":
             target = str(DEFAULT_BASE_DIR / f"rechnungen_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv")
-        out_path = Path(target).resolve()
+        out_path = Path(target)
         try:
+            out_path = out_path.resolve()
             count = export_invoices_to_csv(
                 invoices=invoices,
                 output_path=out_path,
                 profile_filter=args.filter_profile,
                 status_filter=args.filter_status,
+                protected_paths=get_default_paths(args.config_path, args.invoices_db_path),
             )
-        except OSError as e:
+        except (OSError, ValueError, TypeError, RuntimeError, OverflowError, csv.Error) as e:
             if args.json_output:
                 print(json.dumps({
                     "status": "error",

csv_export.py

@@ -0,0 +1,51 @@
+"""Private staging for CSV exports, without GUI or mail dependencies."""
+
+from contextlib import contextmanager
+import logging
+import os
+from pathlib import Path
+import tempfile
+
+
+def _check_target(target, protected_paths):
+    resolved = target.resolve()
+    for source in protected_paths:
+        if resolved == source.resolve():
+            raise ValueError("CSV-Ziel ist eine geschützte Quelldatei. Bitte einen anderen Pfad wählen.")
+        try:
+            same = target.samefile(source)
+        except FileNotFoundError:
+            same = False
+        if same:
+            raise ValueError("CSV-Ziel ist eine geschützte Quelldatei. Bitte einen anderen Pfad wählen.")
+
+
+@contextmanager
+def atomic_csv_output(output_path, protected_paths=()):
+    """Publish only a complete UTF-8-BOM CSV; retain old output on failure.
+
+    Identity checks protect known originals and state files, including aliases.
+    They do not provide a transaction against concurrent external path changes.
+    """
+    target = Path(output_path)
+    protected = tuple(Path(path) for path in protected_paths if path)
+    _check_target(target, protected)
+    target.parent.mkdir(parents=True, exist_ok=True)
+    temporary = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w", encoding="utf-8-sig", newline="", delete=False,
+            dir=target.parent, prefix=".uim-csv-", suffix=".tmp",
+        ) as stream:
+            temporary = Path(stream.name)
+            yield stream
+            stream.flush()
+            os.fsync(stream.fileno())
+        _check_target(target, protected)
+        os.replace(temporary, target)
+    finally:
+        if temporary is not None:
+            try:
+                temporary.unlink(missing_ok=True)
+            except OSError:
+                logging.warning("Eigene temporäre CSV-Datei konnte nicht entfernt werden: %s", temporary)

tests/test_csv_export.py

@@ -38,6 +38,32 @@ def qapp():
     return app
 
 
+def test_export_button_opens_dialog_without_boolean_filepath(tmp_path, monkeypatch, qapp):
+    """A real Qt clicked(bool) must retain the interactive save dialog."""
+    from PySide6.QtCore import QCoreApplication, QEvent
+    from PySide6.QtWidgets import QPushButton
+
+    monkeypatch.setattr(uim, "CONFIG_FILE", tmp_path / "config.json")
+    monkeypatch.setattr(uim, "INVOICES_DB", tmp_path / "invoices.json")
+    window = uim.MainWindow()
+    target = tmp_path / "clicked.csv"
+    calls = []
+    monkeypatch.setattr(uim.QFileDialog, "getSaveFileName", lambda *args, **kwargs: (calls.append("dialog") or (str(target), "CSV")))
+    monkeypatch.setattr(uim.QMessageBox, "information", lambda *args: calls.append("success"))
+    monkeypatch.setattr(uim.QMessageBox, "warning", lambda *args: calls.append("error"))
+    monkeypatch.setattr(window, "_get_selected_invoice_paths", lambda: set())
+    try:
+        window.invoices = [uim.Invoice(id="clicked", profile_name="Prüfung", filename="dummy.pdf", date="2026-10-01", path=str(tmp_path / "dummy.pdf"), amount=12.5)]
+        button = window.findChild(QPushButton, "export_invoices_csv_button")
+        assert button is not None and button.isEnabled()
+        button.click()
+        assert calls == ["dialog", "success"]
+        assert target.read_bytes().startswith(b"\xef\xbb\xbf")
+    finally:
+        window.deleteLater()
+        QCoreApplication.sendPostedEvents(window, QEvent.Type.DeferredDelete)
+
+
 def test_export_invoices_csv_headless_all(tmp_path: Path, monkeypatch, qapp):
     """Test exporting all invoices programmatically with comprehensive columns."""
     invoices_db = tmp_path / "invoices.json"

tests/test_csv_export_transaction.py

@@ -0,0 +1,153 @@
+"""Exports retain old data and never replace source/state files."""
+
+import ast
+import csv
+from datetime import datetime
+import json
+import os
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+import cli
+import csv_export
+
+
+@pytest.fixture
+def gui_export(tmp_path):
+    # Execute the exact GUI method without constructing a window or touching settings.
+    source = Path(__file__).parents[1] / "UniversalInvoiceMail.py"
+    tree = ast.parse(source.read_text(encoding="utf-8-sig"))
+    main = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == "MainWindow")
+    method = next(n for n in main.body if isinstance(n, ast.FunctionDef) and n.name == "export_invoices_csv")
+    module = ast.Module(body=[ast.ImportFrom(module="__future__", names=[ast.alias(name="annotations")], level=0), method], type_ignores=[])
+    namespace = {"Path": Path, "datetime": datetime, "CONFIG_FILE": tmp_path / "config.json", "INVOICES_DB": tmp_path / "invoices.json"}
+    exec(compile(ast.fix_missing_locations(module), str(source), "exec"), namespace)
+    return namespace["export_invoices_csv"]
+
+
+def invoice(tmp_path, **changes):
+    values = dict(date="2026-10-01", profile_name="Äpfel", sender="test@example.invalid", subject="Test", filename="original.pdf", path=str(tmp_path / "original.pdf"), amount=12.5, currency="EUR", review_status="unchecked", notes="Größe; geprüft")
+    values.update(changes)
+    return values
+
+
+@pytest.mark.parametrize("failure", ["encoding", "body", "fsync", "replace"])
+def test_old_export_survives_failure(tmp_path, monkeypatch, failure):
+    target = tmp_path / "old.csv"
+    target.write_bytes(b"previous export")
+    foreign = tmp_path / ".uim-csv-foreign.tmp"
+    foreign.write_bytes(b"foreign")
+    if failure in ("fsync", "replace"):
+        def fail(*args):
+            raise OSError("injected failure")
+        monkeypatch.setattr(csv_export.os, failure, fail)
+    with pytest.raises((OSError, ValueError, UnicodeError)):
+        with csv_export.atomic_csv_output(target) as stream:
+            stream.write("\ud800" if failure == "encoding" else "complete output")
+            if failure == "body":
+                raise ValueError("invalid row")
+    assert target.read_bytes() == b"previous export"
+    assert foreign.read_bytes() == b"foreign"
+    assert list(tmp_path.glob(".uim-csv-*.tmp")) == [foreign]
+
+
+def test_complete_csv_preserves_format(tmp_path):
+    target = tmp_path / "nested" / "output.csv"
+    count = cli.export_invoices_to_csv([invoice(tmp_path)], target)
+    raw = target.read_bytes()
+    assert count == 1 and raw.startswith(b"\xef\xbb\xbf")
+    assert raw.count(b"\r\n") == 2
+    rows = list(csv.reader(raw.decode("utf-8-sig").splitlines(), delimiter=";"))
+    assert rows[1][1] == "Äpfel" and rows[1][6] == "12,50" and rows[1][9] == "Größe; geprüft"
+    assert not list(target.parent.glob(".uim-csv-*.tmp"))
+
+
+@pytest.mark.parametrize("alias", ["direct", "hardlink"])
+def test_filtered_originals_are_protected(tmp_path, alias):
+    original = tmp_path / "excluded.pdf"
+    original.write_bytes(b"source")
+    target = original
+    if alias == "hardlink":
+        target = tmp_path / "alias.csv"
+        os.link(original, target)
+    invoices = [invoice(tmp_path), invoice(tmp_path, path=str(original), profile_name="excluded")]
+    with pytest.raises(ValueError, match="geschützte"):
+        cli.export_invoices_to_csv(invoices, target, profile_filter="Äpfel")
+    assert original.read_bytes() == target.read_bytes() == b"source"
+    assert not list(tmp_path.glob(".uim-csv-*.tmp"))
+
+
+def test_new_alias_before_publication_is_protected(tmp_path):
+    source = tmp_path / "source.pdf"
+    source.write_bytes(b"source")
+    target = tmp_path / "out.csv"
+    with pytest.raises(ValueError, match="geschützte"):
+        with csv_export.atomic_csv_output(target, [source]) as stream:
+            stream.write("new output")
+            os.link(source, target)
+    assert source.read_bytes() == target.read_bytes() == b"source"
+    assert not list(tmp_path.glob(".uim-csv-*.tmp"))
+
+
+@pytest.mark.parametrize("kind", ["config", "database"])
+@pytest.mark.parametrize("alias", [False, True])
+def test_cli_actual_state_paths_protected(tmp_path, capsys, kind, alias):
+    config = tmp_path / "custom-config.json"
+    database = tmp_path / "custom-db.json"
+    config.write_text("{}", encoding="utf-8")
+    database.write_text(json.dumps([invoice(tmp_path)]), encoding="utf-8")
+    source = config if kind == "config" else database
+    before = source.read_bytes()
+    target = source
+    if alias:
+        target = tmp_path / "alias.csv"
+        os.link(source, target)
+    rc = cli.run_cli(["--config", str(config), "--invoices-db", str(database), "--export-csv", str(target), "--json"])
+    assert rc == 1 and json.loads(capsys.readouterr().out)["status"] == "error"
+    assert source.read_bytes() == target.read_bytes() == before
+
+
+@pytest.mark.parametrize("changes", [{"notes": "\ud800"}, {"amount": 10 ** 400}])
+def test_cli_format_error_is_structured_and_preserves_export(tmp_path, capsys, changes):
+    database = tmp_path / "db.json"
+    database.write_text(json.dumps([invoice(tmp_path, **changes)]), encoding="utf-8")
+    target = tmp_path / "old.csv"
+    target.write_bytes(b"old")
+    assert cli.run_cli(["--config", str(tmp_path / "absent.json"), "--invoices-db", str(database), "--export-csv", str(target), "--json"]) == 1
+    assert json.loads(capsys.readouterr().out)["status"] == "error"
+    assert target.read_bytes() == b"old"
+
+
+def test_gui_late_amount_error_preserves_old_export(tmp_path, gui_export):
+    target = tmp_path / "old.csv"
+    target.write_bytes(b"old")
+    logs = []
+    fake = SimpleNamespace(invoices=[SimpleNamespace(**invoice(tmp_path, amount="invalid"))], _get_selected_invoice_paths=lambda: set(), _log=logs.append)
+    assert gui_export(fake, target) is None
+    assert target.read_bytes() == b"old" and "[ERROR]" in logs[-1]
+
+
+@pytest.mark.parametrize("kind", ["excluded", "config", "database"])
+def test_gui_protects_unselected_and_state_files(tmp_path, gui_export, kind):
+    first = SimpleNamespace(**invoice(tmp_path))
+    excluded = SimpleNamespace(**invoice(tmp_path, path=str(tmp_path / "excluded.pdf")))
+    target = Path(excluded.path) if kind == "excluded" else tmp_path / ("config.json" if kind == "config" else "invoices.json")
+    target.write_bytes(b"protected")
+    fake = SimpleNamespace(invoices=[first, excluded], _get_selected_invoice_paths=lambda: {first.path}, _log=lambda text: None)
+    assert gui_export(fake, target) is None
+    assert target.read_bytes() == b"protected"
+
+
+def test_gui_original_removed_during_dialog_stays_protected(tmp_path, gui_export):
+    target = tmp_path / "source.pdf"
+    target.write_bytes(b"original")
+    fake = SimpleNamespace(invoices=[SimpleNamespace(**invoice(tmp_path, path=str(target)))], settings=SimpleNamespace(download_path=str(tmp_path)), _get_selected_invoice_paths=lambda: set(), _log=lambda text: None)
+    def choose(*args):
+        fake.invoices = []
+        return str(target), "CSV"
+    gui_export.__globals__["QFileDialog"] = SimpleNamespace(getSaveFileName=choose)
+    gui_export.__globals__["QMessageBox"] = SimpleNamespace(information=lambda *args: None, warning=lambda *args: None)
+    assert gui_export(fake) is None
+    assert target.read_bytes() == b"original"