140 lines
5 KiB
Python
140 lines
5 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
When the shared mandate pool (PREPAY_MANDATE) is exhausted, notify billing contacts.
|
|
|
|
Recipients: BillingSettings.notifyEmails for the mandate (configure as mandate owner / finance).
|
|
Emails are throttled per mandate to avoid spam (one notification per cooldown window).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import html
|
|
import logging
|
|
import time
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from modules.datamodels.datamodelMessaging import MessagingChannel
|
|
from modules.interfaces.interfaceDbBilling import getInterface as getBillingInterface
|
|
from modules.interfaces.interfaceMessaging import getInterface as getMessagingInterface
|
|
from modules.security.rootAccess import getRootUser
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# mandate_id -> unix timestamp of last pool-exhausted notification email
|
|
_poolExhaustedEmailLastSent: Dict[str, float] = {}
|
|
_DEFAULT_COOLDOWN_SEC = 3600
|
|
|
|
|
|
def _normalizeNotifyEmails(raw: Any) -> List[str]:
|
|
if raw is None:
|
|
return []
|
|
if isinstance(raw, list):
|
|
return [str(e).strip() for e in raw if str(e).strip()]
|
|
if isinstance(raw, str):
|
|
s = raw.strip()
|
|
if not s:
|
|
return []
|
|
# JSON array string
|
|
if s.startswith("["):
|
|
try:
|
|
import json
|
|
|
|
parsed = json.loads(s)
|
|
if isinstance(parsed, list):
|
|
return [str(e).strip() for e in parsed if str(e).strip()]
|
|
except Exception:
|
|
pass
|
|
return [s]
|
|
return []
|
|
|
|
|
|
def maybeEmailMandatePoolExhausted(
|
|
mandateId: str,
|
|
triggeringUserId: str,
|
|
triggeringUserLabel: str,
|
|
currentBalance: float,
|
|
requiredAmount: float,
|
|
cooldownSec: float = _DEFAULT_COOLDOWN_SEC,
|
|
) -> None:
|
|
"""
|
|
Send one email per mandate per cooldown to BillingSettings.notifyEmails.
|
|
|
|
Args:
|
|
mandateId: Mandate whose pool is empty.
|
|
triggeringUserId: User who hit the block.
|
|
triggeringUserLabel: Display (e.g. email or username).
|
|
currentBalance: Pool balance (CHF).
|
|
requiredAmount: Minimum required (CHF).
|
|
cooldownSec: Minimum seconds between emails for this mandate.
|
|
"""
|
|
if not mandateId:
|
|
return
|
|
|
|
now = time.time()
|
|
last = _poolExhaustedEmailLastSent.get(mandateId, 0.0)
|
|
if last and (now - last) < cooldownSec:
|
|
logger.debug(
|
|
"Skip mandate pool exhausted email (cooldown): mandate=%s last=%.0fs ago",
|
|
mandateId,
|
|
now - last,
|
|
)
|
|
return
|
|
|
|
try:
|
|
billing = getBillingInterface(getRootUser(), mandateId)
|
|
settings = billing.getSettings(mandateId) or {}
|
|
recipients = _normalizeNotifyEmails(settings.get("notifyEmails"))
|
|
if not recipients:
|
|
logger.warning(
|
|
"PREPAY_MANDATE pool exhausted for mandate %s but notifyEmails is empty — "
|
|
"configure BillingSettings.notifyEmails for owner alerts",
|
|
mandateId,
|
|
)
|
|
return
|
|
|
|
subject = f"[PowerOn] Mandanten-Budget aufgebraucht (Mandant {mandateId[:8]}…)"
|
|
body = (
|
|
f"Das gemeinsame Guthaben (PREPAY_MANDATE) für diesen Mandanten ist nicht mehr ausreichend.\n\n"
|
|
f"Mandanten-ID: {mandateId}\n"
|
|
f"Aktuelles Guthaben (Pool): CHF {currentBalance:.2f}\n"
|
|
f"Benötigt (mind.): CHF {requiredAmount:.2f}\n\n"
|
|
f"Auslösende/r Benutzer/in: {triggeringUserLabel} (ID: {triggeringUserId})\n\n"
|
|
f"Bitte laden Sie das Mandats-Guthaben in der Billing-Verwaltung auf, "
|
|
f"damit Benutzer wieder AI-Funktionen nutzen können.\n"
|
|
)
|
|
escaped = html.escape(body)
|
|
# Cannot use '\\n' inside f-string {…} expression (SyntaxError); build replacement outside.
|
|
brWithNl = "<br>" + "\n"
|
|
htmlMessage = f"""<!DOCTYPE html>
|
|
<html><head><meta charset="utf-8"></head>
|
|
<body style="font-family: Arial, sans-serif; line-height: 1.6;">
|
|
{escaped.replace(chr(10), brWithNl)}
|
|
</body></html>"""
|
|
|
|
messaging = getMessagingInterface()
|
|
any_ok = False
|
|
for to in recipients:
|
|
try:
|
|
ok = messaging.send(
|
|
channel=MessagingChannel.EMAIL,
|
|
recipient=to,
|
|
subject=subject,
|
|
message=htmlMessage,
|
|
)
|
|
if ok:
|
|
any_ok = True
|
|
else:
|
|
logger.warning("Pool exhausted email failed for %s", to)
|
|
except Exception as send_err:
|
|
logger.error("Error sending pool exhausted email to %s: %s", to, send_err)
|
|
|
|
if any_ok:
|
|
_poolExhaustedEmailLastSent[mandateId] = now
|
|
logger.info(
|
|
"Sent mandate pool exhausted notification for mandate %s to %s recipient(s)",
|
|
mandateId,
|
|
len(recipients),
|
|
)
|
|
except Exception as e:
|
|
logger.error("maybeEmailMandatePoolExhausted failed: %s", e, exc_info=True)
|