Drogheda/04-Source-Code/notifications/notify_mattermost.py

91 lines
3.2 KiB
Python

#!/usr/bin/env python3
"""
notify_mattermost.py — post a message to a Mattermost channel via Incoming Webhook.
Dependency-free (stdlib urllib). The webhook URL is a secret and is read from the
environment — NEVER hardcode it (Helgen ethos: no secrets in the repo).
Setup (one-time):
1. In Mattermost: Main Menu → Integrations → Incoming Webhooks → Add. Pick the
channel, copy the URL (looks like https://<host>/hooks/xxxxxxxxxxxxxxxxx).
2. export MATTERMOST_WEBHOOK_URL="https://<host>/hooks/xxxxxxxxxxxxxxxxx"
Usage:
python3 notify_mattermost.py "Backend hotfix deployed ✅"
echo "multi-line\nstatus" | python3 notify_mattermost.py -
python3 notify_mattermost.py --channel ops --username "Drogheda Bot" "RTK fix restored"
python3 notify_mattermost.py --dry-run "test" # print payload, don't send
Exit codes: 0 ok · 2 usage/config error · 1 send failure.
"""
import argparse
import json
import os
import sys
import urllib.request
import urllib.error
DEFAULT_USERNAME = "Drogheda Bot"
DEFAULT_ICON = ":satellite:"
def build_payload(text, channel=None, username=DEFAULT_USERNAME, icon=DEFAULT_ICON):
payload = {"text": text, "username": username}
if icon:
payload["icon_emoji"] = icon
if channel:
payload["channel"] = channel # override the webhook's default channel
return payload
def send(url, payload, timeout=10):
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
url, data=data, headers={"Content-Type": "application/json"}, method="POST"
)
with urllib.request.urlopen(req, timeout=timeout) as resp:
return resp.status, resp.read().decode("utf-8", "ignore")
def main(argv=None):
p = argparse.ArgumentParser(description="Post a message to Mattermost via webhook.")
p.add_argument("text", help="Message text (Markdown ok). Use '-' to read from stdin.")
p.add_argument("--channel", help="Override channel (name without #).")
p.add_argument("--username", default=DEFAULT_USERNAME)
p.add_argument("--icon", default=DEFAULT_ICON, help="Emoji, e.g. :satellite:")
p.add_argument("--url", default=os.environ.get("MATTERMOST_WEBHOOK_URL"),
help="Webhook URL (defaults to $MATTERMOST_WEBHOOK_URL).")
p.add_argument("--dry-run", action="store_true", help="Print payload; do not send.")
args = p.parse_args(argv)
text = sys.stdin.read().rstrip("\n") if args.text == "-" else args.text
if not text.strip():
print("error: empty message", file=sys.stderr)
return 2
payload = build_payload(text, args.channel, args.username, args.icon)
if args.dry_run:
print(json.dumps(payload, indent=2, ensure_ascii=False))
return 0
if not args.url:
print("error: no webhook URL. Set MATTERMOST_WEBHOOK_URL or pass --url.",
file=sys.stderr)
return 2
try:
status, body = send(args.url, payload)
if status == 200:
print("ok")
return 0
print(f"send failed: HTTP {status}: {body}", file=sys.stderr)
return 1
except urllib.error.URLError as e:
print(f"send failed: {e}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())