Reputation: 11
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