# Invoice Backend Phase 1 (Multi-Tax, IT-Parser, Skonto/Rabatt, Zahlungsmittel) Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Extend the existing XRechnung/ZUGFeRD converter (`main.py`) with four EN16931-compliant backend features — mixed tax rates per line, IT-Systemhaus-specific regex parsing, document-level discount + Skonto payment terms, and a dynamic payment means code — without breaking the currently working extraction, calculation, or XML generation logic.
**Architecture:** All pure business logic (constants, parsing, Decimal math, UBL/CII XML generation) moves out of `main.py` into a new dependency-free module `invoice_logic.py`, so it can be unit-tested with `pytest` outside of a Streamlit runtime. `main.py` keeps only the Streamlit UI and imports from `invoice_logic.py`. Each of the four features is implemented as pure functions in `invoice_logic.py` with dedicated tests, then wired into the two XML generators (respecting the exact UBL/CII element ordering already documented in the existing code comments) and into `main.py` with the minimal UI controls needed to reach the new fields — the polished UI redesign is explicitly Phase 2.
**Tech Stack:** Python 3.11, Streamlit 1.58.0, pydantic 2.13.4, pytest (new, dev-only dependency) for the pure-logic test suite.
## Global Constraints
- Do not change the numeric behavior of any existing calculation for a single-tax-rate invoice with no discount/Skonto — outputs must be byte-identical to the current generator for that case (verified by a golden-output test in Task 1).
- Keep German field labels and BT-/BG-code references in all new UI text and code comments, matching the existing convention in `main.py`.
- Keep the existing "no comments except non-obvious WHY" style; new XML-sequence-order comments follow the same pattern already used throughout `generate_compliant_xrechnung`/`generate_compliant_zugferd_cii`.
- `invoice_logic.py` must have zero `import streamlit` — it must be importable and testable without a Streamlit runtime.
- `pytest` and its version pin go in a new `requirements-dev.txt`, not in `requirements.txt` (keeps the production Docker image unchanged).
- Every new/changed XML element must respect the UBL/CII schema element ordering already documented inline in the current generator functions (see Task 3–4 for the exact insertion points).
- Document-level discount (BT-21) must be **fully EN16931-compliant across mixed tax rates**: when a single discount applies to an invoice containing multiple VAT rates, emit one `AllowanceCharge` / `SpecifiedTradeAllowanceCharge` block per VAT rate, each carrying the proportionally allocated discount amount and that rate's tax category — never a single combined block on one rate.
---
## File Structure
| File | Responsibility |
|---|---|
| `invoice_logic.py` (new) | All pure logic: `VERKAEUFER`/`VERKAEUFER_IBAN` constants, `PAYMENT_MEANS_CODES`, date/rounding helpers, regex-based item/period/contract-type/serial-number extraction, `aggregate_tax_subtotals`, `compute_invoice_totals`, `format_skonto_text`, `generate_compliant_xrechnung`, `generate_compliant_zugferd_cii`, `add_pdfa_output_intent`, `fix_xrechnung_xmp_conformance`, `InvoiceMetadata` pydantic model. |
| `main.py` (modified) | Streamlit UI only: `LLMClient`, `get_cached_llm`, page layout, `st.session_state` handling. Imports everything else from `invoice_logic`. |
| `tests/test_invoice_logic.py` (new) | Unit + integration tests for every function in `invoice_logic.py`. |
| `requirements-dev.txt` (new) | `pytest` pin for local/dev test runs. |
| `pytest.ini` (new) | `pythonpath = .` so tests can `import invoice_logic` from the `tests/` subfolder. |
---
### Task 1: Extract pure logic into `invoice_logic.py` (no behavior change)
**Files:**
- Create: `invoice_logic.py`
- Modify: `main.py:1-14` (imports), `main.py:19-682` (remove moved definitions, add import)
- Create: `requirements-dev.txt`
- Create: `pytest.ini`
- Create: `tests/test_invoice_logic.py`
**Interfaces:**
- Produces: `invoice_logic.VERKAEUFER: dict`, `invoice_logic.VERKAEUFER_IBAN: str`, `invoice_logic.InvoiceMetadata` (pydantic `BaseModel`), `invoice_logic.berechne_faelligkeit(invoice_date_str: str, term_days) -> str`, `invoice_logic.kaufmaennisches_runden(betrag, nachkommastellen=2) -> float`, `invoice_logic.clean_german_float(val_str) -> float`, `invoice_logic.extract_invoice_items_pure_python(text: str) -> list[dict]`, `invoice_logic.map_unece_unit_code(unit_str: str) -> str`, `invoice_logic.parse_to_strict_xml_date(date_str) -> str`, `invoice_logic.format_cii_date(date_str) -> str`, `invoice_logic.generate_compliant_xrechnung(data: dict) -> str`, `invoice_logic.generate_compliant_zugferd_cii(data: dict) -> str`, `invoice_logic.add_pdfa_output_intent(pdf_bytes, icc_path=SRGB_ICC_PATH) -> bytes`, `invoice_logic.fix_xrechnung_xmp_conformance(pdf_bytes) -> bytes`.
- [ ] **Step 1: Create `requirements-dev.txt`**
```
pytest==8.3.4
```
- [ ] **Step 2: Create `pytest.ini`**
```ini
[pytest]
pythonpath = .
```
- [ ] **Step 3: Create `invoice_logic.py` with all pure logic moved verbatim from `main.py`**
Move (cut, don't rewrite) these exact blocks from the current `main.py` into a new file `invoice_logic.py`, in this order, with these imports at the top:
```python
import os
import io
import json
import re
import xml.etree.ElementTree as ET
from xml.dom import minidom
from datetime import datetime, timedelta
from decimal import Decimal, ROUND_HALF_UP
import pikepdf
from pydantic import BaseModel, Field
```
Then, unchanged, in this order:
1. The `InvoiceMetadata` class (currently `main.py:68-87`)
2. `VERKAEUFER_IBAN` and `VERKAEUFER` (currently `main.py:95-111`)
3. `berechne_faelligkeit` (currently `main.py:114-131`)
4. `kaufmaennisches_runden`, `clean_german_float` (currently `main.py:134-154`)
5. `extract_invoice_items_pure_python` (currently `main.py:156-206`)
6. `map_unece_unit_code`, `parse_to_strict_xml_date`, `format_cii_date` (currently `main.py:208-228`)
7. `generate_compliant_xrechnung` (currently `main.py:234-399`)
8. `generate_compliant_zugferd_cii` (currently `main.py:405-600`)
9. `SRGB_ICC_PATH`, `add_pdfa_output_intent`, `fix_xrechnung_xmp_conformance` (currently `main.py:606-667`)
Do not change a single line of logic in this step — pure cut-and-paste.
- [ ] **Step 4: Update `main.py` to import from `invoice_logic` instead of defining these locally**
Replace `main.py:1-14`:
```python
import os
import io
import streamlit as st
import json
import re
import xml.etree.ElementTree as ET
from xml.dom import minidom
from datetime import datetime, timedelta
from decimal import Decimal, ROUND_HALF_UP
import fitz
import requests
from pydantic import BaseModel, Field
import pikepdf
from facturx import generate_from_binary as facturx_generate_from_binary
```
with:
```python
import os
import io
import streamlit as st
import json
import fitz
import requests
from facturx import generate_from_binary as facturx_generate_from_binary
from invoice_logic import (
InvoiceMetadata,
VERKAEUFER,
VERKAEUFER_IBAN,
berechne_faelligkeit,
kaufmaennisches_runden,
clean_german_float,
extract_invoice_items_pure_python,
map_unece_unit_code,
parse_to_strict_xml_date,
format_cii_date,
generate_compliant_xrechnung,
generate_compliant_zugferd_cii,
add_pdfa_output_intent,
fix_xrechnung_xmp_conformance,
)
```
Then delete the moved definitions from the remainder of `main.py` (the class `InvoiceMetadata`, the `VERKAEUFER`/`VERKAEUFER_IBAN` constants, and every function listed in Step 3) — everything from the original `main.py:68` through `main.py:667` that isn't the `LLMClient` class or the `# 4. API CLIENT INITIALISIERUNG` section stays removed. The `LLMClient` class (`main.py:25-62`) and `get_cached_llm` (`main.py:673-678`) remain in `main.py` unchanged.
- [ ] **Step 5: Write the golden-output regression test**
Create `tests/test_invoice_logic.py`:
```python
from decimal import Decimal
import xml.etree.ElementTree as ET
from invoice_logic import (
berechne_faelligkeit,
kaufmaennisches_runden,
clean_german_float,
extract_invoice_items_pure_python,
map_unece_unit_code,
parse_to_strict_xml_date,
generate_compliant_xrechnung,
generate_compliant_zugferd_cii,
VERKAEUFER_IBAN,
)
def _single_rate_invoice():
"""Eine einfache Einzelsteuersatz-Rechnung, wie main.py sie vor Phase 1 erzeugt hat."""
items = [
{"name": "Beratung", "description": "", "quantity": 2.0, "unit": "Std.",
"unit_price": 100.0, "total_amount": 200.0},
]
return {
"invoice_type_code": "380",
"invoice_number": "R26-001",
"invoice_date": "2026-01-15",
"delivery_date": "",
"buyer_reference": "",
"customer_number": "",
"recipient_name": "Test GmbH",
"recipient_email": "kunde@example.com",
"recipient_street": "Musterweg 1",
"recipient_zip": "12345",
"recipient_city": "Berlin",
"items": items,
"total_net_amount": 200.0,
"total_vat_amount": 38.0,
"vat_rate_percent": 19.0,
"total_gross_amount": 238.0,
"iban": VERKAEUFER_IBAN,
"delivery_note_id": "",
"order_id": "",
"contract_id": "",
"service_start_date": "",
"service_end_date": "",
"payment_term_days": "14",
"invoice_note": "",
}
def test_generate_compliant_xrechnung_single_rate_unchanged():
xml_str = generate_compliant_xrechnung(_single_rate_invoice())
root = ET.fromstring(xml_str.encode("utf-8"))
ns = {"cbc": "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2",
"cac": "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"}
assert root.find("cbc:ID", ns).text == "R26-001"
tax_amount = root.find(".//cac:TaxTotal/cbc:TaxAmount", ns).text
assert tax_amount == "38.00"
payable = root.find(".//cac:LegalMonetaryTotal/cbc:PayableAmount", ns).text
assert payable == "238.00"
def test_generate_compliant_zugferd_cii_single_rate_unchanged():
xml_str = generate_compliant_zugferd_cii(_single_rate_invoice())
root = ET.fromstring(xml_str.encode("utf-8"))
ns = {"ram": "urn:un:unece:uncefact:data:standard:ReusableAggregateBusinessInformationEntity:100"}
grand_total = root.find(".//ram:SpecifiedTradeSettlementHeaderMonetarySummation/ram:GrandTotalAmount", ns).text
assert grand_total == "238.00"
def test_berechne_faelligkeit_basic():
assert berechne_faelligkeit("2026-01-15", "14") == "2026-01-29"
def test_kaufmaennisches_runden_half_up():
assert kaufmaennisches_runden(1.005) == 1.01
def test_clean_german_float_thousands_and_comma():
assert clean_german_float("1.234,56") == 1234.56
def test_extract_invoice_items_pure_python_basic():
text = "1\nBeratung\n2,00 Std.\n100,00\n200,00\n"
items = extract_invoice_items_pure_python(text)
assert items[0]["name"] == "Beratung"
assert items[0]["quantity"] == 2.0
assert items[0]["unit_price"] == 100.0
def test_map_unece_unit_code_hours():
assert map_unece_unit_code("Std.") == "HUR"
def test_parse_to_strict_xml_date_german_format():
assert parse_to_strict_xml_date("15.01.2026") == "2026-01-15"
```
- [ ] **Step 4: Install dev dependencies and run the tests**
Run: `pip install -r requirements-dev.txt`
Run: `pytest tests/test_invoice_logic.py -v`
Expected: 7 tests PASS.
- [ ] **Step 5: Verify the Streamlit app still boots**
Run: `python -c "import ast; ast.parse(open('main.py', encoding='utf-8').read()); print('OK')"`
Expected: `OK`
Run: `python -c "import invoice_logic; print('OK')"`
Expected: `OK` (proves `invoice_logic.py` has no accidental Streamlit dependency)
- [ ] **Step 6: Commit**
```bash
git add invoice_logic.py main.py requirements-dev.txt pytest.ini tests/test_invoice_logic.py
git commit -m "refactor: extract pure invoice logic into invoice_logic.py for testability"
```
---
### Task 2: Dynamic payment means code (BT-81)
**Files:**
- Modify: `invoice_logic.py` (add `PAYMENT_MEANS_CODES`, update `generate_compliant_xrechnung`, `generate_compliant_zugferd_cii`)
- Modify: `main.py` (add selectbox, wire into `final_invoice_dict`)
- Modify: `tests/test_invoice_logic.py` (add tests)
**Interfaces:**
- Consumes: `generate_compliant_xrechnung(data)`, `generate_compliant_zugferd_cii(data)` from Task 1.
- Produces: `invoice_logic.PAYMENT_MEANS_CODES: dict[str, str]`. Both generators now read `data.get("payment_means_code", "58")` instead of the hardcoded `"58"`.
- [ ] **Step 1: Write the failing tests**
Add to `tests/test_invoice_logic.py`:
```python
from invoice_logic import PAYMENT_MEANS_CODES, generate_compliant_xrechnung, generate_compliant_zugferd_cii
def test_payment_means_codes_contains_expected_entries():
assert PAYMENT_MEANS_CODES["58"] == "SEPA-Überweisung"
assert set(["30", "31", "42", "48", "58"]).issubset(PAYMENT_MEANS_CODES.keys())
def test_generate_compliant_xrechnung_uses_dynamic_payment_means_code():
data = _single_rate_invoice()
data["payment_means_code"] = "31"
xml_str = generate_compliant_xrechnung(data)
root = ET.fromstring(xml_str.encode("utf-8"))
ns = {"cbc": "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2",
"cac": "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"}
code = root.find(".//cac:PaymentMeans/cbc:PaymentMeansCode", ns).text
assert code == "31"
def test_generate_compliant_xrechnung_defaults_payment_means_code_to_58():
xml_str = generate_compliant_xrechnung(_single_rate_invoice())
root = ET.fromstring(xml_str.encode("utf-8"))
ns = {"cbc": "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2",
"cac": "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"}
code = root.find(".//cac:PaymentMeans/cbc:PaymentMeansCode", ns).text
assert code == "58"
def test_generate_compliant_zugferd_cii_uses_dynamic_payment_means_code():
data = _single_rate_invoice()
data["payment_means_code"] = "48"
xml_str = generate_compliant_zugferd_cii(data)
root = ET.fromstring(xml_str.encode("utf-8"))
ns = {"ram": "urn:un:unece:uncefact:data:standard:ReusableAggregateBusinessInformationEntity:100"}
code = root.find(".//ram:SpecifiedTradeSettlementPaymentMeans/ram:TypeCode", ns).text
assert code == "48"
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `pytest tests/test_invoice_logic.py -k payment_means -v`
Expected: FAIL (`KeyError: 'PAYMENT_MEANS_CODES'` / import error, since it doesn't exist yet).
- [ ] **Step 3: Add `PAYMENT_MEANS_CODES` to `invoice_logic.py`**
Add directly below the `VERKAEUFER` dict definition:
```python
# UN/EDIFACT 4461 Zahlungsmittelcodes (BT-81), Auswahl der fuer B2B/B2G relevanten Codes.
# "58" bleibt der Standardwert (SEPA-Ueberweisung), da alle bisherigen Rechnungen diesen nutzen.
PAYMENT_MEANS_CODES = {
"58": "SEPA-Überweisung",
"30": "Überweisung (allgemein)",
"31": "SEPA-Lastschrift",
"42": "Einzahlung auf Bankkonto",
"48": "Bankkarte (Kredit-/Debitkarte)",
}
```
- [ ] **Step 4: Make the payment means code dynamic in `generate_compliant_xrechnung`**
In `invoice_logic.py`, find:
```python
payment_means = ET.SubElement(root, f"{{{CAC}}}PaymentMeans")
# BT-81 Zahlungsart: Code 58 = SEPA-Überweisung (statt 30, das für Überweisungen
# außerhalb des SEPA-Raums gedacht ist - VD-Valitool-169-Hinweis).
ET.SubElement(payment_means, f"{{{CBC}}}PaymentMeansCode").text = "58"
```
Replace with:
```python
payment_means = ET.SubElement(root, f"{{{CAC}}}PaymentMeans")
# BT-81 Zahlungsart: Standard bleibt 58 (SEPA-Ueberweisung), aber ueber
# PAYMENT_MEANS_CODES pro Rechnung waehlbar (siehe UN/EDIFACT 4461).
ET.SubElement(payment_means, f"{{{CBC}}}PaymentMeansCode").text = str(data.get("payment_means_code", "58"))
```
- [ ] **Step 5: Make the payment means code dynamic in `generate_compliant_zugferd_cii`**
In `invoice_logic.py`, find:
```python
pay_means = ET.SubElement(header_settle, f"{{{RAM}}}SpecifiedTradeSettlementPaymentMeans")
# BT-81 Zahlungsart: Code 58 = SEPA-Überweisung. Code 30 ist für Überweisungen AUSSERHALB
# des SEPA-Raums gedacht (VD-Valitool-169-Hinweis) - da hier immer eine IBAN verwendet wird,
# ist 58 der korrekte Code.
ET.SubElement(pay_means, f"{{{RAM}}}TypeCode").text = "58"
```
Replace with:
```python
pay_means = ET.SubElement(header_settle, f"{{{RAM}}}SpecifiedTradeSettlementPaymentMeans")
# BT-81 Zahlungsart: Standard bleibt 58 (SEPA-Ueberweisung), aber ueber
# PAYMENT_MEANS_CODES pro Rechnung waehlbar (siehe UN/EDIFACT 4461).
ET.SubElement(pay_means, f"{{{RAM}}}TypeCode").text = str(data.get("payment_means_code", "58"))
```
- [ ] **Step 6: Run tests to verify they pass**
Run: `pytest tests/test_invoice_logic.py -k payment_means -v`
Expected: 4 tests PASS.
- [ ] **Step 7: Wire the selectbox into `main.py`**
In `main.py`, find the block starting with `st.markdown("**Absender (Verkäufer):**")`'s sibling column (`c_r`, the "Empfänger" column), and immediately after the `st.markdown("---")` that follows the address columns block (right before `st.markdown("### 3. Positionen")`), insert:
```python
payment_means_codes_list = list(PAYMENT_MEANS_CODES.keys())
payment_means_labels = [f"{code} - {PAYMENT_MEANS_CODES[code]}" for code in payment_means_codes_list]
selected_payment_means_label = st.selectbox(
"Zahlungsmittel (BT-81)",
options=payment_means_labels,
index=payment_means_codes_list.index("58"),
)
selected_payment_means_code = selected_payment_means_label.split(" - ")[0]
```
Add `PAYMENT_MEANS_CODES` to the `from invoice_logic import (...)` block in `main.py` (Task 1, Step 4).
- [ ] **Step 8: Add the field to `final_invoice_dict` in `main.py`**
Find:
```python
"vat_rate_percent": vat_rate, "total_gross_amount": computed_gross_amount, "iban": VERKAEUFER_IBAN,
```
Replace with:
```python
"vat_rate_percent": vat_rate, "total_gross_amount": computed_gross_amount, "iban": VERKAEUFER_IBAN,
"payment_means_code": selected_payment_means_code,
```
- [ ] **Step 9: Manual smoke test**
Run: `streamlit run main.py`
Upload any sample invoice PDF, extract data, change "Zahlungsmittel (BT-81)" to `31 - SEPA-Lastschrift`, export XRechnung, open the downloaded XML and confirm `31`.
- [ ] **Step 10: Commit**
```bash
git add invoice_logic.py main.py tests/test_invoice_logic.py
git commit -m "feat: make payment means code (BT-81) dynamically selectable"
```
---
### Task 3: Mixed tax rates per line (Multi-Tax-Support)
**Files:**
- Modify: `invoice_logic.py` (add `aggregate_tax_subtotals`, `compute_invoice_totals`; update both generators to use per-line rate + multiple tax subtotals)
- Modify: `main.py` (per-item VAT input, totals computed via `compute_invoice_totals`)
- Modify: `tests/test_invoice_logic.py`
**Interfaces:**
- Consumes: `Decimal`, `ROUND_HALF_UP` (already imported in `invoice_logic.py` from Task 1).
- Produces: `invoice_logic.aggregate_tax_subtotals(items: list[dict]) -> list[dict]` where each bucket is `{"rate": float, "category": str, "taxable_amount": float, "tax_amount": float}`. `invoice_logic.compute_invoice_totals(items: list[dict], discount_percent: float = 0.0, discount_amount: float = 0.0) -> dict` returning `{"line_net_total": float, "allowance_amount": float, "tax_basis_total": float, "total_vat_amount": float, "total_gross_amount": float, "tax_buckets": list[dict]}`. These are consumed by Task 4 (discount wiring) unchanged.
- `final_invoice_dict` in `main.py` gains keys `tax_buckets`, `line_net_total`, `allowance_amount` (Task 3 always passes `0.0` for the latter; Task 4 wires it from UI). `total_net_amount` now means "tax basis after allowance" (identical to before when there's no discount).
- [ ] **Step 1: Write the failing tests for `aggregate_tax_subtotals` and `compute_invoice_totals`**
Add to `tests/test_invoice_logic.py`:
```python
from invoice_logic import aggregate_tax_subtotals, compute_invoice_totals
def test_aggregate_tax_subtotals_single_rate():
items = [{"total_amount": 200.0, "vat_rate_percent": 19.0}]
buckets = aggregate_tax_subtotals(items)
assert buckets == [{"rate": 19.0, "category": "S", "taxable_amount": 200.0, "tax_amount": 38.0}]
def test_aggregate_tax_subtotals_mixed_rates():
items = [
{"total_amount": 100.0, "vat_rate_percent": 19.0},
{"total_amount": 50.0, "vat_rate_percent": 7.0},
{"total_amount": 30.0, "vat_rate_percent": 0.0},
]
buckets = aggregate_tax_subtotals(items)
assert buckets == [
{"rate": 0.0, "category": "E", "taxable_amount": 30.0, "tax_amount": 0.0},
{"rate": 7.0, "category": "S", "taxable_amount": 50.0, "tax_amount": 3.5},
{"rate": 19.0, "category": "S", "taxable_amount": 100.0, "tax_amount": 19.0},
]
def test_compute_invoice_totals_mixed_rates_no_discount():
items = [
{"total_amount": 100.0, "vat_rate_percent": 19.0},
{"total_amount": 50.0, "vat_rate_percent": 7.0},
]
totals = compute_invoice_totals(items)
assert totals["line_net_total"] == 150.0
assert totals["allowance_amount"] == 0.0
assert totals["tax_basis_total"] == 150.0
assert totals["total_vat_amount"] == 22.5
assert totals["total_gross_amount"] == 172.5
def test_compute_invoice_totals_single_rate_matches_old_behavior():
items = [{"total_amount": 200.0, "vat_rate_percent": 19.0}]
totals = compute_invoice_totals(items)
assert totals["tax_basis_total"] == 200.0
assert totals["total_vat_amount"] == 38.0
assert totals["total_gross_amount"] == 238.0
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `pytest tests/test_invoice_logic.py -k "aggregate_tax_subtotals or compute_invoice_totals" -v`
Expected: FAIL with `ImportError: cannot import name 'aggregate_tax_subtotals'`.
- [ ] **Step 3: Implement `aggregate_tax_subtotals` and `compute_invoice_totals` in `invoice_logic.py`**
Add after `kaufmaennisches_runden`/`clean_german_float`:
```python
def aggregate_tax_subtotals(items):
"""Gruppiert Rechnungspositionen nach Steuersatz (BG-23) und liefert je Gruppe Netto- und
Steuerbetrag - Grundlage fuer mehrere cac:TaxSubtotal/ram:ApplicableTradeTax-Bloecke bei
gemischten Steuersaetzen auf einer Rechnung."""
buckets = {}
for item in items:
rate = Decimal(str(item.get("vat_rate_percent", 0.0)))
net = Decimal(str(item["total_amount"]))
buckets[rate] = buckets.get(rate, Decimal("0")) + net
result = []
for rate in sorted(buckets.keys()):
taxable = buckets[rate].quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
tax_amount = (taxable * rate / Decimal("100")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
category = "S" if rate > 0 else "E"
result.append({
"rate": float(rate),
"category": category,
"taxable_amount": float(taxable),
"tax_amount": float(tax_amount),
})
return result
def compute_invoice_totals(items, discount_percent=0.0, discount_amount=0.0):
"""Berechnet Netto-/Steuer-/Bruttosummen inkl. optionalem dokumentenweiten Rabatt (BT-21).
Bei gemischten Steuersaetzen wird der Rabatt anteilig ueber die Steuersatz-Buckets verteilt
(Standardverfahren, da EN16931 keinen rabattspezifischen Steuersatz kennt). Ohne discount_*
ist das Ergebnis identisch zur reinen Summenbildung von vor Phase 1."""
line_net_total = sum((Decimal(str(item["total_amount"])) for item in items), Decimal("0"))
line_net_total = line_net_total.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
if discount_amount:
allowance = Decimal(str(discount_amount)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
elif discount_percent:
allowance = (line_net_total * Decimal(str(discount_percent)) / Decimal("100")).quantize(
Decimal("0.01"), rounding=ROUND_HALF_UP)
else:
allowance = Decimal("0.00")
raw_buckets = aggregate_tax_subtotals(items)
if allowance > 0 and line_net_total > 0:
buckets = []
allocated = Decimal("0")
for i, bucket in enumerate(raw_buckets):
taxable = Decimal(str(bucket["taxable_amount"]))
rate = Decimal(str(bucket["rate"]))
if i == len(raw_buckets) - 1:
share = allowance - allocated
else:
share = (allowance * taxable / line_net_total).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
allocated += share
new_taxable = (taxable - share).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
tax_amount = (new_taxable * rate / Decimal("100")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
buckets.append({
"rate": bucket["rate"],
"category": bucket["category"],
"taxable_amount": float(new_taxable),
"tax_amount": float(tax_amount),
"allowance_amount": float(share), # anteiliger Rabatt dieser Steuergruppe (BT-92 je Satz)
})
else:
buckets = raw_buckets
tax_basis_total = sum((Decimal(str(b["taxable_amount"])) for b in buckets), Decimal("0")).quantize(
Decimal("0.01"), rounding=ROUND_HALF_UP)
total_vat_amount = sum((Decimal(str(b["tax_amount"])) for b in buckets), Decimal("0")).quantize(
Decimal("0.01"), rounding=ROUND_HALF_UP)
total_gross_amount = (tax_basis_total + total_vat_amount).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
return {
"line_net_total": float(line_net_total),
"allowance_amount": float(allowance),
"tax_basis_total": float(tax_basis_total),
"total_vat_amount": float(total_vat_amount),
"total_gross_amount": float(total_gross_amount),
"tax_buckets": buckets,
}
```
- [ ] **Step 4: Run tests to verify they pass**
Run: `pytest tests/test_invoice_logic.py -k "aggregate_tax_subtotals or compute_invoice_totals" -v`
Expected: 4 tests PASS.
- [ ] **Step 5: Write the failing generator tests for multi-tax XML output**
Add to `tests/test_invoice_logic.py`:
```python
def _mixed_rate_invoice():
data = _single_rate_invoice()
items = [
{"name": "Beratung", "description": "", "quantity": 2.0, "unit": "Std.",
"unit_price": 100.0, "total_amount": 200.0, "vat_rate_percent": 19.0},
{"name": "Buch", "description": "", "quantity": 1.0, "unit": "Stk.",
"unit_price": 50.0, "total_amount": 50.0, "vat_rate_percent": 7.0},
]
totals = compute_invoice_totals(items)
data["items"] = items
data["tax_buckets"] = totals["tax_buckets"]
data["line_net_total"] = totals["line_net_total"]
data["allowance_amount"] = totals["allowance_amount"]
data["total_net_amount"] = totals["tax_basis_total"]
data["total_vat_amount"] = totals["total_vat_amount"]
data["total_gross_amount"] = totals["total_gross_amount"]
return data
def test_generate_compliant_xrechnung_mixed_rates_two_tax_subtotals():
xml_str = generate_compliant_xrechnung(_mixed_rate_invoice())
root = ET.fromstring(xml_str.encode("utf-8"))
ns = {"cbc": "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2",
"cac": "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"}
subtotals = root.findall(".//cac:TaxTotal/cac:TaxSubtotal", ns)
assert len(subtotals) == 2
percents = sorted(st.find("cac:TaxCategory/cbc:Percent", ns).text for st in subtotals)
assert percents == ["19.00", "7.00"] or percents == ["7.00", "19.00"]
lines = root.findall(".//cac:InvoiceLine", ns)
line_percents = [ln.find(".//cbc:Percent", ns).text for ln in lines]
assert set(line_percents) == {"19.00", "7.00"}
def test_generate_compliant_zugferd_cii_mixed_rates_two_header_taxes():
xml_str = generate_compliant_zugferd_cii(_mixed_rate_invoice())
root = ET.fromstring(xml_str.encode("utf-8"))
ns = {"ram": "urn:un:unece:uncefact:data:standard:ReusableAggregateBusinessInformationEntity:100"}
header_taxes = root.findall(
".//ram:ApplicableHeaderTradeSettlement/ram:ApplicableTradeTax", ns)
assert len(header_taxes) == 2
```
- [ ] **Step 6: Run tests to verify they fail**
Run: `pytest tests/test_invoice_logic.py -k mixed_rates -v`
Expected: FAIL (still only one `TaxSubtotal`/`ApplicableTradeTax` since generators aren't updated yet).
- [ ] **Step 7: Update `generate_compliant_xrechnung` to build multiple `TaxSubtotal` and per-line rates**
Replace the tax-total block:
```python
tax_total = ET.SubElement(root, f"{{{CAC}}}TaxTotal")
ET.SubElement(tax_total, f"{{{CBC}}}TaxAmount", {"currencyID": "EUR"}).text = f"{data['total_vat_amount']:.2f}"
tax_subtotal = ET.SubElement(tax_total, f"{{{CAC}}}TaxSubtotal")
ET.SubElement(tax_subtotal, f"{{{CBC}}}TaxableAmount", {"currencyID": "EUR"}).text = f"{data['total_net_amount']:.2f}"
ET.SubElement(tax_subtotal, f"{{{CBC}}}TaxAmount", {"currencyID": "EUR"}).text = f"{data['total_vat_amount']:.2f}"
tax_cat = ET.SubElement(tax_subtotal, f"{{{CAC}}}TaxCategory")
ET.SubElement(tax_cat, f"{{{CBC}}}ID").text = vat_category
ET.SubElement(tax_cat, f"{{{CBC}}}Percent").text = f"{vat_rate:.2f}"
tax_scheme = ET.SubElement(tax_cat, f"{{{CAC}}}TaxScheme")
ET.SubElement(tax_scheme, f"{{{CBC}}}ID").text = "VAT"
```
with:
```python
tax_total = ET.SubElement(root, f"{{{CAC}}}TaxTotal")
ET.SubElement(tax_total, f"{{{CBC}}}TaxAmount", {"currencyID": "EUR"}).text = f"{data['total_vat_amount']:.2f}"
# BG-23: pro Steuersatz-Bucket ein eigenes cac:TaxSubtotal (Mehrfach-Steuersaetze).
# Fallback fuer Aufrufer ohne tax_buckets (Einzelsteuersatz, byte-identisch zu vor Phase 1):
# synthetischer Ein-Bucket aus dem globalen Steuersatz und den Gesamtsummen.
_tax_buckets = data.get("tax_buckets") or [{
"rate": vat_rate, "category": vat_category,
"taxable_amount": data["total_net_amount"], "tax_amount": data["total_vat_amount"],
}]
for bucket in _tax_buckets:
tax_subtotal = ET.SubElement(tax_total, f"{{{CAC}}}TaxSubtotal")
ET.SubElement(tax_subtotal, f"{{{CBC}}}TaxableAmount", {"currencyID": "EUR"}).text = f"{bucket['taxable_amount']:.2f}"
ET.SubElement(tax_subtotal, f"{{{CBC}}}TaxAmount", {"currencyID": "EUR"}).text = f"{bucket['tax_amount']:.2f}"
tax_cat = ET.SubElement(tax_subtotal, f"{{{CAC}}}TaxCategory")
ET.SubElement(tax_cat, f"{{{CBC}}}ID").text = bucket["category"]
ET.SubElement(tax_cat, f"{{{CBC}}}Percent").text = f"{bucket['rate']:.2f}"
tax_scheme = ET.SubElement(tax_cat, f"{{{CAC}}}TaxScheme")
ET.SubElement(tax_scheme, f"{{{CBC}}}ID").text = "VAT"
```
Note: `vat_category`/`vat_rate` (the old global variables computed via `vat_rate = float(data['vat_rate_percent'])`) stay defined above this block — the `_tax_buckets` fallback reads them so single-rate callers without `tax_buckets` still produce exactly one `TaxSubtotal` identical to the pre-Phase-1 output.
- [ ] **Step 8: Update the `LegalMonetaryTotal` block to add optional `AllowanceTotalAmount` and use `line_net_total`**
Replace:
```python
monetary_total = ET.SubElement(root, f"{{{CAC}}}LegalMonetaryTotal")
ET.SubElement(monetary_total, f"{{{CBC}}}LineExtensionAmount", {"currencyID": "EUR"}).text = f"{data['total_net_amount']:.2f}"
ET.SubElement(monetary_total, f"{{{CBC}}}TaxExclusiveAmount", {"currencyID": "EUR"}).text = f"{data['total_net_amount']:.2f}"
ET.SubElement(monetary_total, f"{{{CBC}}}TaxInclusiveAmount", {"currencyID": "EUR"}).text = f"{data['total_gross_amount']:.2f}"
ET.SubElement(monetary_total, f"{{{CBC}}}PayableAmount", {"currencyID": "EUR"}).text = f"{data['total_gross_amount']:.2f}"
```
with:
```python
monetary_total = ET.SubElement(root, f"{{{CAC}}}LegalMonetaryTotal")
# BT-106 LineExtensionAmount ist die UNRABATTIERTE Positionssumme; BT-109 TaxExclusiveAmount
# ("total_net_amount") ist bereits die um einen etwaigen dokumentenweiten Rabatt (BT-107)
# reduzierte Steuerbasis - siehe compute_invoice_totals(). Ohne Rabatt sind beide identisch.
ET.SubElement(monetary_total, f"{{{CBC}}}LineExtensionAmount", {"currencyID": "EUR"}).text = f"{data.get('line_net_total', data['total_net_amount']):.2f}"
ET.SubElement(monetary_total, f"{{{CBC}}}TaxExclusiveAmount", {"currencyID": "EUR"}).text = f"{data['total_net_amount']:.2f}"
ET.SubElement(monetary_total, f"{{{CBC}}}TaxInclusiveAmount", {"currencyID": "EUR"}).text = f"{data['total_gross_amount']:.2f}"
if data.get("allowance_amount", 0.0) > 0:
ET.SubElement(monetary_total, f"{{{CBC}}}AllowanceTotalAmount", {"currencyID": "EUR"}).text = f"{data['allowance_amount']:.2f}"
ET.SubElement(monetary_total, f"{{{CBC}}}PayableAmount", {"currencyID": "EUR"}).text = f"{data['total_gross_amount']:.2f}"
```
- [ ] **Step 9: Update the per-line `InvoiceLine` loop to use each item's own VAT rate**
Replace:
```python
item_tax = ET.SubElement(item_elem, f"{{{CAC}}}ClassifiedTaxCategory")
ET.SubElement(item_tax, f"{{{CBC}}}ID").text = vat_category
ET.SubElement(item_tax, f"{{{CBC}}}Percent").text = f"{vat_rate:.2f}"
item_scheme = ET.SubElement(item_tax, f"{{{CAC}}}TaxScheme")
ET.SubElement(item_scheme, f"{{{CBC}}}ID").text = "VAT"
```
with:
```python
# BG-30 Positionssteuer: nutzt den eigenen Steuersatz der Position, falls vorhanden
# (Multi-Tax-Support) - fällt sonst auf den globalen Rechnungssteuersatz zurück.
line_vat_rate = float(item.get("vat_rate_percent", data['vat_rate_percent']))
line_vat_category = "S" if line_vat_rate > 0.0 else "E"
item_tax = ET.SubElement(item_elem, f"{{{CAC}}}ClassifiedTaxCategory")
ET.SubElement(item_tax, f"{{{CBC}}}ID").text = line_vat_category
ET.SubElement(item_tax, f"{{{CBC}}}Percent").text = f"{line_vat_rate:.2f}"
item_scheme = ET.SubElement(item_tax, f"{{{CAC}}}TaxScheme")
ET.SubElement(item_scheme, f"{{{CBC}}}ID").text = "VAT"
```
- [ ] **Step 10: Apply the analogous changes to `generate_compliant_zugferd_cii`**
Replace the per-line tax block:
```python
line_settle = ET.SubElement(line_item, f"{{{RAM}}}SpecifiedLineTradeSettlement")
trade_tax = ET.SubElement(line_settle, f"{{{RAM}}}ApplicableTradeTax")
ET.SubElement(trade_tax, f"{{{RAM}}}TypeCode").text = "VAT"
ET.SubElement(trade_tax, f"{{{RAM}}}CategoryCode").text = vat_category
ET.SubElement(trade_tax, f"{{{RAM}}}RateApplicablePercent").text = f"{vat_rate:.2f}"
```
with:
```python
line_vat_rate = float(item.get("vat_rate_percent", data['vat_rate_percent']))
line_vat_category = "S" if line_vat_rate > 0.0 else "E"
line_settle = ET.SubElement(line_item, f"{{{RAM}}}SpecifiedLineTradeSettlement")
trade_tax = ET.SubElement(line_settle, f"{{{RAM}}}ApplicableTradeTax")
ET.SubElement(trade_tax, f"{{{RAM}}}TypeCode").text = "VAT"
ET.SubElement(trade_tax, f"{{{RAM}}}CategoryCode").text = line_vat_category
ET.SubElement(trade_tax, f"{{{RAM}}}RateApplicablePercent").text = f"{line_vat_rate:.2f}"
```
Replace the header tax block:
```python
header_tax = ET.SubElement(header_settle, f"{{{RAM}}}ApplicableTradeTax")
ET.SubElement(header_tax, f"{{{RAM}}}CalculatedAmount").text = f"{data['total_vat_amount']:.2f}"
ET.SubElement(header_tax, f"{{{RAM}}}TypeCode").text = "VAT"
ET.SubElement(header_tax, f"{{{RAM}}}BasisAmount").text = f"{data['total_net_amount']:.2f}"
ET.SubElement(header_tax, f"{{{RAM}}}CategoryCode").text = vat_category
ET.SubElement(header_tax, f"{{{RAM}}}RateApplicablePercent").text = f"{vat_rate:.2f}"
```
with:
```python
# BG-23: pro Steuersatz-Bucket ein eigenes ram:ApplicableTradeTax (Mehrfach-Steuersaetze).
# Fallback fuer Aufrufer ohne tax_buckets (Einzelsteuersatz, byte-identisch zu vor Phase 1):
# synthetischer Ein-Bucket aus dem globalen Steuersatz und den Gesamtsummen.
_tax_buckets = data.get("tax_buckets") or [{
"rate": vat_rate, "category": vat_category,
"taxable_amount": data["total_net_amount"], "tax_amount": data["total_vat_amount"],
}]
for bucket in _tax_buckets:
header_tax = ET.SubElement(header_settle, f"{{{RAM}}}ApplicableTradeTax")
ET.SubElement(header_tax, f"{{{RAM}}}CalculatedAmount").text = f"{bucket['tax_amount']:.2f}"
ET.SubElement(header_tax, f"{{{RAM}}}TypeCode").text = "VAT"
ET.SubElement(header_tax, f"{{{RAM}}}BasisAmount").text = f"{bucket['taxable_amount']:.2f}"
ET.SubElement(header_tax, f"{{{RAM}}}CategoryCode").text = bucket["category"]
ET.SubElement(header_tax, f"{{{RAM}}}RateApplicablePercent").text = f"{bucket['rate']:.2f}"
```
Replace the header monetary summation block:
```python
header_sum = ET.SubElement(header_settle, f"{{{RAM}}}SpecifiedTradeSettlementHeaderMonetarySummation")
ET.SubElement(header_sum, f"{{{RAM}}}LineTotalAmount").text = f"{data['total_net_amount']:.2f}"
ET.SubElement(header_sum, f"{{{RAM}}}TaxBasisTotalAmount").text = f"{data['total_net_amount']:.2f}"
ET.SubElement(header_sum, f"{{{RAM}}}TaxTotalAmount", {"currencyID": "EUR"}).text = f"{data['total_vat_amount']:.2f}"
ET.SubElement(header_sum, f"{{{RAM}}}GrandTotalAmount").text = f"{data['total_gross_amount']:.2f}"
ET.SubElement(header_sum, f"{{{RAM}}}DuePayableAmount").text = f"{data['total_gross_amount']:.2f}"
```
with:
```python
header_sum = ET.SubElement(header_settle, f"{{{RAM}}}SpecifiedTradeSettlementHeaderMonetarySummation")
ET.SubElement(header_sum, f"{{{RAM}}}LineTotalAmount").text = f"{data.get('line_net_total', data['total_net_amount']):.2f}"
if data.get("allowance_amount", 0.0) > 0:
ET.SubElement(header_sum, f"{{{RAM}}}AllowanceTotalAmount").text = f"{data['allowance_amount']:.2f}"
ET.SubElement(header_sum, f"{{{RAM}}}TaxBasisTotalAmount").text = f"{data['total_net_amount']:.2f}"
ET.SubElement(header_sum, f"{{{RAM}}}TaxTotalAmount", {"currencyID": "EUR"}).text = f"{data['total_vat_amount']:.2f}"
ET.SubElement(header_sum, f"{{{RAM}}}GrandTotalAmount").text = f"{data['total_gross_amount']:.2f}"
ET.SubElement(header_sum, f"{{{RAM}}}DuePayableAmount").text = f"{data['total_gross_amount']:.2f}"
```
- [ ] **Step 11: Run tests to verify they pass**
Run: `pytest tests/test_invoice_logic.py -v`
Expected: all tests so far PASS (single-rate golden tests from Task 1 must still pass unchanged — this proves backward compatibility).
- [ ] **Step 12: Wire per-item VAT rate input into `main.py`**
Find the item loop columns:
```python
c_qty, c_unit, c_price, c_total = st.columns(4)
with c_qty:
item_quantity = st.number_input("Menge (BT-129) :red[*]", value=float(item.get("quantity") or 0.0), format="%.3f", step=0.1, key=f"item_qty_{inv_id}_{idx}")
with c_unit:
item_unit = st.text_input("Einheit (BT-130) :red[*]", value=item.get("unit", "Stk."), key=f"item_unit_{inv_id}_{idx}")
with c_price:
item_price = st.number_input("Einzelpreis (BT-146, €) :red[*]", value=float(item.get("unit_price") or 0.0), format="%.2f", key=f"item_price_{inv_id}_{idx}")
with c_total:
raw_item_total = item_quantity * item_price
item_total_amount = kaufmaennisches_runden(raw_item_total)
st.markdown("**Gesamtbetrag (€):**")
st.markdown(f"**{item_total_amount:.2f} €**")
edited_items.append({
"name": item_name,
"description": item_description,
"quantity": item_quantity,
"unit": item_unit,
"unit_price": item_price,
"total_amount": item_total_amount
})
```
Replace with:
```python
c_qty, c_unit, c_price, c_vat, c_total = st.columns(5)
with c_qty:
item_quantity = st.number_input("Menge (BT-129) :red[*]", value=float(item.get("quantity") or 0.0), format="%.3f", step=0.1, key=f"item_qty_{inv_id}_{idx}")
with c_unit:
item_unit = st.text_input("Einheit (BT-130) :red[*]", value=item.get("unit", "Stk."), key=f"item_unit_{inv_id}_{idx}")
with c_price:
item_price = st.number_input("Einzelpreis (BT-146, €) :red[*]", value=float(item.get("unit_price") or 0.0), format="%.2f", key=f"item_price_{inv_id}_{idx}")
with c_vat:
item_vat_rate = st.number_input(
"Steuersatz (BT-152, %) :red[*]",
value=float(item.get("vat_rate_percent") or data.get("vat_rate_percent") or 19.0),
key=f"item_vat_{inv_id}_{idx}",
)
with c_total:
raw_item_total = item_quantity * item_price
item_total_amount = kaufmaennisches_runden(raw_item_total)
st.markdown("**Gesamtbetrag (€):**")
st.markdown(f"**{item_total_amount:.2f} €**")
edited_items.append({
"name": item_name,
"description": item_description,
"quantity": item_quantity,
"unit": item_unit,
"unit_price": item_price,
"total_amount": item_total_amount,
"vat_rate_percent": item_vat_rate,
})
```
- [ ] **Step 13: Default the VAT rate of newly-added positions to the invoice's global rate**
Find:
```python
if st.button("➕ Neue Position hinzufügen"):
data["items"].append({"name": "Neue Dienstleistung", "description": "", "quantity": 1.0, "unit": "Stk.", "unit_price": 0.0})
st.rerun()
```
Replace with:
```python
if st.button("➕ Neue Position hinzufügen"):
data["items"].append({
"name": "Neue Dienstleistung", "description": "", "quantity": 1.0, "unit": "Stk.",
"unit_price": 0.0, "vat_rate_percent": float(data.get("vat_rate_percent") or 19.0),
})
st.rerun()
```
- [ ] **Step 14: Replace the manual Decimal-summation block with `compute_invoice_totals`**
Find:
```python
st.markdown("---")
vat_rate = st.number_input("Steuersatz (BT-152, %) :red[*]", value=float(data.get("vat_rate_percent") or 19.0))
# Summen mit Decimal bilden, damit die Netto-Summe exakt der Summe der (bereits gerundeten)
# Positionsbeträge entspricht. Reine float-Summation kann minimal driften und dann zu
# BR-CO-10-Validierungsfehlern führen (Summe der Zeilen != TaxExclusiveAmount).
net_dec = sum((Decimal(str(item["total_amount"])) for item in edited_items), Decimal("0"))
vat_dec = (net_dec * Decimal(str(vat_rate)) / Decimal("100")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
gross_dec = (net_dec + vat_dec).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
computed_net_amount = float(net_dec.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))
computed_vat_amount = float(vat_dec)
computed_gross_amount = float(gross_dec)
```
Replace with:
```python
st.markdown("---")
# Der frühere globale Steuersatz dient jetzt nur noch als Vorbelegung für neu
# hinzugefügte Positionen (siehe "Neue Position hinzufügen" oben) - abgerechnet wird
# pro Position mit dem jeweils eingetragenen Steuersatz (Multi-Tax-Support).
vat_rate = st.number_input("Standard-Steuersatz für neue Positionen (%)", value=float(data.get("vat_rate_percent") or 19.0))
totals = compute_invoice_totals(edited_items)
computed_net_amount = totals["tax_basis_total"]
computed_vat_amount = totals["total_vat_amount"]
computed_gross_amount = totals["total_gross_amount"]
```
Add `compute_invoice_totals` to the `from invoice_logic import (...)` block in `main.py`.
- [ ] **Step 15: Pass the new fields into `final_invoice_dict`**
Find:
```python
final_invoice_dict = {
"invoice_type_code": selected_type_code,
"invoice_number": inv_num, "invoice_date": inv_date, "delivery_date": inv_delivery,
"buyer_reference": buyer_ref, "customer_number": cust_num, "recipient_name": r_name, "recipient_email": r_email,
"recipient_street": r_street, "recipient_zip": r_zip, "recipient_city": r_city,
"items": edited_items, "total_net_amount": computed_net_amount, "total_vat_amount": computed_vat_amount,
"vat_rate_percent": vat_rate, "total_gross_amount": computed_gross_amount, "iban": VERKAEUFER_IBAN,
"payment_means_code": selected_payment_means_code,
```
Replace with:
```python
final_invoice_dict = {
"invoice_type_code": selected_type_code,
"invoice_number": inv_num, "invoice_date": inv_date, "delivery_date": inv_delivery,
"buyer_reference": buyer_ref, "customer_number": cust_num, "recipient_name": r_name, "recipient_email": r_email,
"recipient_street": r_street, "recipient_zip": r_zip, "recipient_city": r_city,
"items": edited_items, "total_net_amount": computed_net_amount, "total_vat_amount": computed_vat_amount,
"vat_rate_percent": vat_rate, "total_gross_amount": computed_gross_amount, "iban": VERKAEUFER_IBAN,
"payment_means_code": selected_payment_means_code,
"tax_buckets": totals["tax_buckets"],
"line_net_total": totals["line_net_total"],
"allowance_amount": totals["allowance_amount"],
```
- [ ] **Step 16: Manual smoke test with mixed rates**
Run: `streamlit run main.py`. Extract any invoice, set one position to 19% and another to 7%, export XRechnung, and confirm the downloaded XML has two `` elements with the correct `Percent`/`TaxableAmount`/`TaxAmount` values.
- [ ] **Step 17: Commit**
```bash
git add invoice_logic.py main.py tests/test_invoice_logic.py
git commit -m "feat: support mixed VAT rates per line (BG-23) in both XML generators"
```
---
### Task 4: Skonto (BT-20 payment terms) & globale Rabatte (BT-21)
**Files:**
- Modify: `invoice_logic.py` (add `format_skonto_text`; add `cac:PaymentTerms`/`ram:SpecifiedTradePaymentTerms.Description` and `cac:AllowanceCharge`/`ram:SpecifiedTradeAllowanceCharge`)
- Modify: `main.py` (discount/Skonto input fields, wire `discount_percent` into `compute_invoice_totals`)
- Modify: `tests/test_invoice_logic.py`
**Interfaces:**
- Consumes: `berechne_faelligkeit` (Task 1), `compute_invoice_totals(items, discount_percent, discount_amount)` (Task 3, already supports discount params).
- Produces: `invoice_logic.format_skonto_text(skonto_percent, skonto_days, invoice_date_str) -> str`. `final_invoice_dict` gains `allowance_reason`, `skonto_percent`, `skonto_days`.
- [ ] **Step 1: Write the failing tests for `format_skonto_text`**
Add to `tests/test_invoice_logic.py`:
```python
from invoice_logic import format_skonto_text
def test_format_skonto_text_with_valid_percent_and_days():
text = format_skonto_text(2.0, 10, "2026-01-15")
assert "2026-01-25" in text
assert "2%" in text
def test_format_skonto_text_returns_empty_when_no_percent():
assert format_skonto_text(0, 10, "2026-01-15") == ""
assert format_skonto_text(None, 10, "2026-01-15") == ""
def test_format_skonto_text_returns_empty_when_no_days():
assert format_skonto_text(2.0, "", "2026-01-15") == ""
assert format_skonto_text(2.0, None, "2026-01-15") == ""
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `pytest tests/test_invoice_logic.py -k skonto -v`
Expected: FAIL with `ImportError: cannot import name 'format_skonto_text'`.
- [ ] **Step 3: Implement `format_skonto_text` in `invoice_logic.py`**
Add after `berechne_faelligkeit`:
```python
def format_skonto_text(skonto_percent, skonto_days, invoice_date_str):
"""Baut den Freitext fuer eine Skonto-Vereinbarung. EN16931 kennt kein eigenes strukturiertes
Skonto-Feld - der Text wird daher als cbc:Note (UBL) bzw. ram:Description (CII) neben dem
regulaeren Faelligkeitsdatum in den Zahlungsbedingungen (BT-20) abgelegt. Gibt "" zurueck,
wenn kein gueltiges Skonto vereinbart ist (KEIN Rateraten bei fehlenden Angaben)."""
try:
percent = float(skonto_percent)
except (TypeError, ValueError):
return ""
if percent <= 0:
return ""
try:
days = int(skonto_days)
except (TypeError, ValueError):
return ""
if days <= 0:
return ""
percent_str = f"{percent:.2f}".rstrip("0").rstrip(".")
skonto_date = berechne_faelligkeit(invoice_date_str, days)
if skonto_date:
return f"Bei Zahlung bis {skonto_date} gewähren wir {percent_str}% Skonto."
return f"Bei Zahlung innerhalb von {days} Tagen gewähren wir {percent_str}% Skonto."
```
- [ ] **Step 4: Run tests to verify they pass**
Run: `pytest tests/test_invoice_logic.py -k skonto -v`
Expected: 3 tests PASS.
- [ ] **Step 5: Write the failing generator tests for AllowanceCharge and PaymentTerms**
Add to `tests/test_invoice_logic.py`:
```python
def _discounted_invoice():
data = _single_rate_invoice()
items = [{"total_amount": 200.0, "vat_rate_percent": 19.0, "name": "Beratung",
"description": "", "quantity": 2.0, "unit": "Std.", "unit_price": 100.0}]
totals = compute_invoice_totals(items, discount_percent=10.0)
data["items"] = items
data["tax_buckets"] = totals["tax_buckets"]
data["line_net_total"] = totals["line_net_total"]
data["allowance_amount"] = totals["allowance_amount"]
data["allowance_reason"] = "Treuerabatt"
data["total_net_amount"] = totals["tax_basis_total"]
data["total_vat_amount"] = totals["total_vat_amount"]
data["total_gross_amount"] = totals["total_gross_amount"]
data["skonto_percent"] = 2.0
data["skonto_days"] = "10"
return data
def test_generate_compliant_xrechnung_includes_allowance_charge_and_payment_terms():
xml_str = generate_compliant_xrechnung(_discounted_invoice())
root = ET.fromstring(xml_str.encode("utf-8"))
ns = {"cbc": "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2",
"cac": "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"}
allowance = root.find(".//cac:AllowanceCharge", ns)
assert allowance.find("cbc:AllowanceChargeReason", ns).text == "Treuerabatt"
assert allowance.find("cbc:Amount", ns).text == "20.00"
payment_terms_note = root.find(".//cac:PaymentTerms/cbc:Note", ns).text
assert "Skonto" in payment_terms_note
def test_generate_compliant_zugferd_cii_includes_allowance_charge_and_payment_terms_description():
xml_str = generate_compliant_zugferd_cii(_discounted_invoice())
root = ET.fromstring(xml_str.encode("utf-8"))
ns = {"ram": "urn:un:unece:uncefact:data:standard:ReusableAggregateBusinessInformationEntity:100"}
allowance = root.find(".//ram:SpecifiedTradeAllowanceCharge", ns)
assert allowance.find("ram:ActualAmount", ns).text == "20.00"
assert allowance.find("ram:Reason", ns).text == "Treuerabatt"
description = root.find(".//ram:SpecifiedTradePaymentTerms/ram:Description", ns).text
assert "Skonto" in description
def test_generate_compliant_xrechnung_mixed_rate_discount_emits_allowance_per_rate():
"""Voll EN16931-konform: ein Rabatt auf eine Rechnung mit 19%- UND 7%-Positionen erzeugt
JE Steuersatz einen eigenen AllowanceCharge-Block mit anteiligem Rabattbetrag."""
items = [
{"total_amount": 100.0, "vat_rate_percent": 19.0, "name": "A", "description": "",
"quantity": 1.0, "unit": "Stk.", "unit_price": 100.0},
{"total_amount": 100.0, "vat_rate_percent": 7.0, "name": "B", "description": "",
"quantity": 1.0, "unit": "Stk.", "unit_price": 100.0},
]
totals = compute_invoice_totals(items, discount_percent=10.0)
data = _single_rate_invoice()
data["items"] = items
data["tax_buckets"] = totals["tax_buckets"]
data["line_net_total"] = totals["line_net_total"]
data["allowance_amount"] = totals["allowance_amount"]
data["allowance_reason"] = "Aktionsrabatt"
data["total_net_amount"] = totals["tax_basis_total"]
data["total_vat_amount"] = totals["total_vat_amount"]
data["total_gross_amount"] = totals["total_gross_amount"]
xml_str = generate_compliant_xrechnung(data)
root = ET.fromstring(xml_str.encode("utf-8"))
ns = {"cbc": "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2",
"cac": "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"}
allowances = root.findall(".//cac:AllowanceCharge", ns)
assert len(allowances) == 2
percents = sorted(a.find("cac:TaxCategory/cbc:Percent", ns).text for a in allowances)
assert percents == ["19.00", "7.00"] or percents == ["7.00", "19.00"]
total_alloc = sum(float(a.find("cbc:Amount", ns).text) for a in allowances)
assert abs(total_alloc - 20.0) < 0.001
```
- [ ] **Step 6: Run tests to verify they fail**
Run: `pytest tests/test_invoice_logic.py -k "allowance_charge or mixed_rate_discount" -v`
Expected: FAIL (elements don't exist yet).
- [ ] **Step 7: Add `cac:AllowanceCharge` and `cac:PaymentTerms` to `generate_compliant_xrechnung`**
Immediately before the line `tax_total = ET.SubElement(root, f"{{{CAC}}}TaxTotal")` (from Task 3, Step 7), insert:
```python
# BT-20 Skonto-Vereinbarung als Freitext in cac:PaymentTerms/cbc:Note - EN16931 kennt kein
# eigenes strukturiertes Skonto-Feld. Muss laut UBL-Schema NACH PaymentMeans und VOR
# AllowanceCharge/TaxTotal stehen.
_skonto_text = format_skonto_text(data.get("skonto_percent"), data.get("skonto_days"), data["invoice_date"])
if _skonto_text:
payment_terms = ET.SubElement(root, f"{{{CAC}}}PaymentTerms")
ET.SubElement(payment_terms, f"{{{CBC}}}Note").text = _skonto_text
# BT-21 Dokumentenweiter Rabatt (BG-20). Muss laut UBL-Schema NACH PaymentTerms und VOR
# TaxTotal stehen. Voll EN16931-konform: bei gemischten Steuersaetzen wird JE Steuersatz ein
# eigener AllowanceCharge-Block mit dem anteilig zugeordneten Rabattbetrag erzeugt, damit jede
# Steuergruppe (BG-23) ihren eigenen Rabattanteil fuehrt (BR-CO-10/BR-DEC).
if data.get("allowance_amount", 0.0) > 0:
for bucket in data.get("tax_buckets") or []:
bucket_allowance = bucket.get("allowance_amount", 0.0)
if bucket_allowance <= 0:
continue
allowance_charge = ET.SubElement(root, f"{{{CAC}}}AllowanceCharge")
ET.SubElement(allowance_charge, f"{{{CBC}}}ChargeIndicator").text = "false"
ET.SubElement(allowance_charge, f"{{{CBC}}}AllowanceChargeReason").text = str(data.get("allowance_reason") or "Rabatt")
ET.SubElement(allowance_charge, f"{{{CBC}}}Amount", {"currencyID": "EUR"}).text = f"{bucket_allowance:.2f}"
ac_tax_cat = ET.SubElement(allowance_charge, f"{{{CAC}}}TaxCategory")
ET.SubElement(ac_tax_cat, f"{{{CBC}}}ID").text = bucket["category"]
ET.SubElement(ac_tax_cat, f"{{{CBC}}}Percent").text = f"{bucket['rate']:.2f}"
ac_tax_scheme = ET.SubElement(ac_tax_cat, f"{{{CAC}}}TaxScheme")
ET.SubElement(ac_tax_scheme, f"{{{CBC}}}ID").text = "VAT"
```
- [ ] **Step 8: Add `ram:SpecifiedTradeAllowanceCharge` and the Skonto `ram:Description` to `generate_compliant_zugferd_cii`**
Find the service period block (`BillingSpecifiedPeriod`) followed by the due-date block:
```python
if data.get("service_start_date") and data.get("service_end_date"):
period = ET.SubElement(header_settle, f"{{{RAM}}}BillingSpecifiedPeriod")
start_pt = ET.SubElement(period, f"{{{RAM}}}StartDateTime")
ET.SubElement(start_pt, f"{{{UDT}}}DateTimeString", {"format": "102"}).text = format_cii_date(data["service_start_date"])
end_pt = ET.SubElement(period, f"{{{RAM}}}EndDateTime")
ET.SubElement(end_pt, f"{{{UDT}}}DateTimeString", {"format": "102"}).text = format_cii_date(data["service_end_date"])
# BT-9 Fälligkeitsdatum, abgeleitet aus dem tatsächlichen Zahlungsziel der Rechnung. Erfüllt
# allein BR-CO-25 (Fälligkeitsdatum ODER Zahlungsbedingungen - wir nutzen nur das Datum).
# Kein fester Fallback: ohne gültiges Zahlungsziel wird der Block weggelassen (die UI blockiert
# dann den Export).
_cii_due = berechne_faelligkeit(data["invoice_date"], data.get("payment_term_days"))
if _cii_due:
payment_terms = ET.SubElement(header_settle, f"{{{RAM}}}SpecifiedTradePaymentTerms")
due_date_elem = ET.SubElement(payment_terms, f"{{{RAM}}}DueDateDateTime")
ET.SubElement(due_date_elem, f"{{{UDT}}}DateTimeString", {"format": "102"}).text = _cii_due.replace("-", "")
```
Replace with:
```python
if data.get("service_start_date") and data.get("service_end_date"):
period = ET.SubElement(header_settle, f"{{{RAM}}}BillingSpecifiedPeriod")
start_pt = ET.SubElement(period, f"{{{RAM}}}StartDateTime")
ET.SubElement(start_pt, f"{{{UDT}}}DateTimeString", {"format": "102"}).text = format_cii_date(data["service_start_date"])
end_pt = ET.SubElement(period, f"{{{RAM}}}EndDateTime")
ET.SubElement(end_pt, f"{{{UDT}}}DateTimeString", {"format": "102"}).text = format_cii_date(data["service_end_date"])
# BT-21 Dokumentenweiter Rabatt (BG-20). Muss laut CII-Schema NACH BillingSpecifiedPeriod und
# VOR SpecifiedTradePaymentTerms stehen. Voll EN16931-konform: bei gemischten Steuersaetzen wird
# JE Steuersatz ein eigener SpecifiedTradeAllowanceCharge-Block mit dem anteilig zugeordneten
# Rabattbetrag erzeugt (jede Steuergruppe fuehrt ihren eigenen Rabattanteil).
if data.get("allowance_amount", 0.0) > 0:
for bucket in data.get("tax_buckets") or []:
bucket_allowance = bucket.get("allowance_amount", 0.0)
if bucket_allowance <= 0:
continue
allowance_charge = ET.SubElement(header_settle, f"{{{RAM}}}SpecifiedTradeAllowanceCharge")
indicator = ET.SubElement(allowance_charge, f"{{{RAM}}}ChargeIndicator")
ET.SubElement(indicator, f"{{{UDT}}}Indicator").text = "false"
ET.SubElement(allowance_charge, f"{{{RAM}}}ActualAmount").text = f"{bucket_allowance:.2f}"
ET.SubElement(allowance_charge, f"{{{RAM}}}Reason").text = str(data.get("allowance_reason") or "Rabatt")
ac_tax = ET.SubElement(allowance_charge, f"{{{RAM}}}CategoryTradeTax")
ET.SubElement(ac_tax, f"{{{RAM}}}TypeCode").text = "VAT"
ET.SubElement(ac_tax, f"{{{RAM}}}CategoryCode").text = bucket["category"]
ET.SubElement(ac_tax, f"{{{RAM}}}RateApplicablePercent").text = f"{bucket['rate']:.2f}"
# BT-9 Fälligkeitsdatum, abgeleitet aus dem tatsächlichen Zahlungsziel der Rechnung, sowie
# optional die BT-20 Skonto-Freitextbeschreibung. ram:Description muss laut Schema VOR
# ram:DueDateDateTime stehen. Kein fester Fallback: ohne gültiges Zahlungsziel UND ohne
# Skonto wird der Block weggelassen (die UI blockiert dann den Export bzgl. Zahlungsziel).
_cii_due = berechne_faelligkeit(data["invoice_date"], data.get("payment_term_days"))
_skonto_text = format_skonto_text(data.get("skonto_percent"), data.get("skonto_days"), data["invoice_date"])
if _cii_due or _skonto_text:
payment_terms = ET.SubElement(header_settle, f"{{{RAM}}}SpecifiedTradePaymentTerms")
if _skonto_text:
ET.SubElement(payment_terms, f"{{{RAM}}}Description").text = _skonto_text
if _cii_due:
due_date_elem = ET.SubElement(payment_terms, f"{{{RAM}}}DueDateDateTime")
ET.SubElement(due_date_elem, f"{{{UDT}}}DateTimeString", {"format": "102"}).text = _cii_due.replace("-", "")
```
- [ ] **Step 9: Run tests to verify they pass**
Run: `pytest tests/test_invoice_logic.py -v`
Expected: all tests PASS, including the new `allowance_charge` and `mixed_rate_discount` tests and every earlier test (single-rate and mixed-rate golden tests still pass — `allowance_amount` is `0.0` for those fixtures, so the new blocks are correctly skipped).
- [ ] **Step 10: Wire discount/Skonto input fields into `main.py`**
Find the line `vat_rate = st.number_input("Standard-Steuersatz für neue Positionen (%)", ...)` (from Task 3, Step 14) and insert directly below it, before the `totals = compute_invoice_totals(edited_items)` line:
```python
c_disc1, c_disc2, c_disc3, c_disc4 = st.columns(4)
with c_disc1:
discount_percent = st.number_input("Rabatt (BT-21, %)", value=0.0, min_value=0.0, max_value=100.0, step=0.5)
with c_disc2:
discount_reason = st.text_input("Rabattgrund", value="")
with c_disc3:
skonto_percent = st.number_input("Skonto (BT-20, %)", value=0.0, min_value=0.0, max_value=100.0, step=0.5)
with c_disc4:
skonto_days = st.text_input("Skonto-Frist (Tage)", value="")
totals = compute_invoice_totals(edited_items, discount_percent=discount_percent)
```
(Remove the plain `totals = compute_invoice_totals(edited_items)` line added in Task 3, Step 14 — it's replaced by the line above.)
- [ ] **Step 11: Add the new fields to `final_invoice_dict`**
Find (from Task 3, Step 15):
```python
"tax_buckets": totals["tax_buckets"],
"line_net_total": totals["line_net_total"],
"allowance_amount": totals["allowance_amount"],
```
Replace with:
```python
"tax_buckets": totals["tax_buckets"],
"line_net_total": totals["line_net_total"],
"allowance_amount": totals["allowance_amount"],
"allowance_reason": discount_reason,
"skonto_percent": skonto_percent,
"skonto_days": skonto_days,
```
Add `format_skonto_text` to the `from invoice_logic import (...)` block in `main.py` (only needed if `main.py` calls it directly for a live preview — otherwise the generators call it internally and no import is required in `main.py`; skip the import if unused).
- [ ] **Step 12: Manual smoke test**
Run: `streamlit run main.py`. Set "Rabatt (BT-21, %)" to `10`, "Rabattgrund" to `Treuerabatt`, "Skonto (BT-20, %)" to `2`, "Skonto-Frist (Tage)" to `10`. Export XRechnung and confirm the XML contains `` with the correct amount and `` mentioning "Skonto".
- [ ] **Step 13: Commit**
```bash
git add invoice_logic.py main.py tests/test_invoice_logic.py
git commit -m "feat: add document-level discount (BT-21) and Skonto payment terms (BT-20)"
```
---
### Task 5: IT-Systemhaus regex parser extensions
**Files:**
- Modify: `invoice_logic.py` (extend `extract_invoice_items_pure_python`'s unit pattern and `map_unece_unit_code`; add `extract_service_period`, `detect_contract_type`, `extract_hardware_serial_numbers`)
- Modify: `main.py` (wire fallback extraction, minimal read-only display)
- Modify: `tests/test_invoice_logic.py`
**Interfaces:**
- Produces: `invoice_logic.extract_service_period(text: str) -> tuple[str, str]` (each `""` or `YYYY-MM-DD`), `invoice_logic.detect_contract_type(text: str) -> str` (one of `"Managed Services"`, `"Cloud-Abonnement"`, `"Projekthonorar"`, or `""`), `invoice_logic.extract_hardware_serial_numbers(text: str) -> list[str]`.
- `extract_invoice_items_pure_python` gains support for the additional unit keywords `Monat(e)`, `Monate`, `PT`, `Personentag(e)`, `Lizenz(en)`, `Wartungspauschale` without breaking any of the existing keyword matches (Task 1's golden test for `Std.` must keep passing).
- [ ] **Step 1: Write the failing tests for the extended unit pattern**
Add to `tests/test_invoice_logic.py`:
```python
def test_extract_invoice_items_pure_python_recognizes_person_days():
text = "1\nProjektarbeit\n5,00 Personentage\n800,00\n4000,00\n"
items = extract_invoice_items_pure_python(text)
assert items[0]["quantity"] == 5.0
assert items[0]["unit"] == "Personentage"
def test_extract_invoice_items_pure_python_recognizes_license_months():
text = "1\nCloud-Lizenz\n3,00 Monat(e)\n49,00\n147,00\n"
items = extract_invoice_items_pure_python(text)
assert items[0]["unit"] == "Monat(e)"
def test_map_unece_unit_code_extended_units():
assert map_unece_unit_code("Monat(e)") == "MON"
assert map_unece_unit_code("Personentage") == "DAY"
assert map_unece_unit_code("Lizenz(en)") == "C62"
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `pytest tests/test_invoice_logic.py -k "person_days or license_months or extended_units" -v`
Expected: FAIL (current unit pattern doesn't recognize `Personentage`/`Monat(e)` and `map_unece_unit_code` doesn't map `"monat"`).
- [ ] **Step 3: Extend the unit pattern in `extract_invoice_items_pure_python`**
Find:
```python
unit_pattern = re.compile(r"^(\d+(?:[\.,]\d+)?)\s*(Tag\(e\)|Std\.|Stk\.|Stunden|Stück|stk|stck|stück)$", re.IGNORECASE)
```
Replace with:
```python
# IT-Systemhaus-Erweiterung: zusaetzlich Monats-/Lizenz-/Personentag-Abrechnungseinheiten,
# wie sie bei Managed-Services-, Cloud-Abo- und Projekthonorar-Rechnungen ueblich sind.
unit_pattern = re.compile(
r"^(\d+(?:[\.,]\d+)?)\s*(Tag\(e\)|Std\.|Stk\.|Stunden|Stück|stk|stck|stück|"
r"Monat\(e\)|Monate|Personentage|Personentag|PT|Lizenz\(en\)|Lizenzen|Lizenz|Wartungspauschale)$",
re.IGNORECASE,
)
```
- [ ] **Step 4: Extend `map_unece_unit_code`**
Find:
```python
def map_unece_unit_code(unit_str):
u = str(unit_str).lower()
if "tag" in u or "day" in u: return "DAY"
elif "std" in u or "stund" in u or "hour" in u or "hur" in u: return "HUR"
return "C62"
```
Replace with:
```python
def map_unece_unit_code(unit_str):
u = str(unit_str).lower()
if "monat" in u: return "MON"
if "personentag" in u or u == "pt": return "DAY"
if "tag" in u or "day" in u: return "DAY"
if "std" in u or "stund" in u or "hour" in u or "hur" in u: return "HUR"
return "C62"
```
- [ ] **Step 5: Run tests to verify they pass**
Run: `pytest tests/test_invoice_logic.py -v`
Expected: all tests PASS, including the pre-existing `test_extract_invoice_items_pure_python_basic` and `test_map_unece_unit_code_hours` (proves the extension didn't break the original keyword matches).
- [ ] **Step 6: Write the failing tests for `extract_service_period`, `detect_contract_type`, `extract_hardware_serial_numbers`**
Add to `tests/test_invoice_logic.py`:
```python
from invoice_logic import extract_service_period, detect_contract_type, extract_hardware_serial_numbers
def test_extract_service_period_finds_range():
text = "Managed Services\nLeistungszeitraum: 01.03.2026 bis 31.03.2026\n1\n..."
start, end = extract_service_period(text)
assert start == "2026-03-01"
assert end == "2026-03-31"
def test_extract_service_period_returns_empty_when_absent():
assert extract_service_period("Keine Zeitangabe hier.") == ("", "")
def test_detect_contract_type_managed_services():
assert detect_contract_type("Wartungsvertrag Managed Services Tier 2") == "Managed Services"
def test_detect_contract_type_cloud_abo():
assert detect_contract_type("Ihr Cloud-Abonnement für 2026") == "Cloud-Abonnement"
def test_detect_contract_type_returns_empty_when_no_match():
assert detect_contract_type("Ganz normale Rechnung ohne Vertragstyp") == ""
def test_extract_hardware_serial_numbers_finds_sn_pattern():
text = "Server Dell R750\nS/N: ABCD-1234-XYZ\nSwitch Cisco\nSeriennummer: SN9988771"
serials = extract_hardware_serial_numbers(text)
assert "ABCD-1234-XYZ" in serials
assert "SN9988771" in serials
def test_extract_hardware_serial_numbers_empty_when_absent():
assert extract_hardware_serial_numbers("Keine Seriennummern hier.") == []
```
- [ ] **Step 7: Run tests to verify they fail**
Run: `pytest tests/test_invoice_logic.py -k "service_period or contract_type or serial_numbers" -v`
Expected: FAIL with `ImportError`.
- [ ] **Step 8: Implement the three new extraction functions in `invoice_logic.py`**
Add after `extract_invoice_items_pure_python`:
```python
SERVICE_PERIOD_PATTERN = re.compile(
r"(?:Leistungszeitraum|Zeitraum|Einsatzzeitraum)\s*:?\s*"
r"(?:vom\s+)?(\d{1,2}[.\-]\d{1,2}[.\-]\d{4})\s*(?:bis|-|–)\s*(\d{1,2}[.\-]\d{1,2}[.\-]\d{4})",
re.IGNORECASE,
)
def extract_service_period(text):
"""Regex-Fallback fuer den Leistungszeitraum (z.B. 'Leistungszeitraum: 01.03.2026 bis
31.03.2026'), falls das LLM diesen nicht extrahiert hat. Gibt (start, end) als YYYY-MM-DD
zurueck, oder ("", "") wenn nichts gefunden wurde."""
match = SERVICE_PERIOD_PATTERN.search(text)
if not match:
return "", ""
start = parse_to_strict_xml_date(match.group(1).replace("-", "."))
end = parse_to_strict_xml_date(match.group(2).replace("-", "."))
return start, end
CONTRACT_TYPE_KEYWORDS = {
"Managed Services": re.compile(r"managed\s+services?", re.IGNORECASE),
"Cloud-Abonnement": re.compile(r"cloud[\s\-]?(abo(nnement)?|subscription)", re.IGNORECASE),
"Projekthonorar": re.compile(r"projekt(honorar|vergütung|pauschale)", re.IGNORECASE),
}
def detect_contract_type(text):
"""Erkennt den IT-Systemhaus-Vertragstyp anhand von Schluesselwoertern im Rechnungstext.
Gibt das erste gefundene Label zurueck oder "", wenn keines passt (keine Vermutung)."""
for label, pattern in CONTRACT_TYPE_KEYWORDS.items():
if pattern.search(text):
return label
return ""
SERIAL_NUMBER_PATTERN = re.compile(
r"(?:S/?N|Seriennummer|Serial(?:\s*No\.?)?)\s*:?\s*([A-Z0-9][A-Z0-9\-]{4,})",
re.IGNORECASE,
)
def extract_hardware_serial_numbers(text):
"""Findet Hardware-Seriennummern im Rechnungstext (Muster 'S/N: ...', 'Seriennummer: ...').
Gibt eine Liste der gefundenen Seriennummern in Text-Reihenfolge zurueck, ohne Duplikate."""
found = []
for match in SERIAL_NUMBER_PATTERN.finditer(text):
serial = match.group(1).strip()
if serial not in found:
found.append(serial)
return found
```
- [ ] **Step 9: Run tests to verify they pass**
Run: `pytest tests/test_invoice_logic.py -v`
Expected: all tests PASS.
- [ ] **Step 10: Wire the fallback extraction and minimal display into `main.py`**
Find:
```python
if final_data.get("recipient_email") == VERKAEUFER["email"]:
final_data["recipient_email"] = ""
final_data["items"] = extract_invoice_items_pure_python(pdf_text)
st.session_state.invoice_counter += 1
st.session_state.extracted_data = final_data
```
Replace with:
```python
if final_data.get("recipient_email") == VERKAEUFER["email"]:
final_data["recipient_email"] = ""
final_data["items"] = extract_invoice_items_pure_python(pdf_text)
# IT-Systemhaus-Regex-Fallbacks: nur ergaenzen, wenn das LLM die entsprechenden
# Felder nicht bereits geliefert hat.
if not final_data.get("service_start_date") and not final_data.get("service_end_date"):
fallback_start, fallback_end = extract_service_period(pdf_text)
if fallback_start and fallback_end:
final_data["service_start_date"] = fallback_start
final_data["service_end_date"] = fallback_end
final_data["contract_type"] = detect_contract_type(pdf_text)
final_data["hardware_serials"] = extract_hardware_serial_numbers(pdf_text)
st.session_state.invoice_counter += 1
st.session_state.extracted_data = final_data
```
Add `extract_service_period`, `detect_contract_type`, `extract_hardware_serial_numbers` to the `from invoice_logic import (...)` block in `main.py`.
- [ ] **Step 11: Add a minimal, non-intrusive read-only display**
Find the line `st.caption("Mit :red[*] markierte Felder sind EN16931-Pflichtfelder und müssen für einen gültigen Export befüllt sein.")` and insert directly after it:
```python
if data.get("contract_type"):
st.caption(f"🏷️ Erkannter Vertragstyp: **{data['contract_type']}**")
if data.get("hardware_serials"):
st.caption(f"🔧 Erkannte Seriennummern: {', '.join(data['hardware_serials'])}")
```
- [ ] **Step 12: Manual smoke test**
Run: `streamlit run main.py`. Upload a sample PDF containing text like `Leistungszeitraum: 01.03.2026 bis 31.03.2026`, `Managed Services`, and `S/N: ABC-12345`. Extract data and confirm the two new caption lines appear with the correct values, and that a `3,00 Personentage` line item is parsed with `unit == "Personentage"`.
- [ ] **Step 13: Commit**
```bash
git add invoice_logic.py main.py tests/test_invoice_logic.py
git commit -m "feat: extend regex parser for IT-Systemhaus service periods, contract types, and serial numbers"
```
---
### Task 6: End-to-end integration test (all four features combined)
**Files:**
- Modify: `tests/test_invoice_logic.py`
**Interfaces:**
- Consumes: everything produced in Tasks 1–5.
- [ ] **Step 1: Write the integration test**
Add to `tests/test_invoice_logic.py`:
```python
def test_full_invoice_with_all_phase1_features_combined():
"""Realistische IT-Systemhaus-Rechnung: gemischte Steuersaetze, Rabatt, Skonto,
dynamischer Zahlungsmittelcode und IT-spezifische Abrechnungseinheiten - alles zusammen."""
items = [
{"name": "Managed-Services-Pauschale", "description": "", "quantity": 1.0, "unit": "Monat(e)",
"unit_price": 500.0, "total_amount": 500.0, "vat_rate_percent": 19.0},
{"name": "Fachliteratur", "description": "", "quantity": 1.0, "unit": "Stk.",
"unit_price": 40.0, "total_amount": 40.0, "vat_rate_percent": 7.0},
]
totals = compute_invoice_totals(items, discount_percent=5.0)
data = _single_rate_invoice()
data["items"] = items
data["tax_buckets"] = totals["tax_buckets"]
data["line_net_total"] = totals["line_net_total"]
data["allowance_amount"] = totals["allowance_amount"]
data["allowance_reason"] = "Neukundenrabatt"
data["total_net_amount"] = totals["tax_basis_total"]
data["total_vat_amount"] = totals["total_vat_amount"]
data["total_gross_amount"] = totals["total_gross_amount"]
data["skonto_percent"] = 2.0
data["skonto_days"] = "14"
data["payment_means_code"] = "31"
ubl_xml = generate_compliant_xrechnung(data)
ubl_root = ET.fromstring(ubl_xml.encode("utf-8"))
ubl_ns = {"cbc": "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2",
"cac": "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"}
assert len(ubl_root.findall(".//cac:TaxTotal/cac:TaxSubtotal", ubl_ns)) == 2
assert ubl_root.find(".//cac:PaymentMeans/cbc:PaymentMeansCode", ubl_ns).text == "31"
ubl_allowances = ubl_root.findall(".//cac:AllowanceCharge", ubl_ns)
assert len(ubl_allowances) == 2 # ein Rabatt-Block je Steuersatz (19% und 7%)
assert abs(sum(float(a.find("cbc:Amount", ubl_ns).text) for a in ubl_allowances) - 27.0) < 0.001
assert "Skonto" in ubl_root.find(".//cac:PaymentTerms/cbc:Note", ubl_ns).text
cii_xml = generate_compliant_zugferd_cii(data)
cii_root = ET.fromstring(cii_xml.encode("utf-8"))
cii_ns = {"ram": "urn:un:unece:uncefact:data:standard:ReusableAggregateBusinessInformationEntity:100"}
assert len(cii_root.findall(".//ram:ApplicableHeaderTradeSettlement/ram:ApplicableTradeTax", cii_ns)) == 2
assert cii_root.find(".//ram:SpecifiedTradeSettlementPaymentMeans/ram:TypeCode", cii_ns).text == "31"
cii_allowances = cii_root.findall(".//ram:SpecifiedTradeAllowanceCharge", cii_ns)
assert len(cii_allowances) == 2 # ein Rabatt-Block je Steuersatz (19% und 7%)
assert abs(sum(float(a.find("ram:ActualAmount", cii_ns).text) for a in cii_allowances) - 27.0) < 0.001
```
- [ ] **Step 2: Run the full test suite**
Run: `pytest tests/test_invoice_logic.py -v`
Expected: every test in the file PASSES (this includes all Task 1–5 tests plus this new integration test — nothing regressed).
- [ ] **Step 3: Full-file syntax and import sanity check**
Run: `python -c "import ast; ast.parse(open('main.py', encoding='utf-8').read()); print('main.py OK')"`
Run: `python -c "import ast; ast.parse(open('invoice_logic.py', encoding='utf-8').read()); print('invoice_logic.py OK')"`
Expected: both print `OK`.
- [ ] **Step 4: Manual end-to-end smoke test in the running app**
Run: `streamlit run main.py`. Upload a real sample invoice, add a second position with a different VAT rate, set a 5% discount and a 2%/14-day Skonto, choose payment means `31 - SEPA-Lastschrift`, and export both XRechnung XML and the ZUGFeRD PDF. Confirm both downloads succeed without errors and the XML contains the expected multi-tax, allowance, payment-terms, and payment-means elements.
- [ ] **Step 5: Commit**
```bash
git add tests/test_invoice_logic.py
git commit -m "test: add end-to-end integration test covering all Phase 1 features together"
```
---
## Self-Review Notes
- **Spec coverage:** Multi-tax (Task 3) ✅, IT-Systemhaus regex extensions (Task 5) ✅, Skonto + globaler Rabatt (Task 4) ✅, dynamic payment means code (Task 2) ✅. Testability prerequisite (Task 1) and final integration proof (Task 6) round out the plan.
- **No placeholders:** every step contains complete, runnable code and exact `pytest`/`python` commands with expected results.
- **Type/name consistency verified:** `compute_invoice_totals` (Task 3) is defined with `discount_percent`/`discount_amount` parameters from the start so Task 4 only adds the UI wiring, not a signature change. `tax_buckets`/`allowance_amount`/`line_net_total` keys are used identically across Tasks 3, 4, and 6. `PAYMENT_MEANS_CODES`, `format_skonto_text`, `extract_service_period`, `detect_contract_type`, `extract_hardware_serial_numbers` are each defined exactly once (Tasks 2, 4, 5) and consumed with matching names everywhere else.