awx_webook.py

import json

from errbot import BotPlugin, botcmd, webhook

AWX_OUTPUT = '#awx_output'


class AWX(BotPlugin):
    """
    This plugin implements AWX webhook

    """

    @webhook('/awx-webhook/', raw=True, methods=['POST'])
    def awx_post(self, request):
        channel = self.build_identifier(AWX_OUTPUT)
        if request.headers['content-type'].lower() in ['application/json',
                                                       'text/json']:
            json_data = request.get_json()
            json_text = json.dumps(json_data, indent=4)
            if 'friendly_name' not in json_data:
                self.send(channel,
                    ':ghost: JSON Body is \n\n' + f"```\n{json_text}\n```")
                return 'OK'
        else:
            self.send(channel, ':ghost: Body is \n\n' + repr(request.data))
            return "OK"

        if 'friendly_name' not in json_data:
            return
        assert json_data['friendly_name'] == 'Job', json_data
        job_url = json_data['url']
        job_name = json_data['name']
        failed_hosts = []
        ok_hosts = []
        changed_hosts = []
        unchanged_hosts = []
        for host_name, host_data in json_data['hosts'].items():
            if host_data.get('failed', True):
                failed_hosts.append(host_name)
            else:
                ok_hosts.append(host_name)
            if host_data.get('changed', True):
                changed_hosts.append(host_name)
            else:
                unchanged_hosts.append(host_name)

        if failed_hosts:
            icon = ':red_circle:'
            msg = 'Failed hosts: ' + ', '.join(failed_hosts)
            job_state = 'failed'
        else:
            icon = ':heavy_check_mark:'
            msg = f'{len(ok_hosts)} hosts succeeded'
            job_state = 'succeeded'
        self.send(channel,
            f'{icon} Job **{job_name}** {job_url} {job_state}\n{msg}')

        if changed_hosts:
            icon = ':warning:'
            changed_str = ', '.join(changed_hosts)
            self.send(channel,
                      f'{icon} Some hosts {len(changed_hosts)} has been changed: {changed_str}')

        return "OK"