munin-notify.py

#!/usr/bin/env python3
import webbrowser

from fnmatch import fnmatch
from optparse import OptionParser

import requests

from lxml import html

import gi
gi.require_version('Gtk', '3.0')
gi.require_version('Notify', '0.7')
from gi.repository import GLib, Notify  # noqa


class Config:
    def __init__(self):
        parser = OptionParser(usage='usage: %prog url [options]')
        parser.add_option('-i', '--ignore', dest='ignore', action='callback',
                          callback=self._add_ignore, type='string',
                          help='Comma-separated list of names, composed of the server followed by its probes to ignore. '
                               'Unix filename pattern matching is supported. '
                               '.e.g."bak*, nginx, back?p". '
                               'The parameter can be repeated.'
                          )
        parser.add_option('-g', '--group', dest='group', action='append',
                          help='Group to include. The parameter can be repeated')

        self.options, self.args = parser.parse_args()

    def _add_ignore(self, option, opt, value, parser):
        ignores = getattr(parser.values, option.dest, None)
        if ignores is None:
            ignores = {}
            setattr(parser.values, option.dest, ignores)
        values = value.split(',')
        server, probes = values[0], values[1:]
        ignores[server] = probes


class App:
    def __init__(self, url, groups=None, ignores=None):
        Notify.init("Munin Odoo")
        self.url = url
        self.groups = groups
        self.ignores = ignores
        self.notification = Notify.Notification.new("")
        self.notification.set_urgency(2)
        self.notification.add_action(
            "clicked",
            "Open",
            self.open,
            None,
        )
        self.check()

    def check(self):
        try:
            content = requests.get(self.url).content
        except Exception:
            return
        tree = html.fromstring(content)
        alerts = {}
        base_xpath = ''
        if self.groups:
            base_xpath += '//ul[contains(@class, "groupview")]/li/span/a[%s]/../..' % ' or '.join('text()="%s"' % group for group in self.groups)
        alert_nodes = tree.xpath(base_xpath + '//a[contains(@class, "crit")]')
        for node in alert_nodes:
            server = node.xpath('preceding-sibling::span')[0].text_content()
            probe = node.text_content()
            if self.ignore_match(server, probe):
                continue
            alerts.setdefault(server, []).append(probe)
            if len(alerts) > 5:
                break
        if alerts:
            self.notification.update('Munin Alerts!', '\n'.join('%s: %s' % (server, ' '.join(alerts)) for server, alerts in alerts.items()))
            self.notification.show()
        GLib.timeout_add_seconds(600, self.check)

    def ignore_match(self, server, probe):
        ignores = self.ignores or {}
        for server_rule, probe_rules in ignores.items():
            if fnmatch(server, server_rule):
                for probe_rule in probe_rules:
                    if fnmatch(probe, probe_rule):
                        return True
        return False

    def open(self, notification, action_name, data):
        webbrowser.open(self.url, new=2, autoraise=True)


config = Config()
app = App(config.args[0], groups=config.options.group, ignores=config.options.ignore)
GLib.MainLoop().run()