alerts: Teams webhook fan-out (contract 0.14.0) + printedparts detail revision column

send_webhook(title,text) posts alerts to an optional webhook (Teams Incoming
Webhook / Workflow, or generic JSON) via alert_webhook_url + alert_webhook_format
settings; send_alert fans out to it alongside email; exposed on shopdb.api
(0.13.0->0.14.0, PLUGIN-HOOKS synced); low-stock posts on its custom-recipient
path too. Also: recent-transactions table shows the consumed print-file revision.
This commit is contained in:
cproudlock
2026-07-22 10:21:42 -04:00
parent d141fef203
commit 38c7ec347b
9 changed files with 162 additions and 11 deletions

View File

@@ -36,7 +36,7 @@ from .plugins import plugin_manager
# unattended endpoints (e.g. the GE-Enforce fetch API) can authorize a scoped
# managed service token without importing core token internals. Additive name
# on the import surface, minor bump.
__contract_version__ = '0.13.0'
__contract_version__ = '0.14.0'
# Product release version (see ADR-007). The product version and the
# plugin-contract version above are distinct series with independent

View File

@@ -80,7 +80,7 @@ from shopdb.core.services.dualpath import (
# Legacy employee directory lookup (read-only) used by notifications
from shopdb.utils.employee_db import employee_connection
from shopdb.utils.mailer import send_email, send_alert
from shopdb.utils.mailer import send_email, send_alert, send_webhook
# CMMC USB check-in/out database (read-write) used by the usb plugin
from shopdb.utils.cmmc_usb_db import cmmc_usb_connection
@@ -271,6 +271,7 @@ __all__ = [
'employee_connection',
'send_email',
'send_alert',
'send_webhook',
'User',
'Role',
# CMMC USB check-in/out database

View File

@@ -838,6 +838,23 @@ def build_default_settings():
'description': 'Public base URL of this site (e.g. '
'https://host/ops), used to build links in emails'
},
{
'key': 'alert_webhook_url',
'value': '',
'valuetype': 'string',
'category': 'email',
'description': 'Optional webhook URL alerts also POST to (e.g. a '
'Microsoft Teams Incoming Webhook or Workflow)'
},
{
'key': 'alert_webhook_format',
'value': 'teams',
'valuetype': 'string',
'category': 'email',
'description': 'Alert webhook payload format: teams (classic '
'Incoming Webhook), adaptivecard (Teams Workflow), '
'or json (generic {title,text})'
},
# Audit log settings
{
'key': 'audit_retention_days',

View File

@@ -19,6 +19,8 @@ import os
import re
import smtplib
import ssl
import requests
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.utils import formataddr
@@ -175,13 +177,89 @@ def send_email(to, subject, html, text=None):
return ok
def get_webhook_config():
"""Alert-webhook config from settings: the URL and the payload format."""
settings = {}
if has_app_context():
from shopdb.core.api.settings import get_cached_settings
try:
settings = get_cached_settings() or {}
except Exception:
settings = {}
return {
'url': (settings.get('alert_webhook_url') or '').strip(),
# 'teams' = classic Incoming Webhook (MessageCard); 'adaptivecard' =
# Teams Workflow (Power Automate); 'json' = generic {title,text}.
'format': (settings.get('alert_webhook_format') or 'teams').strip().lower(),
}
def _webhook_payload(fmt, title, text):
if fmt == 'adaptivecard':
return {
'type': 'message',
'attachments': [{
'contentType': 'application/vnd.microsoft.card.adaptive',
'content': {
'type': 'AdaptiveCard',
'$schema': 'http://adaptivecards.io/schemas/adaptive-card.json',
'version': '1.4',
'body': [
{'type': 'TextBlock', 'weight': 'Bolder',
'size': 'Medium', 'text': title, 'wrap': True},
{'type': 'TextBlock', 'text': text, 'wrap': True},
],
},
}],
}
if fmt == 'json':
return {'title': title, 'text': text}
# default 'teams' = classic Incoming Webhook connector MessageCard
return {
'@type': 'MessageCard',
'@context': 'https://schema.org/extensions',
'summary': title,
'themeColor': 'D93F3F',
'title': title,
'text': text,
}
def send_webhook(title, text):
"""POST an alert to the configured webhook (Teams, etc.). Best-effort:
returns (ok, error); a no-op ((False, None)) when no URL is configured, and
never raises so it can never block the caller."""
config = get_webhook_config()
if not config['url']:
return False, None
try:
response = requests.post(
config['url'], json=_webhook_payload(config['format'], title, text),
timeout=10)
if response.status_code >= 400:
return False, f'HTTP {response.status_code}'
_log().info('Alert webhook posted: %s', title)
return True, None
except Exception as exception:
error = str(exception)
_log().warning('Alert webhook failed: %s', error)
return False, error
def send_alert(subject, html, text=None):
"""Send an alert to the site's configured alert_recipients. Returns False
when email is off or no alert recipients are configured."""
"""Fan an alert out to the configured channels: the site's alert_recipients
(email) and the alert webhook (Teams, etc.). Each channel is independent and
best-effort; returns True if the EMAIL leg sent."""
body = text or _html_to_text(html)
# Webhook fans out alongside email, independent of SMTP being configured.
try:
send_webhook(subject, body)
except Exception:
pass
config = get_smtp_config()
recipients = _normalize_recipients(config['alert_recipients'])
if not recipients:
_log().warning('Alert not sent: no alert_recipients configured.')
_log().warning('Alert email not sent: no alert_recipients configured.')
return False
return send_email(recipients, subject, html, text=text)