leocrm/app/plugins/builtins/report_generator/einvoice.py

"""E-Invoice format layer — EN16931 / XRechnung CII XML generation (Phase L5).

This module is deliberately a pure FORMAT layer: it turns validated invoice
data (plain dicts) into Cross Industry Invoice XML. It knows nothing about
how invoices are stored — the future sales module will own the entities
and dock via the ``einvoice_data()`` contract hook
(``resolve_einvoice_data``).

Profiles:
- ``en16931``: GuidelineID ``urn:cen.eu:en16931:2017``
- ``xrechnung``: ``...#compliant#urn:xoev-de:kosit:standard:xrechnung_3.0``

All monetary math uses ``Decimal`` quantized to 2 places (commercial
rounding) so header sums are consistent with line tax calculations.
XML escaping is delegated to ElementTree — no manual string concatenation.
"""

from __future__ import annotations

import uuid
from dataclasses import dataclass, field
from datetime import date
from decimal import ROUND_HALF_UP, Decimal
from xml.etree import ElementTree as ET

# ─── CII namespaces (XRechnung 3.0 / D16B) ──────────────────────────────────

NS_RSM = "urn:un:unece:uncefact:data:standard:CrossIndustryInvoice:100"
NS_RAM = "urn:un:unece:uncefact:data:standard:ReusableAggregateBusinessInformationEntity:100"
NS_UDT = "urn:un:unece:uncefact:data:standard:UnqualifiedDataType:100"

GUIDELINE_EN16931 = "urn:cen.eu:en16931:2017"
GUIDELINE_XRECHNUNG = (
    "urn:cen.eu:en16931:2017#compliant#urn:xoev-de:kosit:standard:xrechnung_3.0"
)

_CENT = Decimal("0.01")


def _money(value) -> Decimal:
    """Quantize to 2 decimal places (commercial rounding, BR-CL-16)."""
    return Decimal(str(value)).quantize(_CENT, rounding=ROUND_HALF_UP)


def _fmt_date(value: str) -> str:
    """ISO date (YYYY-MM-DD) → CII format 102 (YYYYMMDD). Raises ValueError."""
    parsed = date.fromisoformat(str(value).strip())
    return parsed.strftime("%Y%m%d")


# ─── Validation ──────────────────────────────────────────────────────────────


@dataclass
class EInvoiceValidationError(ValueError):
    """Raised when mandatory EN16931 fields are missing or invalid.

    ``missing`` carries human-readable entries including BT/BG field codes
    so the API can surface exactly which business terms fail.
    """

    missing: list[str] = field(default_factory=list)


def _non_empty(data: dict, key: str) -> str | None:
    value = data.get(key)
    if isinstance(value, str) and value.strip():
        return value.strip()
    return None


def validate_einvoice_data(data: dict) -> None:
    """Validate mandatory EN16931 business terms (subset enforced today).

    Raises EInvoiceValidationError with the collected BT/BG entries.
    """
    if not isinstance(data, dict):
        raise EInvoiceValidationError(missing=["payload: Objekt erwartet"])

    missing: list[str] = []

    if not _non_empty(data, "invoice_number"):
        missing.append("BT-1: Rechnungsnummer fehlt")
    if not _non_empty(data, "issue_date"):
        missing.append("BT-2: Rechnungsdatum fehlt")
    else:
        try:
            _fmt_date(data["issue_date"])
        except ValueError:
            missing.append("BT-2: Rechnungsdatum muss YYYY-MM-DD sein")
    if not _non_empty(data, "type_code"):
        missing.append("BT-3: Rechnungsart fehlt")
    if not _non_empty(data, "currency"):
        missing.append("BT-5: Währung fehlt")

    # Seller: name + at least one tax registration (BT-31 USt-IdNr or
    # BT-32 Steuernummer — BR-DE-16 for German invoices).
    if not _non_empty(data, "seller_name"):
        missing.append("BT-27: Verkäufername fehlt")
    seller_vat = _non_empty(data, "seller_vat_id")
    seller_tax = _non_empty(data, "seller_tax_id")
    if not seller_vat and not seller_tax:
        missing.append("BT-31: USt-IdNr. oder BT-32: Steuernummer des Verkäufers fehlt")

    if not _non_empty(data, "buyer_name"):
        missing.append("BT-10: Empfängername fehlt")

    if data.get("due_date"):
        try:
            _fmt_date(str(data["due_date"]))
        except ValueError:
            missing.append("BT-9: Fälligkeitsdatum muss YYYY-MM-DD sein")

    lines = data.get("line_items")
    if not isinstance(lines, list) or len(lines) == 0:
        missing.append("BG-25: mindestens eine Rechnungsposition erforderlich")
    else:
        for i, line in enumerate(lines, start=1):
            if not isinstance(line, dict):
                missing.append(f"Position {i}: Objekt erwartet")
                continue
            if not _non_empty(line, "name"):
                missing.append(f"BT-126: Positionsname fehlt (Position {i})")
            try:
                Decimal(str(line.get("unit_net_price", "0")))
            except Exception:  # noqa: BLE001
                missing.append(f"BT-146: Einzelpreis ungültig (Position {i})")

    if missing:
        raise EInvoiceValidationError(missing=missing)


# ─── XML rendering ───────────────────────────────────────────────────────────


def _sub(parent: ET.Element, tag: str, text: str | None = None, **attrib) -> ET.Element:
    el = ET.SubElement(parent, f"{{{NS_RAM}}}{tag}", {k: str(v) for k, v in attrib.items()})
    if text is not None:
        el.text = str(text)
    return el


def _date_el(parent: ET.Element, tag: str, iso_date: str) -> None:
    wrapper = _sub(parent, tag)
    dt = ET.SubElement(wrapper, f"{{{NS_UDT}}}DateTimeString")
    dt.set("format", "102")
    dt.text = _fmt_date(iso_date)


def _address(parent: ET.Element, addr: dict | None) -> None:
    addr = addr or {}
    postal = _sub(parent, "PostalTradeAddress")
    if addr.get("street"):
        _sub(postal, "LineOne", str(addr["street"]))
    if addr.get("postal_code"):
        _sub(postal, "PostcodeCode", str(addr["postal_code"]))
    if addr.get("city"):
        _sub(postal, "CityName", str(addr["city"]))
    _sub(postal, "CountryID", str(addr.get("country") or "DE"))


def _line_sums(data: dict) -> tuple[list[dict], Decimal, dict[str, list[Decimal]]]:
    """Compute per-line and total sums.

    Returns (line_amounts, line_total, taxes) where line_amounts[i] is the
    net amount of line i (supplied ``line_net_amount`` wins over
    quantity × unit price — discounts already applied), and taxes maps
    vat_rate → [basis, tax_amount] aggregated for the header breakdown.
    """
    line_amounts: list[Decimal] = []
    line_total = Decimal("0.00")
    taxes: dict[str, list[Decimal]] = {}

    for line in data.get("line_items") or []:
        qty = Decimal(str(line.get("quantity", 1)))
        price = _money(line.get("unit_net_price", 0))
        if line.get("line_net_amount") is not None:
            net = _money(line["line_net_amount"])
        else:
            net = _money(qty * price)
        line_amounts.append(net)
        line_total = _money(line_total + net)

        rate = str(line.get("vat_rate", 0))
        basis, tax = taxes.get(rate, [Decimal("0.00"), Decimal("0.00")])
        tax_amount = _money(net * Decimal(rate) / Decimal(100))
        taxes[rate] = [_money(basis + net), _money(tax + tax_amount)]

    return line_amounts, _money(line_total), taxes


def render_einvoice_xml(data: dict) -> str:
    """Render validated invoice data to CII XML (UTF-8, declaration header)."""
    validate_einvoice_data(data)

    profile = str(data.get("profile") or "en16931").lower()
    guideline = GUIDELINE_XRECHNUNG if profile == "xrechnung" else GUIDELINE_EN16931

    ET.register_namespace("rsm", NS_RSM)
    ET.register_namespace("ram", NS_RAM)
    ET.register_namespace("udt", NS_UDT)
    root = ET.Element(f"{{{NS_RSM}}}CrossIndustryInvoice")

    # ── ExchangedDocumentContext (BT-24) ──
    ctx = ET.SubElement(root, f"{{{NS_RSM}}}ExchangedDocumentContext")
    guideline_param = _sub(ctx, "GuidelineSpecifiedDocumentContextParameter")
    _sub(guideline_param, "ID", guideline)

    # ── ExchangedDocument (BT-1..BT-22) ──
    doc = ET.SubElement(root, f"{{{NS_RSM}}}ExchangedDocument")
    _sub(doc, "ID", data["invoice_number"])
    _sub(doc, "TypeCode", data["type_code"])
    _date_el(doc, "IssueDateTime", data["issue_date"])
    if data.get("note"):
        note = _sub(doc, "IncludedNote")
        _sub(note, "Content", str(data["note"]))

    # ── SupplyChainTradeTransaction ──
    txn = ET.SubElement(root, f"{{{NS_RSM}}}SupplyChainTradeTransaction")

    line_amounts, line_total, taxes = _line_sums(data)
    for idx, (line, net) in enumerate(zip(data["line_items"], line_amounts, strict=True), start=1):
        item = ET.SubElement(txn, f"{{{NS_RSM}}}IncludedSupplyChainTradeLineItem")
        line_doc = _sub(item, "AssociatedDocumentLineDocument")
        _sub(line_doc, "LineID", str(idx))

        product = _sub(item, "SpecifiedTradeProduct")
        _sub(product, "Name", line["name"])

        agreement = _sub(item, "SpecifiedLineTradeAgreement")
        net_price = _sub(agreement, "NetPriceProductTradePrice")
        _sub(net_price, "ChargeAmount", _money(line.get("unit_net_price", 0)))

        delivery = _sub(item, "SpecifiedLineTradeDelivery")
        _sub(delivery, "BilledQuantity", Decimal(str(line.get("quantity", 1))), unitCode=line.get("unit") or "HUR")

        settlement = _sub(item, "SpecifiedLineTradeSettlement")
        line_tax = _sub(settlement, "ApplicableTradeTax")
        _sub(line_tax, "TypeCode", "VAT")
        rate = Decimal(str(line.get("vat_rate", 0)))
        _sub(line_tax, "RateApplicablePercent", rate)
        _sub(line_tax, "BasisAmount", net)
        _sub(line_tax, "CalculatedAmount", _money(net * rate / Decimal(100)))
        line_sum = _sub(settlement, "SpecifiedTradeSettlementLineMonetarySummation")
        _sub(line_sum, "LineTotalAmount", net)

    # ── ApplicableHeaderTradeAgreement ──
    agreement_h = ET.SubElement(txn, f"{{{NS_RAM}}}ApplicableHeaderTradeAgreement")
    if data.get("buyer_reference"):
        _sub(agreement_h, "BuyerReference", str(data["buyer_reference"]))

    seller = _sub(agreement_h, "SellerTradeParty")
    _sub(seller, "Name", data["seller_name"])
    _address(seller, data.get("seller_address"))
    tax_reg = _sub(seller, "SpecifiedTaxRegistration")
    if data.get("seller_vat_id"):
        _sub(tax_reg, "ID", str(data["seller_vat_id"]), schemeID="VA")
    elif data.get("seller_tax_id"):
        _sub(tax_reg, "ID", str(data["seller_tax_id"]), schemeID="FC")

    buyer = _sub(agreement_h, "BuyerTradeParty")
    _sub(buyer, "Name", data["buyer_name"])
    _address(buyer, data.get("buyer_address"))

    # ── ApplicableHeaderTradeDelivery (BT-72) ──
    delivery_h = ET.SubElement(txn, f"{{{NS_RAM}}}ApplicableHeaderTradeDelivery")
    event = _sub(delivery_h, "ActualDeliverySupplyChainEvent")
    _date_el(event, "OccurrenceDateTime", data.get("delivery_date") or data["issue_date"])

    # ── ApplicableHeaderTradeSettlement ──
    settlement_h = ET.SubElement(txn, f"{{{NS_RAM}}}ApplicableHeaderTradeSettlement")
    currency = str(data["currency"])
    _sub(settlement_h, "InvoiceCurrencyCode", currency)

    if data.get("payment_means_code"):
        means = _sub(settlement_h, "SpecifiedTradeSettlementPaymentMeans")
        _sub(means, "TypeCode", str(data["payment_means_code"]))

    # header tax breakdown per VAT rate (BG-23)
    tax_total = Decimal("0.00")
    for rate in sorted(taxes, key=Decimal):
        basis, amount = taxes[rate]
        header_tax = _sub(settlement_h, "ApplicableTradeTax")
        _sub(header_tax, "TypeCode", "VAT")
        _sub(header_tax, "BasisAmount", basis)
        _sub(header_tax, "CalculatedAmount", amount)
        _sub(header_tax, "RateApplicablePercent", Decimal(rate))
        tax_total = _money(tax_total + amount)

    grand_total = _money(line_total + tax_total)

    terms = _sub(settlement_h, "SpecifiedTradePaymentTerms")
    if data.get("payment_terms_text"):
        _sub(terms, "Description", str(data["payment_terms_text"]))
    if data.get("due_date"):
        _date_el(terms, "DueDateDateTime", str(data["due_date"]))

    sums = _sub(settlement_h, "SpecifiedTradeSettlementHeaderMonetarySummation")
    _sub(sums, "LineTotalAmount", line_total)
    _sub(sums, "TaxTotalAmount", tax_total, currencyID=currency)
    _sub(sums, "GrandTotalAmount", grand_total)
    _sub(sums, "DuePayableAmount", grand_total)

    return ET.tostring(root, encoding="unicode", xml_declaration=False)


# ─── Contract resolution (sales module docking point) ───────────────────────


async def resolve_einvoice_data(
    db,
    tenant_id: uuid.UUID,
    entity_type: str,
    entity_id: uuid.UUID,
) -> dict | None:
    """Resolve invoice data for an entity via the ``einvoice_data()`` contract.

    The future sales module will expose::

        async def einvoice_data(db, tenant_id, entity_id, entity_type) -> dict

    Returning validated-shaped invoice data (same fields as the inline
    render endpoint). No contribution → None (caller answers 404).
    """
    from app.plugins.builtins.contracts import get_contract_registry
    from app.plugins.registry import get_registry

    for plugin_name in get_registry().list_discovered():
        contract = get_contract_registry().get_contract(plugin_name)
        fn = getattr(contract, "einvoice_data", None)
        if fn is None:
            continue
        try:
            result = await fn(db, tenant_id, entity_id, entity_type)
        except Exception:  # noqa: BLE001 — broken contribution must not 500
            continue
        if result:
            return result
    return None