358 lines
14 KiB
Python
358 lines
14 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
Stripe Checkout service for billing credit top-ups.
|
|
Creates Checkout Sessions for redirect-based payment flow.
|
|
|
|
CH-Treuhand-Konformitaet (Issue 2026-04-20):
|
|
- Bei jedem Checkout wird ein Stripe-Customer mit der Mandanten-Rechnungsadresse
|
|
angelegt/aktualisiert (Name, Adresse, E-Mail, optional UID/MWST-Nr).
|
|
- Auf dem Checkout wird `invoice_creation` aktiviert, damit Stripe automatisch
|
|
eine Rechnung mit Status `paid` erzeugt (statt nur einer Quittung). Die Rechnung
|
|
enthaelt die volle Empfaengeradresse und einen Footer-Hinweis "bezahlt via
|
|
Kreditkarte am ...".
|
|
- MWST 8.1% (CH) wird ueber `automatic_tax: enabled=true` aufgeschlagen, sofern
|
|
Stripe Tax fuer den Account aktiviert ist (siehe wiki/d-guides/stripe-ch-vat.md).
|
|
Alternativ kann ueber APP_CONFIG `STRIPE_TAX_RATE_ID_CH_VAT` ein vordefinierter
|
|
Tax-Rate angehaengt werden (8.1% inclusive=false).
|
|
"""
|
|
|
|
import logging
|
|
from typing import Any, Dict, List, Optional
|
|
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
|
|
|
|
from modules.shared.configuration import APP_CONFIG
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Server-side allowed amounts in CHF - never trust client
|
|
ALLOWED_AMOUNTS_CHF = [10, 25, 50, 100, 250, 500]
|
|
|
|
|
|
def _str(value: Any) -> Optional[str]:
|
|
"""Trim ``value`` to a non-empty string or return ``None``."""
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, str):
|
|
value = str(value)
|
|
trimmed = value.strip()
|
|
return trimmed or None
|
|
|
|
|
|
def _buildStripeAddress(invoiceAddress: Optional[Dict[str, Any]]) -> Optional[Dict[str, str]]:
|
|
"""Translate the structured Mandate invoice fields into a Stripe ``address`` object.
|
|
|
|
Returns ``None`` when the address is not complete enough for Stripe
|
|
(line1 + city are the practical minimum); the caller then falls back
|
|
to a guest checkout where Stripe collects the address from the user.
|
|
"""
|
|
if not invoiceAddress:
|
|
return None
|
|
address: Dict[str, Optional[str]] = {
|
|
"line1": _str(invoiceAddress.get("line1")),
|
|
"line2": _str(invoiceAddress.get("line2")),
|
|
"postal_code": _str(invoiceAddress.get("postalCode")),
|
|
"city": _str(invoiceAddress.get("city")),
|
|
"state": _str(invoiceAddress.get("state")),
|
|
"country": (_str(invoiceAddress.get("country")) or "CH").upper()[:2],
|
|
}
|
|
cleaned = {k: v for k, v in address.items() if v}
|
|
if not cleaned.get("line1") or not cleaned.get("city"):
|
|
return None
|
|
return cleaned
|
|
|
|
|
|
def _buildStripeTaxIdData(invoiceAddress: Optional[Dict[str, Any]]) -> List[Dict[str, str]]:
|
|
"""Translate UID/MWST-Nr into Stripe ``tax_id_data`` (CHE -> ``ch_vat``)."""
|
|
if not invoiceAddress:
|
|
return []
|
|
vat = _str(invoiceAddress.get("vatNumber"))
|
|
if not vat:
|
|
return []
|
|
upper = vat.upper()
|
|
if upper.startswith("CHE"):
|
|
return [{"type": "ch_vat", "value": vat[:50]}]
|
|
if upper.startswith("LI"):
|
|
return [{"type": "li_uid", "value": vat[:50]}]
|
|
if len(upper) >= 4 and upper[:2].isalpha() and any(c.isdigit() for c in upper):
|
|
return [{"type": "eu_vat", "value": vat[:50]}]
|
|
logger.info("Skipping unrecognized invoice VAT number format: %s", vat)
|
|
return []
|
|
|
|
|
|
def _ensureStripeCustomer(
|
|
mandateId: str,
|
|
mandateLabel: str,
|
|
invoiceAddress: Optional[Dict[str, Any]],
|
|
settings: Optional[Dict[str, Any]],
|
|
billingInterface,
|
|
) -> Optional[str]:
|
|
"""Create or update the Stripe Customer for the given mandate.
|
|
|
|
Maps the structured invoice fields from ``Mandate`` to:
|
|
- ``customer.name`` (companyName, fallback mandateLabel)
|
|
- ``customer.email`` (if invoiceEmail is set)
|
|
- ``customer.address`` (line1/line2/postal_code/city/state/country)
|
|
- ``customer.shipping`` (mirrors address with contactName, when set)
|
|
- ``customer.tax_id_data`` (UID/MWST-Nr; only on create -- Stripe API
|
|
does not allow modifying tax_id_data via Customer.modify, the
|
|
existing tax IDs would have to be replaced via tax_ids.create).
|
|
|
|
Returns the Stripe customer id (or ``None`` if creation failed and we
|
|
should fall back to a guest checkout).
|
|
"""
|
|
from modules.shared.stripeClient import getStripeClient, stripeToDict
|
|
stripe = getStripeClient()
|
|
|
|
address = _buildStripeAddress(invoiceAddress)
|
|
name = _str((invoiceAddress or {}).get("companyName")) or mandateLabel
|
|
email = _str((invoiceAddress or {}).get("email"))
|
|
contactName = _str((invoiceAddress or {}).get("contactName"))
|
|
vatNumber = _str((invoiceAddress or {}).get("vatNumber"))
|
|
|
|
metadata = {
|
|
"mandateId": mandateId,
|
|
"mandateLabel": mandateLabel,
|
|
}
|
|
if vatNumber:
|
|
metadata["vatNumber"] = vatNumber
|
|
if contactName:
|
|
metadata["contactName"] = contactName
|
|
|
|
customerId = (settings or {}).get("stripeCustomerId") if settings else None
|
|
|
|
payload: Dict[str, Any] = {
|
|
"name": name,
|
|
"metadata": metadata,
|
|
}
|
|
if email:
|
|
payload["email"] = email
|
|
if address:
|
|
payload["address"] = address
|
|
if contactName:
|
|
payload["shipping"] = {
|
|
"name": f"{contactName} ({name})" if name else contactName,
|
|
"address": address,
|
|
}
|
|
|
|
try:
|
|
if customerId:
|
|
stripe.Customer.modify(customerId, **payload)
|
|
else:
|
|
taxIdData = _buildStripeTaxIdData(invoiceAddress)
|
|
createPayload = dict(payload)
|
|
if taxIdData:
|
|
createPayload["tax_id_data"] = taxIdData
|
|
customer = stripe.Customer.create(**createPayload)
|
|
customerId = stripeToDict(customer).get("id") or getattr(customer, "id", None)
|
|
if customerId and billingInterface is not None and settings is not None:
|
|
try:
|
|
billingInterface.updateSettings(settings["id"], {"stripeCustomerId": customerId})
|
|
except Exception as ex:
|
|
logger.warning("Failed to persist stripeCustomerId for mandate %s: %s", mandateId, ex)
|
|
return customerId
|
|
except Exception as ex:
|
|
logger.error("Stripe Customer create/update failed for mandate %s: %s", mandateId, ex)
|
|
return customerId # may be None, falls back to guest checkout
|
|
|
|
|
|
def _resolveCheckoutTaxRates() -> List[str]:
|
|
"""Return tax-rate IDs to apply manually if Stripe Tax is not enabled."""
|
|
raw = APP_CONFIG.get("STRIPE_TAX_RATE_ID_CH_VAT") or ""
|
|
return [r.strip() for r in str(raw).split(",") if r.strip()]
|
|
|
|
|
|
def _isAutomaticTaxEnabled() -> bool:
|
|
"""Read STRIPE_AUTOMATIC_TAX_ENABLED as a real boolean.
|
|
|
|
APP_CONFIG._loadEnv stores all .env values as raw strings, so a naive
|
|
``bool(APP_CONFIG.get(key, False))`` would return True for the strings
|
|
``"false"``, ``"0"`` or ``"no"`` (any non-empty string is truthy in
|
|
Python). We therefore parse the value explicitly.
|
|
"""
|
|
raw = APP_CONFIG.get("STRIPE_AUTOMATIC_TAX_ENABLED", False)
|
|
if isinstance(raw, bool):
|
|
return raw
|
|
if raw is None:
|
|
return False
|
|
return str(raw).strip().lower() in {"true", "1", "yes", "on"}
|
|
|
|
|
|
def _normalizeReturnUrl(returnUrl: str) -> str:
|
|
"""
|
|
Validate and normalize an absolute frontend return URL.
|
|
|
|
Allowed examples:
|
|
- https://nyla.poweron-center.net/billing/transactions
|
|
- https://nyla-int.poweron-center.net/billing/transactions?tab=overview
|
|
"""
|
|
if not returnUrl:
|
|
raise ValueError("returnUrl is required")
|
|
|
|
parsed = urlsplit(returnUrl.strip())
|
|
|
|
if parsed.scheme not in ("http", "https"):
|
|
raise ValueError("returnUrl must use http or https")
|
|
|
|
if not parsed.netloc:
|
|
raise ValueError("returnUrl must contain a host")
|
|
|
|
if parsed.username or parsed.password:
|
|
raise ValueError("returnUrl must not contain credentials")
|
|
|
|
query_items = [
|
|
(key, value)
|
|
for key, value in parse_qsl(parsed.query, keep_blank_values=True)
|
|
if key not in {"success", "canceled", "session_id"}
|
|
]
|
|
normalized_query = urlencode(query_items, doseq=True)
|
|
normalized_path = parsed.path or "/"
|
|
|
|
# Keep scheme + host + path + query, strip fragment for deterministic redirects.
|
|
return urlunsplit((parsed.scheme, parsed.netloc, normalized_path, normalized_query, ""))
|
|
|
|
|
|
def create_checkout_session(
|
|
mandate_id: str,
|
|
user_id: Optional[str],
|
|
amount_chf: float,
|
|
return_url: str,
|
|
mandate_label: Optional[str] = None,
|
|
invoice_address: Optional[Dict[str, Any]] = None,
|
|
settings: Optional[Dict[str, Any]] = None,
|
|
billing_interface=None,
|
|
) -> str:
|
|
"""
|
|
Create a Stripe Checkout Session for credit top-up.
|
|
|
|
Amount and currency are validated server-side. The client-provided amount
|
|
must match an allowed preset.
|
|
|
|
CH-Treuhand-Konformitaet:
|
|
- Reuses or creates a Stripe Customer with the mandate's invoice address.
|
|
- Activates `invoice_creation` so Stripe issues a proper invoice (status
|
|
`paid`) carrying the full recipient address and VAT.
|
|
- Adds 8.1% Swiss VAT either via Stripe Tax (`automatic_tax`) or via a
|
|
manually configured `STRIPE_TAX_RATE_ID_CH_VAT` tax-rate.
|
|
|
|
Args:
|
|
mandate_id: Target mandate ID
|
|
user_id: Target user ID for audit trail (optional)
|
|
amount_chf: Amount in CHF (must be in ALLOWED_AMOUNTS_CHF)
|
|
return_url: Absolute frontend URL used for Stripe success/cancel redirects
|
|
mandate_label: Human-readable mandate name (used as Customer name fallback)
|
|
invoice_address: Dict assembled in routeBilling from the structured
|
|
``Mandate.invoice*`` fields. Recognised keys: companyName,
|
|
contactName, email, line1, line2, postalCode, city, state,
|
|
country, vatNumber. Pass ``None`` for guest checkout (Stripe
|
|
then collects line1/postal_code/city itself via
|
|
``billing_address_collection: required``).
|
|
settings: Mandate billing settings (carries `stripeCustomerId`, `id`)
|
|
billing_interface: Billing interface (for persisting stripeCustomerId)
|
|
|
|
Returns:
|
|
Stripe Checkout Session URL for redirect
|
|
|
|
Raises:
|
|
ValueError: If amount is invalid
|
|
"""
|
|
if amount_chf not in ALLOWED_AMOUNTS_CHF:
|
|
raise ValueError(
|
|
f"Invalid amount {amount_chf} CHF. Allowed: {ALLOWED_AMOUNTS_CHF}"
|
|
)
|
|
|
|
from modules.shared.stripeClient import getStripeClient
|
|
stripe = getStripeClient()
|
|
|
|
base_return_url = _normalizeReturnUrl(return_url)
|
|
query_separator = "&" if "?" in base_return_url else "?"
|
|
success_url = f"{base_return_url}{query_separator}success=true&session_id={{CHECKOUT_SESSION_ID}}"
|
|
cancel_url = f"{base_return_url}{query_separator}canceled=true"
|
|
|
|
amount_cents = int(round(amount_chf * 100))
|
|
|
|
metadata = {
|
|
"mandateId": mandate_id,
|
|
"amountChf": str(amount_chf),
|
|
}
|
|
if user_id:
|
|
metadata["userId"] = user_id
|
|
|
|
customerId = _ensureStripeCustomer(
|
|
mandateId=mandate_id,
|
|
mandateLabel=mandate_label or mandate_id,
|
|
invoiceAddress=invoice_address,
|
|
settings=settings,
|
|
billingInterface=billing_interface,
|
|
)
|
|
|
|
taxRateIds = _resolveCheckoutTaxRates()
|
|
autoTaxEnabled = _isAutomaticTaxEnabled()
|
|
|
|
line_item: Dict[str, Any] = {
|
|
"price_data": {
|
|
"currency": "chf",
|
|
"unit_amount": amount_cents,
|
|
"product_data": {
|
|
"name": "Guthaben aufladen",
|
|
"description": "AI Service Guthaben (CHF) inkl. MWST 8.1%",
|
|
},
|
|
},
|
|
"quantity": 1,
|
|
}
|
|
if taxRateIds and not autoTaxEnabled:
|
|
line_item["tax_rates"] = taxRateIds
|
|
|
|
invoice_data: Dict[str, Any] = {
|
|
"description": f"Guthaben-Aufladung {amount_chf:.2f} CHF (Mandant: {mandate_label or mandate_id})",
|
|
"metadata": metadata,
|
|
"footer": (
|
|
"Diese Rechnung wurde bereits via Kreditkarte bezahlt. "
|
|
"MWST-Nr. PowerOn: siehe Stripe-Rechnungs-Template. "
|
|
"Bei Fragen: billing@poweron-center.net"
|
|
),
|
|
}
|
|
customFields: List[Dict[str, str]] = []
|
|
if invoice_address:
|
|
vat = _str(invoice_address.get("vatNumber"))
|
|
if vat:
|
|
customFields.append({"name": "UID-Nr. Empfaenger", "value": vat[:30]})
|
|
contactName = _str(invoice_address.get("contactName"))
|
|
if contactName:
|
|
customFields.append({"name": "z. H.", "value": contactName[:30]})
|
|
if customFields:
|
|
invoice_data["custom_fields"] = customFields[:4]
|
|
|
|
sessionPayload: Dict[str, Any] = {
|
|
"mode": "payment",
|
|
"line_items": [line_item],
|
|
"success_url": success_url,
|
|
"cancel_url": cancel_url,
|
|
"metadata": metadata,
|
|
"invoice_creation": {
|
|
"enabled": True,
|
|
"invoice_data": invoice_data,
|
|
},
|
|
}
|
|
if customerId:
|
|
sessionPayload["customer"] = customerId
|
|
sessionPayload["customer_update"] = {"address": "auto", "name": "auto", "shipping": "auto"}
|
|
else:
|
|
sessionPayload["billing_address_collection"] = "required"
|
|
if invoice_address and _str(invoice_address.get("email")):
|
|
sessionPayload["customer_email"] = _str(invoice_address.get("email"))
|
|
if autoTaxEnabled:
|
|
sessionPayload["automatic_tax"] = {"enabled": True}
|
|
|
|
session = stripe.checkout.Session.create(**sessionPayload)
|
|
|
|
if not session or not session.url:
|
|
raise ValueError("Stripe Checkout Session creation failed")
|
|
|
|
logger.info(
|
|
f"Created Stripe Checkout Session {session.id} for mandate {mandate_id}, "
|
|
f"amount {amount_chf} CHF, customer={customerId or 'guest'}, "
|
|
f"taxRates={taxRateIds}, autoTax={autoTaxEnabled}"
|
|
)
|
|
|
|
return session.url
|