Alex Orbit
Alex Orbit

Reputation: 11

Make a script to call pager duty's API and update on call

I made this script it works, I'm just looking for ways to make it better/reading to get better.

#!/usr/bin/python3
import requests, logging, json, subprocess, re, smtplib
from datetime import datetime, date

# logger things
log_file = '/var/log/pagerduty.log'
## logging things
logging.basicConfig(level=logging.INFO, format='%(process)d-%(asctime)s-%(levelname)s-%(funcName)s %(message)s',
                    filename=log_file)
###
today = datetime.now()
api_key = '<removed>'
file_path = '<removed>'
# teams_webhook_url = <removed>'
headers = {
    "Accept": "application/vnd.pagerduty+json;version=2",
    "Authorization": f"Token token={api_key}",
    "Content-Type": "application/json"
}

userInfo = {
    "Override": {
        "id": None,
        "name": None,
        "date_start": None,
        "date_end": None,
        "mobile": None,
        "turned_back": None
    },
    "onCall": {
        "id": None,
        "name": None,
        "date": None,
        "mobile": None,
    }
}

schedule_id = '<removed>'
start_time = today.replace(hour=0, minute=0)
end_time = today.replace(hour=23, minute=59)
print(start_time)

with open(file_path, 'r') as file:
    oldFacts = json.load(file)

params = {
    'since': start_time.strftime('%Y-%m-%dT%H:%M:%SZ'),
    'until': end_time.strftime('%Y-%m-%dT%H:%M:%SZ')
}
print(params)


# gets whos on call for real to get there mobile number
def whosOnCall():
    endpointoncall = f'https://api.pagerduty.com/schedules/{schedule_id}'
    responseOnCall = requests.get(endpointoncall, headers=headers, params=params)
    if responseOnCall.status_code == 200:
        oncallrawdata = responseOnCall.json()
        schedule_entries = oncallrawdata['schedule']['schedule_layers'][0]['rendered_schedule_entries']
        for facts in schedule_entries:
            logging.info(facts)
            userInfo['onCall']['id'] = facts['user']['id']
            userInfo['onCall']['name'] = facts['user']['summary']
            break
    else:
        logging.info(f"Failed to fetch overrides. Status code: {responseOnCall.status_code}")
        logging.info(responseOnCall.text)
        message = {"text": "Failed to fetch overrides. Status code: " + str(responseOnCall.status_code) + "🦭🦭 "}
        # teamsMessage(message)


# gett the overrides and userID to get the number later
def userid():
    endpointOverrides = f'https://api.pagerduty.com/schedules/{schedule_id}/overrides'
    response = requests.get(endpointOverrides, headers=headers, params=params)
    if response.status_code == 200:
        overrides_data = response.json()
        overrides = overrides_data['overrides']
        if not overrides:
            logging.info(f"no call  forwarded needed")
            return False
        else:
            # returns a list of dics for overrides
            for override in overrides:
                save_override_info(override)
                return True
    else:
        logging.info(f"Failed to fetch overrides. Status code: {response.status_code}")
        logging.info(response.text)
        message = {"text": "Failed to fetch overrides. Status code: " + str({response.status_code}) + " 🦭🦭 "}
        # teamsMessage(message)


def save_override_info(override):
    date_start = str(override['start'])
    date_end = str(override['end'])
    userInfo['Override']['id'] = override['user']['id']
    userInfo['Override']['date_start'] = date_start[:16]
    userInfo['Override']['date_end'] = date_end[:16]
    userInfo['Override']['name'] = override['user']['summary']
    logging.info(f" this is the info override found {userInfo}")
    return True


# puts the number from pager duty
def phonenumber(userID, whatType):
    endpointMobileNumber = f"https://api.pagerduty.com/users/{userID}/contact_methods/"
    responseMobile = requests.get(endpointMobileNumber, headers=headers)
    if responseMobile.status_code == 200:
        mobileData = responseMobile.json()
        for contactMethod in mobileData['contact_methods']:
            if contactMethod['type'] == 'phone_contact_method':
                userInfo[whatType]['mobile'] = checknumber(contactMethod['address'])
            elif contactMethod['type'] == 'email_contact_method':
                userInfo[whatType]['email'] = contactMethod['address']
    else:
        logging.info(f"Failed to fetch overrides. Status code: {responseMobile.status_code}")
        message = {"text": "Failed to fetch overrides. Status code: " + str(responseMobile.status_code) + " 🦭🦭"}
        # teamsMessage(message)
        logging.info(responseMobile.text)


## email soooooon
def emails(whattype, to, number, overname, output):
    sender = '<removed>'
    try:
        if whattype in ['Override', 'onCall_return']:
            message_subject = 'Whoah man, you\'re  on call'
            message_body = {
                f'🦭🦭! \n\n Call forward for on call number <removed> has been set to your mobile {number} '
                f'{output}'}
        else:
            message_subject = "Whoah man, override found you  "
            message_body = {f'You are not on call  🦭🦭! \n\n Call forward for on call number <removed> has been '
                            f'set to {overname} person {output}'}
        message = f'From: On Call<{sender}>\nTo: To Person {to}\nSubject: {message_subject}\n\n{message_body}'
        message_utf = message.encode('utf-8')
        session = smtplib.SMTP(<removed>')
        session.sendmail(sender, '<removed>'', message_utf)
        logging.info(f'Successfully sent email to {to}')
    except Exception as e:
        logging.info(f"Error: unable to send email {e}")


# def teamsMessage(message):
#    response = requests.post(teams_webhook_url, json=message)
#    if response.status_code == 200:
#        logging.info("Message as been sent")
#    else:
#        logging.info(f"Failed to send message. status code {response.status_code}")


# forwarding the on call 0800 number
def forwardNumbers(number):
    try:
        cmd = ["/usr/local/bin/updateoncallforward", number]
        run = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
        stdout, stderr = run.communicate()
        stdout_without_password = re.sub(r'"password": ".*?"', '"password": "********"', stdout)
        logging.info(f"Subprocess Output: {stdout_without_password}")
        return stdout_without_password
    except FileNotFoundError:
        message = {"text": "updateoncallforward executable not found. help me SOS 🦭🦭"}
        # teamsMessage(message)
        logging.info("updateoncallforward executable not found. 🦭🦭")
    except subprocess.CalledProcessError as e:
        message = {"text": "python call forward script exited with non-zero exit code: " + str(e.returncode) + "🦭🦭"}
        # teamsMessage(message)
        logging.info(f"updateoncallforward script exited with non-zero exit code: {e.returncode}")


# checks the number from pager duty as the numbers starts without the starting 0 adds that to the number
def checknumber(phone_number):
    pattern = r'^02\d{9}|02\d{8}$'
    if not str(phone_number).startswith('0'):
        phone_number = '0' + phone_number
    if re.match(pattern, phone_number):
        return phone_number
    else:
        logging.info(f"somthing is wrong with the phone number the number is question is {phone_number}")
        message = {"text": "somthing is wrong with the phone number the number is question is {phone_number}"}
        # teamsMessage(message)


def main():
    whosOnCall()
    output = ""
    current_server_time = datetime.now()
    start_time =  current_server_time.strftime("%Y-%m-%dT%H:%M")
    print(start_time)
    override_needed = userid()
    if start_time == userInfo['Override']['date_start']:
        if override_needed is True and (oldFacts['Override']['turned_back'] == 0 or oldFacts['Override']['name']
                                        != userInfo['Override']['name']):
            for dic in userInfo:
                phonenumber(userInfo[dic]['id'], dic)
            output = forwardNumbers(userInfo['Override']['mobile'])
            for emailsAddresses in userInfo:
                if emailsAddresses == "Override":
                    emails(emailsAddresses, userInfo[emailsAddresses]['email'], userInfo[emailsAddresses]['mobile'],
                           userInfo[emailsAddresses]['name'], output)
                else:
                    emails(emailsAddresses, '<removed>'', userInfo[emailsAddresses]['mobile'],
                           userInfo[emailsAddresses]['name'], output)
            emails('onCall', userInfo['onCall']['email'], userInfo['Override']['mobile'],
                   userInfo['Override']['name'], output)
            message = {"text": "override found for today forwarded on call number from " + str(userInfo['onCall']['name'])
                               + "to " + str(userInfo['Override']['name']) + str(output)}
            #teamsMessage(message)
            userInfo['Override']['turned_back'] = 1
            with open(file_path, 'w') as file_override:
                json.dump(userInfo, file_override, indent=4)
    elif (
            override_needed is False
            and oldFacts['Override']['turned_back'] == 1
            and start_time == oldFacts['Override']['date_end']
    ):
        output = forwardNumbers(oldFacts['onCall']['mobile'])
        for emailsAddresses in oldFacts:
            if emailsAddresses == "Override":
                emails(emailsAddresses, oldFacts[emailsAddresses]['email'], oldFacts[emailsAddresses]['mobile'],
                       oldFacts[emailsAddresses]['name'], output)
            else:
                emails(emailsAddresses, '<removed>', oldFacts[emailsAddresses]['mobile'],
                       oldFacts[emailsAddresses]['name'], output)

        #emails('Override_return', oldFacts['Override']['email'], oldFacts['Override']['mobile'],
        #       oldFacts['Override']['name'], output)
        #emails('onCall_return', oldFacts['onCall']['email'], oldFacts['Override']['mobile'],
        #       oldFacts['Override']['name'], output)
        message = {"text": "Override returned to " + str(oldFacts['onCall']['name']) + "that was taken over from " +
                           str(oldFacts['Override']['name']) + "  could we just double check this worked 🦭🦭"}
        # teamsMessage(message)
        oldFacts['Override']['turned_back'] = 0
        with open(file_path, 'w') as file:
            json.dump(oldFacts, file, indent=4)
        logging.info(f"on call returned to  {oldFacts['onCall']['name']}")
    else:
        logging.info("nothing to be done")
    logging.info("-" * 60)


if __name__ == "__main__":
    main()

# alert help im lost

Also when it finds an override it runs a different script that updates the on call number on net, and also updates the offnet number, forwards the output of that script so you know it works.

Upvotes: 0

Views: 337

Answers (0)

Related Questions