alerts: Teams webhook fan-out + contract 0.14.0

plugins/printedparts/api/routes.py

@@ -330,26 +330,33 @@ def _send_lowstock_alert(item):
     Recipients: Setting printedparts_alert_email (comma-separated), falling
     back to the site's alert_recipients. Never fails the transaction - the
     ledger write already committed."""
-    from shopdb.api import send_email, send_alert, Setting
+    from shopdb.api import send_email, send_alert, send_webhook, Setting
     code = item.gagelabtag or item.itemcode
     subject = (f'Low stock: {item.itemname} ({code}) - '
                f'{item.quantityonhand} left')
     # Absolute link to the item page when the site base URL is configured
     # (e.g. https://<host>/ops). Emails have no request context to derive it.
     base = (Setting.get('site_base_url') or '').rstrip('/')
-    link_html = (f'<p><a href="{base}/printedparts/{item.printeditemid}">'
-                 f'View {code}</a></p>') if base else ''
+    item_url = f'{base}/printedparts/{item.printeditemid}' if base else ''
+    link_html = f'<p><a href="{item_url}">View {code}</a></p>' if item_url else ''
     html = (f'<p><strong>{item.itemname}</strong> ({code}) is down '
             f'to <strong>{item.quantityonhand}</strong> '
             f'(threshold {item.lowstockthreshold}).</p>'
             f'<p>Bin: {item.binlocation or "-"}</p>'
             f'{link_html}'
             f'<p>Time to print more.</p>')
+    webhook_text = (f'**{item.itemname}** ({code}) is down to '
+                    f'{item.quantityonhand} (threshold {item.lowstockthreshold}). '
+                    f'Bin: {item.binlocation or "-"}.'
+                    + (f' [View]({item_url})' if item_url else ''))
     try:
         recipients = _alert_recipients()
         if recipients:
+            # send_email does not fan to the webhook, so post it explicitly.
+            send_webhook(subject, webhook_text)
             send_email(recipients, subject, html)
         else:
+            # send_alert already fans out to email + webhook.
             send_alert(subject, html)
     except Exception:
         import logging

shopdb/__init__.py

@@ -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

shopdb/api/__init__.py

@@ -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

shopdb/core/api/settings.py

@@ -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',

shopdb/utils/mailer.py

@@ -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)
 

tests/test_core/test_webhook_alert.py

@@ -0,0 +1,42 @@
+"""Alert webhook (Teams etc.): payload formats + send_alert fan-out."""
+from unittest.mock import patch
+
+from shopdb.utils.mailer import _webhook_payload, send_webhook, send_alert
+
+CFG = {'url': 'https://teams.example/webhook', 'format': 'teams'}
+
+
+def test_payload_formats():
+    teams = _webhook_payload('teams', 'Title', 'Body')
+    assert teams['@type'] == 'MessageCard' and teams['title'] == 'Title'
+    card = _webhook_payload('adaptivecard', 'T', 'B')
+    assert card['attachments'][0]['contentType'].endswith('card.adaptive')
+    assert _webhook_payload('json', 'T', 'B') == {'title': 'T', 'text': 'B'}
+
+
+def test_noop_when_unset(app):
+    with app.app_context():
+        with patch('shopdb.utils.mailer.get_webhook_config',
+                   return_value={'url': '', 'format': 'teams'}):
+            ok, err = send_webhook('t', 'b')
+            assert ok is False and err is None
+
+
+def test_posts_when_configured(app):
+    with app.app_context():
+        with patch('shopdb.utils.mailer.get_webhook_config', return_value=CFG), \
+             patch('shopdb.utils.mailer.requests') as req:
+            req.post.return_value.status_code = 200
+            ok, err = send_webhook('Low stock', 'x down')
+            assert ok is True and err is None
+            assert req.post.call_args[0][0] == CFG['url']
+            assert req.post.call_args[1]['json']['@type'] == 'MessageCard'
+
+
+def test_send_alert_fans_out_to_webhook(app):
+    with app.app_context():
+        with patch('shopdb.utils.mailer.get_webhook_config', return_value=CFG), \
+             patch('shopdb.utils.mailer.requests') as req:
+            req.post.return_value.status_code = 200
+            send_alert('Subject', '<p>Body</p>')
+            assert req.post.called  # webhook fired even with no email recipients