gistfile1.txt

#!/usr/bin/env python3
import requests
import json
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
import dateutil.parser
from datetime import datetime, timezone, timedelta
import boto3
import os
import re
from botocore.exceptions import ClientError
from base64 import b64decode
from enum import IntEnum
import humanize
import argparse


def get_secret(secret_name, region_name):
    # Create a Secrets Manager client
    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager',
        region_name=region_name
    )

    try:
        get_secret_value_response = client.get_secret_value(
            SecretId=secret_name
        )
    except ClientError as e:
            raise e
    else:
        return get_secret_value_response.get('SecretString')


PARSER = argparse.ArgumentParser(description='Aerial Attack Bot')
PARSER.add_argument('-l', '--local', dest='local_env', action='store_true',
                    help='use if running locally and you want to use your local thor and sys admin environment vars')
PARSER.add_argument('-c', '--channel', dest='slack_channel', type=str, default='#autocap-bot-prod',
                    help='pass in a custom slack channel for alerting')
PARSER.add_argument('-b', '--base-url', dest='base_url', type=str, default='https://admin.hudl.com',
                    help='pass in the url for a specific branch')

ARGS = PARSER.parse_args()

CHANNEL = ARGS.slack_channel
BASE_URL = ARGS.base_url
PROD_ADMIN_PASSWORD = os.environ['PROD_ADMIN_PASSWORD']
REGION_NAME = 'us-east-1'

if ARGS.local_env:
    SYS_ADMIN_PASSWORD = os.environ['SYS_ADMIN_PASSWORD']
    if BASE_URL is not 'https://admin.hudl.com':
        SYS_ADMIN_PASSWORD = os.environ['THOR_PASSWORD']
    SLACK_API_KEY = os.environ['SLACK_API_KEY']
    event = None
    context = None
else:
    SYS_ADMIN_PASSWORD = get_secret(PROD_ADMIN_PASSWORD, REGION_NAME)

    SLACK_API_KEY = os.environ['SLACK_API_KEY']
    # SLACK_API_KEY = boto3.client('kms').decrypt(CiphertextBlob=b64decode(ENC_SLACK_API_KEY))['Plaintext'].decode(
    #     'utf-8')

SLACK = WebClient(token=SLACK_API_KEY)
LOGIN = f"{BASE_URL}/api/v2/login"
DATA = {
    "username": "[email protected]",
    "password": SYS_ADMIN_PASSWORD
}

S = requests.Session()
S.post(LOGIN, DATA)

COLORS = {
    'success': '#78a100',
    'info': '#009ce3',
    'error': '#e81c00',
}

class SetupStatus(IntEnum):
    Unknown = 0
    Ordered = 1
    Delivered = 2
    Initialized = 4
    CamerasConfigured = 8
    Disabled = 16
    Installed = 32

# https://github.com/hudl/hudl-automaticcapture/blob/750176c92f9d1da61389b0ca03ace1177155133f/src/Hudl.AutomaticCapture.Client/Enum/SupportEventIssueType.cs#L7
class SupportEventIssueType(IntEnum):
    Unknown = 0
    DeviceError = 1
    Offline = 2
    ReadyForCalibration = 3
    Note = 4
    HardwareIssue = 5
    PublishEventError = 6
    RestreamDestinationError = 7
    ReplacementError = 8
    CalibrationAutoUpdateError = 9


def string_to_datetime(string):
    d = dateutil.parser.parse(string)
    return d


def get_installation(installation_id):
    try:
        url = f"{BASE_URL}/api/v2/automatic-capture/installations/{installation_id}"
        r = S.get(url)
        installation_info = r.json()
        return installation_info
    except Exception as e:
        print(f'    [get_installation] {e}')


def post_offline_message(install, channel):
    """
    posts a pre-formatted message in slack
    """
    installation_id = install['installationId']
    device_id = install['deviceId'].upper()

    install_url = f"{BASE_URL}/admin/automaticcapture/installations/{installation_id}"

    name_and_id = device_id

    return {    "fallback": "Focus Device Status Update!",
                "color": COLORS['error'],
                "text": "<{}|*{}*>  Offline since {}".format(install_url, name_and_id, humanize.naturaldelta(time_of_run_rounded_down - string_to_datetime(install['latestStatus']['createdAt'])))
           }

def post_error_message(install, codes, channel):
    """
    posts a pre-formatted message in slack
    """
    installation_id = install['installationId']
    install_url = f"{BASE_URL}/admin/automaticcapture/installations/{installation_id}"
    name_and_id = install['name']

    print(codes)

    error_string = ""
    for _, code in codes.items():
        error_string += "{0}: {1} {2}\n".format(code['statusCode'], code['expectedValue'], code['description'])

    return {    "fallback": "Focus Device Status Update!",
                "color": COLORS['error'],
                "text": "<{}|*{}*> Error {}".format(install_url, name_and_id, error_string)
            }


def post_message(attachments, channel):
    a = [{    "pretext": "NTF Installation update:",
              "fallback": "Focus Device Status Update!"}] + attachments

    print(a)
    try:
        resp = SLACK.chat_postMessage(
            channel=channel,
            attachments=a
        )
    except SlackApiError as e:
        print(e) 

def print_better(list_of_issues):
    new_s = ""
    for item in list_of_issues:
        new_s += f"{item}\n"
    return new_s


def check_if_online(installation_status: str) -> bool:
    return (time_of_run_rounded_down - timedelta(minutes=5)) <= string_to_datetime(installation_status['latestStatus']['createdAt'])


def main(event, context):
    global time_of_run  # Variables declared at the top of the module aren't refreshed every run on AWS
    time_of_run = datetime.now(timezone.utc) - timedelta(minutes=10)

    global time_of_run_rounded_down
    time_of_run_rounded_down = time_of_run - timedelta(minutes=time_of_run.minute % 10, seconds=time_of_run.second, microseconds=time_of_run.microsecond)

    installation_list = ["605a2dc7a3cabc0001225e0a",
                         "605a151ba5a92d00010b83ca",
                         "605a5283ea79950001034ce1",
                         "605a5d0206b19b0001a06ad5",
                         "605b3baad9ec030001be97ba",
                         "605a1c7ca3cabc0001219aae",
                         "605b58728f3ad9000145f1eb",
                         "605b3b73c8e5f80001a61716",
                         "605a4dbaa1f1310001fa5cb6",
                         "605a526be9417b0001d68703",
                         "605a0e71c139ad000139172c",
                         "605a526be9417b0001d68703",
                         "605b3b9d7e24da00012e2578",
                         "605a3bd2e9417b0001d56f77",
                         "605a5292512f7a00015f02b3",
                         "605a4d982bc7c70001af69f7",
                         "605a33436e82a80001030d71"]

    offline_list = []
    error_list = []

    for i in installation_list:
        installation = get_installation(i)    
        if not check_if_online(installation):
            offline_list.append(installation)
            print(installation['name'], check_if_online(installation))
        try:
            if len(installation['latestStatus']['statusCodes']) > 0:
                error_list.append((installation, installation['latestStatus']['statusCodes']))
        except Exception as e:
            pass

    messages = []
    for install in offline_list:
        messages.append(post_offline_message(install, CHANNEL))
    for install, codes in error_list:
        messages.append(post_error_message(install, codes, CHANNEL))
        
    post_message(messages, CHANNEL)


if __name__ == "__main__":
    main(event, context)