""" Ansible callback plugin — send deploy notifications to Gotify (always) and Pushover (on failure). Enable by placing in callback_plugins/ directory of the playbook. Configure via environment variables: NOTIFY_GOTIFY_URL - Gotify server URL (e.g., http://192.168.1.29:8765) NOTIFY_GOTIFY_TOKEN - Gotify application token NOTIFY_PUSHOVER_TOKEN - Pushover API token (for failure alerts) NOTIFY_PUSHOVER_USER - Pushover user key (for failure alerts) NOTIFY_ENABLED - Set to "true" to enable (default: disabled for local runs) """ import json import os import urllib.request import urllib.parse from collections import defaultdict from datetime import datetime from ansible.plugins.callback import CallbackBase DOCUMENTATION = """ name: notify_result type: notification short_description: Send deploy results to Gotify/Pushover description: - Sends a Gotify message on every playbook completion - Sends a Pushover message only on failure """ class CallbackModule(CallbackBase): CALLBACK_VERSION = 2.0 CALLBACK_TYPE = "notification" CALLBACK_NAME = "notify_result" CALLBACK_NEEDS_WHITELIST = False def __init__(self): super().__init__() self.disabled = os.environ.get("NOTIFY_ENABLED", "false").lower() != "true" self.gotify_url = os.environ.get("NOTIFY_GOTIFY_URL", "") self.gotify_token = os.environ.get("NOTIFY_GOTIFY_TOKEN", "") self.pushover_token = os.environ.get("NOTIFY_PUSHOVER_TOKEN", "") self.pushover_user = os.environ.get("NOTIFY_PUSHOVER_USER", "") self.play_name = "" # Track changed/failed tasks per host self.changed_tasks = defaultdict(list) # host -> [task_name, ...] self.failed_tasks = defaultdict(list) # host -> [(task_name, msg), ...] self.unreachable_hosts = [] self.rebooted_hosts = [] def v2_playbook_on_play_start(self, play): self.play_name = play.get_name() def v2_runner_on_failed(self, result, ignore_errors=False): if not ignore_errors: host = result._host.get_name() task = result._task.get_name() msg = result._result.get("msg", "") self.failed_tasks[host].append((task, msg)) def v2_runner_on_unreachable(self, result): host = result._host.get_name() if host not in self.unreachable_hosts: self.unreachable_hosts.append(host) def v2_runner_on_ok(self, result): if result.is_changed(): host = result._host.get_name() task = result._task.get_name() self.changed_tasks[host].append(task) if "reboot" in task.lower(): self.rebooted_hosts.append(host) def v2_playbook_on_stats(self, stats): if self.disabled: return hosts = sorted(stats.processed.keys()) has_failures = any( stats.summarize(h)["failures"] > 0 or stats.summarize(h)["unreachable"] > 0 for h in hosts ) timestamp = datetime.now().strftime("%Y-%m-%d %H:%M") if has_failures: title = f"FAILED — {self.play_name}" message = self._build_failure_message(hosts, stats, timestamp) priority = 8 elif self.changed_tasks: title = f"OK — {self.play_name}" message = self._build_changes_message(hosts, stats, timestamp) priority = 4 else: title = f"OK — {self.play_name}" message = f"{timestamp}\n\nNo changes across {len(hosts)} host(s)" priority = 2 self._send_gotify(title, message, priority) if has_failures: self._send_pushover(title, message) def _build_failure_message(self, hosts, stats, timestamp): lines = [timestamp, ""] # Failed tasks if self.failed_tasks: lines.append("FAILURES:") for host in sorted(self.failed_tasks.keys()): lines.append(f"\n {host}:") for task, msg in self.failed_tasks[host]: short_msg = msg[:120] if msg else "" lines.append(f" ✗ {task}") if short_msg: lines.append(f" {short_msg}") # Unreachable if self.unreachable_hosts: lines.append(f"\nUNREACHABLE: {', '.join(self.unreachable_hosts)}") # Changes that did succeed if self.changed_tasks: lines.append("\nChanges before failure:") for host in sorted(self.changed_tasks.keys()): lines.append(f" {host}: {len(self.changed_tasks[host])} changed") return "\n".join(lines) def _build_changes_message(self, hosts, stats, timestamp): lines = [timestamp, ""] for host in sorted(self.changed_tasks.keys()): tasks = self.changed_tasks[host] s = stats.summarize(host) lines.append(f"{host} ({s['changed']} changed, {s['ok']} ok):") for task in tasks: lines.append(f" • {task}") lines.append("") # Hosts with no changes unchanged = [h for h in hosts if h not in self.changed_tasks] if unchanged: lines.append(f"No changes: {', '.join(unchanged)}") if self.rebooted_hosts: lines.append(f"\n⟳ Rebooted: {', '.join(self.rebooted_hosts)}") return "\n".join(lines) def _send_gotify(self, title, message, priority): if not self.gotify_url or not self.gotify_token: return try: url = f"{self.gotify_url}/message?token={self.gotify_token}" data = json.dumps({ "title": title, "message": message, "priority": priority, "extras": { "client::display": {"contentType": "text/plain"} }, }).encode() req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"}) urllib.request.urlopen(req, timeout=10) except Exception as e: self._display.warning(f"Gotify notification failed: {e}") def _send_pushover(self, title, message): if not self.pushover_token or not self.pushover_user: return try: url = "https://api.pushover.net/1/messages.json" data = urllib.parse.urlencode({ "token": self.pushover_token, "user": self.pushover_user, "title": title, "message": message[:1024], "priority": 1, }).encode() req = urllib.request.Request(url, data=data) urllib.request.urlopen(req, timeout=10) except Exception as e: self._display.warning(f"Pushover notification failed: {e}")