Python Scripts: Difference between revisions
(49 intermediate revisions by the same user not shown) | |||
Line 2:
__TOC__
<br />
= Code Snippets =
== Print without next line ==
import sys
sys.stdout.write('.')
You may also need to call to ensure stdout is flushed immediately:
sys.stdout.flush()
== Regex ==
*Extract a particular value(like 10.5, 100.0):
value = re.findall(r'([0-9]{3}.[0-9]|[0-9]{2}.[0-9]|[0-9].[0-9])(?=\sid)', output)
# Print all matches
print value
# Print 2nd Match
print value[1]
== Arrays ==
*Using Array
x = [1,2,3,4,5]
for i in range(5):
print x[i]
*Append to Array
f = []
for i in range(30):
f.append(i)
print f
== Pause ==
import time
time.sleep(5)
== Print File Extensions ==
import os.path
extension = os.path.splitext(filename)[1][1:]
== EasySNMP ==
{{notice|This snippet needs more testing.}}
sudo apt-get install libsnmp-dev snmp-mibs-downloader
sudo apt-get install gcc python-dev
sudo pip install easysnmp
from easysnmp import snmp_get
snmp_get('.1.3.6.1.4.1.2021.10.1.3.1', hostname='10.107.88.93', community='public', version=2)
= Calculator =
Line 68 ⟶ 117:
</pre>
= Schedule output =
sudo pip install schedule
<pre>
import schedule
import time
def job():
print("I'm working...")
schedule.every(1).second.do(job)
schedule.every(1).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("21:14").do(job)
while 1:
schedule.run_pending()
time.sleep(1)
</pre>
= Fill Routing Table of Remote Server =
<pre>
Line 214 ⟶ 282:
</pre>
;Terminal Plot
sudo pip install terminalplot
<pre>
from terminalplot import plot
#import gnuplotlib as gp
import subprocess as sub
import re
x = 0
a = []
b = []
try:
p = sub.Popen(('sudo', 'tcpdump', '-l'), stdout=sub.PIPE)
for row in iter(p.stdout.readline, b''):
y = re.search('length (\d+)', row)
if y:
z = int(y.group(1))
if z != 0:
x += 1
print x
b.append(z)
a.append(x)
if x == 10:
plot(a,b)
print a
print b
# Remove below # if you don't want continuous output
# break
x = 0
a = []
b = []
except KeyboardInterrupt:
print("User killed command.")
</pre>
;Plot using Stdout
<pre>
import subprocess as sub
import re
import sys
x = 0
try:
p = sub.Popen(('sudo', 'tcpdump', '-l'), stdout=sub.PIPE)
for row in iter(p.stdout.readline, b''):
y = re.search('length (\d+)', row)
if y:
z = int(y.group(1))
x += 1
if (z < 190):
for i in range(z):
sys.stdout.write('=')
print (">")
except KeyboardInterrupt:
print("User killed command.")
</pre>
= Speaking Passwords =
;eSpeak Password
<pre>
import os
Line 224 ⟶ 354:
else:
os.system("espeak 'wrong password'")
def again():
check()
again()
check()
again()
</pre>
;Using pyttsx
<pre>
import pyttsx
engine = pyttsx.init()
def check():
x = int(input("Input your password: "))
if (x == 1234):
engine.say('correct password')
engine.runAndWait()
else:
engine.say('wrong password')
engine.runAndWait()
def again():
Line 385 ⟶ 537:
</pre>
= Reboot Netscaler via
{{notice|This script is under construction.}}
*Needs Authentication alerts
Line 394 ⟶ 546:
import subprocess as sub
import time
# Nitro API calls using Curl
Line 428 ⟶ 572:
</pre>
;
<pre>
import subprocess as sub
Line 437 ⟶ 581:
username = "nsroot"
passwd = "pwd@123"
# Boolean true means Warm reboot, else false
boolean = "true"
# Nitro API calls using Curl
uptimeapi = "curl -s -k -X GET -H 'Content-Type:application/json' --basic --user %s:%s http://%s/nitro/v1/stat/system?attrs=starttime" %(username, passwd, host)
rebootapi = "curl -s -k -X POST -H 'Content-Type:application/vnd.com.citrix.netscaler.reboot+json' --basic --user %s:%s -d '{'reboot':{'warm':
print uptimeapi
Line 453 ⟶ 596:
sub.Popen(rebootapi.split(), stdout=sub.PIPE)
time.sleep(
uptime2 = sub.Popen(uptimeapi.split(), stdout=sub.PIPE)
Line 464 ⟶ 607:
print "Reboot successful"
</pre>
= Reboot Netscaler via API Call using SDK =
Download & Install Nitro REST SDK: [https://www.citrix.co.in/community/citrix-developer/netscaler/nitro-sdk.html citrix.co.in]
==Save config & Reboot ==
<pre>
import sys
from nssrc.com.citrix.netscaler.nitro.exception.nitro_exception import nitro_exception
from nssrc.com.citrix.netscaler.nitro.resource.base.base_resource import base_resource
from nssrc.com.citrix.netscaler.nitro.service.nitro_service import nitro_service
ip = "10.107.88.78"
username = "nsroot"
password = "pwd@123"
try:
client = nitro_service(ip,"http")
client.set_credential(username,password)
client.timeout = 20
client.save_config()
client.reboot(True)
client.logout()
print "Config saved, Rebooting now.."
except nitro_exception as e:
print("Exception::errorcode="+str(e.errorcode)+",message="+ e.message)
</pre>
== Save Netscaler config, Reboot & Verify ==
*Need to remove repeating code lines for connection from each function
<syntaxhighlight lang="python">
import sys
import time
from nssrc.com.citrix.netscaler.nitro.exception.nitro_exception import nitro_exception
from nssrc.com.citrix.netscaler.nitro.resource.base.base_resource import base_resource
from nssrc.com.citrix.netscaler.nitro.service.nitro_service import nitro_service
from nssrc.com.citrix.netscaler.nitro.resource.stat.system.system_stats import system_stats
class ns_edit :
def __init__(self, ip, username, password):
self.ip = ip
self.username = username
self.password = password
try:
self.client = nitro_service(self.ip,"http")
self.client.set_credential(self.username,self.password)
except nitro_exception as e:
print("Exception::errorcode="+str(e.errorcode)+",message="+ e.message)
def uptime(self):
try:
node = system_stats.get(self.client)
for i in node:
print "Last reboot time is %s" %i.starttime
except nitro_exception as e:
print("Exception::errorcode="+str(e.errorcode)+",message="+ e.message)
def reboot(self):
try:
self.client.reboot(True)
print "Node rebooted"
except nitro_exception as e:
print("Exception::errorcode="+str(e.errorcode)+",message="+ e.message)
if __name__ == '__main__':
ns = ns_edit("10.107.88.78", "nsroot", "pwd@123")
ns.uptime()
ns.reboot()
time.sleep(60)
ns.uptime()
</syntaxhighlight>
= Python Dice =
<syntaxhighlight lang="python">
#!/usr/bin/env python3
from tkinter import *
from random import randint
import sys
root = Tk()
header = Label(root, text="Dice App v1.0", font=("Times", 35))
header.pack()
photo = PhotoImage(file="dice.gif")
label = Label(image=photo)
label.pack(side=RIGHT)
e = Entry(root, width=10, font=("Helvetica", 30, "bold"), justify=CENTER)
e.pack()
e.delete(0, END)
e.insert(0, "Ready!!")
def dice():
e.delete(0, END)
e.insert(0, randint(1,6))
def exit():
print("Bye Bye!!")
sys.exit()
a = Button(root, text="Roll Dice", width=20, command=dice)
a.pack()
f = Button(root, text="Exit App", width=20, fg="Red", command=exit)
f.pack()
root.geometry("400x200")
root.mainloop()
</syntaxhighlight>
= Weather Parcer =
*Need to register a account & get API Key from http://openweathermap.org
<syntaxhighlight lang="py>
import socket
import json
s = socket.socket()
addr = socket.getaddrinfo('api.openweathermap.org', 80)
s.connect(addr[0][4])
s.send(b'GET http://api.openweathermap.org/data/2.5/weather?q=Bangalore&appid=84d996853fcc4db149bc40acb09a3ef7_1 HTTP/1.0\r\n\r\n')
html = s.recv(1000)
div = html.split(b'\r\n\r\n')[-1]
data = json.loads(div)
main = data['weather'][0]['main']
print("Main =", main)
desc = data['weather'][0]['description']
print("Description =", desc)
Celcius = data['main']['temp'] - 273.15
rnd = str(round(Celcius,2))
print("Temperature =", rnd, "\xb0C")
pres = data['main']['pressure']
print("Pressure =", pres, "mbar")
hum = data['main']['humidity']
print("Humidity =", hum, "%")
vis = int(data['visibility'])/1000
print("Visibility =", vis, "Km?")
speed = data['wind']['speed']
try:
dir = data['wind']['deg']
except KeyError:
dir = 0
rnd2 = str(round(dir,2))
print("Wind Speed =", speed, "Km/h?")
print("Wind Direction =", rnd2, "Degree")
</syntaxhighlight>
= Plot Temperature using Thingspeak =
*Need to register a account & get API Key from http://openweathermap.org
*Need to register a account & get API Key from http://thingspeak.com
== Upload Temperature ==
<syntaxhighlight lang="py>
import socket
import json
from time import strftime,localtime,sleep
while True:
# Fetch Temperature from openweathermap.org
s = socket.socket()
addr = socket.getaddrinfo('api.openweathermap.org', 80)
s.connect(addr[0][4])
s.send(b'GET http://api.openweathermap.org/data/2.5/weather?q=Bangalore&appid=84d996853fcc4db149bc40acb09a3ef7_1 HTTP/1.0\r\n\r\n')
html = s.recv(1000)
div = html.split(b'\r\n\r\n')[-1]
data = json.loads(div.decode('utf-8'))
cel = data['main']['temp'] - 273.15
temp = float(round(cel,2))
# Upload Temperature Data to thingspeak.com
s2 = socket.socket()
addr2 = socket.getaddrinfo('api.thingspeak.com', 80)
s2.connect(addr2[0][4])
s2.send(b'POST https://api.thingspeak.com/update?api_key=XB15HC17CZH6KYMV_1&field1=%f HTTP/1.0\r\n\r\n' % temp)
html2 = s2.recv(1000)
div2 = html2.split(b' ')[1]
# Check if logged data successfully
if int(div2) == 200:
currtime = strftime("%H:%M:%S %d-%b-%Y", localtime())
print("Uploaded Data @", currtime,":", "Temp =", temp, "\xb0C")
else:
print("HTTP Error:", int(div2))
# Wait for 10 minutes
sleep(600)
</syntaxhighlight>
== Bulk Upload Parameters ==
<syntaxhighlight lang="python">
import socket
import json
from time import strftime,localtime,sleep
import requests
# Fetch Temperature from openweathermap.org
def weather():
s = socket.socket()
addr = socket.getaddrinfo('api.openweathermap.org', 80)
s.connect(addr[0][4])
s.send(b'GET http://api.openweathermap.org/data/2.5/weather?q=Bangalore&units=metric&appid=84d996853fcc4db149bc40acb09a3ef7_1 HTTP/1.0\r\n\r\n')
html = s.recv(1000)
s.close()
div = html.split(b'\r\n\r\n')[-1]
data = json.loads(div.decode('utf-8'))
return data
def thing():
# Call weather function
data = weather()
temp = data['main']['temp']
pres = data['main']['pressure']
hum = data['main']['humidity']
vis = int(data['visibility'])/1000
speed = data['wind']['speed']
try:
dir = data['wind']['deg']
except KeyError:
dir = 0
ctime = strftime("%Y-%m-%d %H:%M:%S +0530", localtime())
print("Data Fetched @",ctime,":","Temp(C)=",temp,"\xb0C","Press(hPa)=",pres,"Hum(%)=",hum,"Vis(Km)=",vis,"Speed(m/s)=",speed,"Dir(deg)=",dir)
# Upload Temperature Data to thingspeak.com
payload = {"write_api_key":"G97NZS1XWXCS798C_1","updates":[{"created_at":ctime,"field1":temp,"field2":pres,"field3":hum,"field4":vis,"field5":speed,"field6":dir}]}
url = 'https://api.thingspeak.com/channels/396314/bulk_update.json'
headers = {'content-type': 'application/json'}
response = requests.post(url, data=json.dumps(payload), headers=headers)
return response
while True:
response = thing()
# Check if logged data successfully
if response.status_code == 202:
print("Data successfully uploaded :", response.status_code)
else:
print("HTTP Error Code:", response.status_code)
# Wait for 10 minutes
sleep(600)
</syntaxhighlight>
== Slack Webhook POST ==
<syntaxhighlight lang="python">
#!/usr/bin/python
import json
import requests
requests.packages.urllib3.disable_warnings()
# Create the webhook at https://my.slack.com/services/new/incoming-webhook/
webhook_url = 'https://hooks.slack.com/services/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'
slack_data = {'text': "Sev1 Incident: Pool is Down :skull:"}
response = requests.post(
webhook_url, data=json.dumps(slack_data),
headers={'Content-Type': 'application/json'}
)
if response.status_code != 200:
raise ValueError(
'Request to slack returned an error %s, the response is:\n%s'
% (response.status_code, response.text)
)
</syntaxhighlight>
<br />
|
Latest revision as of 14:11, 5 May 2018
Code Snippets
Print without next line
import sys sys.stdout.write('.')
You may also need to call to ensure stdout is flushed immediately:
sys.stdout.flush()
Regex
- Extract a particular value(like 10.5, 100.0):
value = re.findall(r'([0-9]{3}.[0-9]|[0-9]{2}.[0-9]|[0-9].[0-9])(?=\sid)', output) # Print all matches print value # Print 2nd Match print value[1]
Arrays
- Using Array
x = [1,2,3,4,5] for i in range(5): print x[i]
- Append to Array
f = [] for i in range(30): f.append(i) print f
Pause
import time time.sleep(5)
Print File Extensions
import os.path extension = os.path.splitext(filename)[1][1:]
EasySNMP
This snippet needs more testing. |
sudo apt-get install libsnmp-dev snmp-mibs-downloader sudo apt-get install gcc python-dev sudo pip install easysnmp
from easysnmp import snmp_get snmp_get('.1.3.6.1.4.1.2021.10.1.3.1', hostname='10.107.88.93', community='public', version=2)
Calculator
#!/usr/bin/python #Define Function def calc(): operation = raw_input(''' Please type operator: + for addition - for subtraction * for multiplication / for division ''') #try: num1 = float(input('First Number: ')) num2 = float(input('Second Number: ')) #except: #print("Invalid Input") if operation == '+': add = num1 + num2 print("Addition is: %d" % int(add)) elif operation == '-': sub = num1 - num2 print("Subtraction is %d:" %int(sub)) elif operation == '*': mul = num1 * num2 print("Multiplication is %d:" %int(mul)) elif operation == '/': div = num1 / num2 print("Division is %.2f:" %float(div)) else: print("Invalid Operator") again() def again(): calc_again = raw_input(''' Do you want to calculate again? Please type Y for Yes or N for No. ''') if calc_again.upper() == "Y": calc() elif calc_again.upper() == "N": print('See you later on.. Bye') else: again() def welcome(): print(''' Welcome to My Calculator ''') #Call Function welcome() calc()
Schedule output
sudo pip install schedule
import schedule import time def job(): print("I'm working...") schedule.every(1).second.do(job) schedule.every(1).minutes.do(job) schedule.every().hour.do(job) schedule.every().day.at("21:14").do(job) while 1: schedule.run_pending() time.sleep(1)
Fill Routing Table of Remote Server
import paramiko ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect( '192.168.1.7', username = 'root', password = 'pwd@123' ) for x in range(1,256): for y in range(256): for z in range(256): for a in range(1): ssh.exec_command( "route add -net %d.%d.%d.%d netmask 255.255.255.0 gw 192.168.1.1 " %(x,y,z,a)) ssh.close()
Paramiko execute commands
Source: sebastiandahlgren.se
import paramiko import time import select import sys # NetScaler/Server Parameters host = '10.107.88.78' user = 'nsroot' passwd = 'pwd@123' cmd = 'show runningconfig' i = 1 # Try to connect to the host, Retry a few times if it fails. while True: print "Trying to connect to %s (%i/10)" % (host, i) try: ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect( host, username = user, password = passwd ) print "Connected to %s" % host break except paramiko.AuthenticationException: print "Authentication failed when connecting to %s" % host sys.exit(1) except: print "Could not SSH to %s, waiting for it to start" % host i += 1 time.sleep(2) # If we could not connect within time limit if i == 10: print "Could not connect to %s. Giving up" % host sys.exit(1) # Send the command (non-blocking) stdin, stdout, stderr = ssh.exec_command( cmd ) # Wait for the command to terminate while not stdout.channel.exit_status_ready(): # Only print data if there is data to read in the channel if stdout.channel.recv_ready(): rl, wl, xl = select.select([stdout.channel], [], [], 0.0) if len(rl) > 0: # Print data from stdout print stdout.channel.recv(1024), # Disconnect from the host print "\nCommand done, closing SSH connection" ssh.close()
Docker add multiple servers remotely
import paramiko import time ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect( '10.107.88.93', username = 'aman', password = 'pwd@123' ) for i in range(44400,44405): ssh.exec_command("docker run --name docker-nginx%d -p %d:8000 -d nginx" %(i,i)) print("Docker server created..") time.sleep(1) print("Done") ssh.close()
CPU check of Remote Server
import paramiko from terminalplot import plot ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect( '10.107.88.93', username = 'aman', password = 'pwd@123' ) for i in range(10): stdin, stdout, stderr = ssh.exec_command( 'top -bn2 | grep "Cpu(s)"' ) output = stdout.read() stat = output.split() idle = float(stat[24]) cpu = (100-idle) print cpu ssh.close()
TCPDump
- Print captured packets
import subprocess as sub p= sub.Popen(('sudo', 'tcpdump', '-l'), stdout=sub.PIPE) for row in iter(p.stdout.readline, b''): print row.rstrip()
- Print Packet Length
import subprocess as sub import re try: p = sub.Popen(('sudo', 'tcpdump', '-l'), stdout=sub.PIPE) for row in iter(p.stdout.readline, b''): y = re.search('length (\d+)', row) if y: x = int(y.group(1)) if x !=0: print x except KeyboardInterrupt: print("User killed command.")
- Terminal Plot
sudo pip install terminalplot
from terminalplot import plot #import gnuplotlib as gp import subprocess as sub import re x = 0 a = [] b = [] try: p = sub.Popen(('sudo', 'tcpdump', '-l'), stdout=sub.PIPE) for row in iter(p.stdout.readline, b''): y = re.search('length (\d+)', row) if y: z = int(y.group(1)) if z != 0: x += 1 print x b.append(z) a.append(x) if x == 10: plot(a,b) print a print b # Remove below # if you don't want continuous output # break x = 0 a = [] b = [] except KeyboardInterrupt: print("User killed command.")
- Plot using Stdout
import subprocess as sub import re import sys x = 0 try: p = sub.Popen(('sudo', 'tcpdump', '-l'), stdout=sub.PIPE) for row in iter(p.stdout.readline, b''): y = re.search('length (\d+)', row) if y: z = int(y.group(1)) x += 1 if (z < 190): for i in range(z): sys.stdout.write('=') print (">") except KeyboardInterrupt: print("User killed command.")
Speaking Passwords
- eSpeak Password
import os def check(): x = int(input("Input your password: ")) if (x == 1234): os.system("espeak 'correct password'") else: os.system("espeak 'wrong password'") def again(): check() again() check() again()
- Using pyttsx
import pyttsx engine = pyttsx.init() def check(): x = int(input("Input your password: ")) if (x == 1234): engine.say('correct password') engine.runAndWait() else: engine.say('wrong password') engine.runAndWait() def again(): check() again() check() again()
NetScaler NITRO REST Server Monitor using SSH
- Monitors CPU Utilization of Backend Server, Disables service if CPU Exceeds 50%
- Requires: NitroRestClient
import paramiko import sys import re import time from bin import NitroRestClient # Netscaler Parameters nsip = "10.107.88.78" nsun = "nsroot" nspwd = "pwd@123" nssrv = "Ubuntu_Server" # Server Parameters srvip = "10.107.88.93" srvun = "aman" srvpwd = "Passion@123" # Loop Logic variables x = 1 y = 0 # Attempt SSH Connection ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect( srvip, username = srvun, password = srvpwd ) # Attempt instantiation of NitroRestClient try: Client = NitroRestClient.NitroRestClient(nsip, nsun, nspwd) except: print("Couldn't create Nitro Session, check username and password and network connectivity") sys.exit(1) # Check servername for validity try: normalizedservername = Client.servernamecheck(nssrv) except ValueError as e: print("Invalid server name was provided. Please check server name and try again.") sys.exit(2) # Main Program try: while True: stdin, stdout, stderr = ssh.exec_command( 'top -bn2 | grep "Cpu(s)"' ) output = stdout.read() value = re.findall(r'([0-9]{3}.[0-9]|[0-9]{2}.[0-9]|[0-9].[0-9])\sid', output) cpu = (100-float(value[1])) print cpu if cpu < 50: if x == 1: print("CPU is normal..") time.sleep(5) else: print("Server enabled") Client.enablephysicalserver(normalizedservername) x = 1 y = 0 else: if y == 1: print("Server remains disabled..") time.sleep(5) else: Client.disablephysicalserver(normalizedservername) print("Server disabled") time.sleep(1) y = 1 x = 0 except KeyboardInterrupt: print("Script Killed by user") ssh.close() sys.exit(0)
NetScaler NITRO REST Server Monitor using SNMP
- Monitors CPU Utilization of Backend Server, Disables service if Load Exceeds 0.50
- Requires: NitroRestClient
- Requires SNMP configured on Server & SNMP Utils installed on the PC where script is run.
import time import subprocess import sys import re from bin import NitroRestClient # Netscaler Parameters & Credentials nsip = "10.107.88.78" nsun = "nsroot" nspwd = "pwd@123" nssrv = "Ubuntu_Server" # Loop Logic variables x = 1 y = 0 # Attempt instantiation of NitroRestClient try: Client = NitroRestClient.NitroRestClient(nsip, nsun, nspwd) except: print("Couldn't create Nitro Session, check username and password and network connectivity") sys.exit(1) # Check servername for validity try: normalizedservername = Client.servernamecheck(nssrv) except ValueError as e: print("Invalid server name was provided. Please check server name and try again.") sys.exit(2) # Main Program try: while True: snmpload = subprocess.Popen(["snmpget", "-v", "2c", "-O", "qv", "-c", "public", "10.107.88.93", ".1.3.6.1.4.1.2021.10.1.3.1"], stdout=subprocess.PIPE) output, err = snmpload.communicate() value = re.sub(r'^"|"$', '', output) load = float(value) print(load) if load < 0.10: if x == 1: print("Load is normal..") time.sleep(5) else: print("Server enabled") Client.enablephysicalserver(normalizedservername) x = 1 y = 0 else: if y == 1: print("Server remains disabled..") time.sleep(5) else: Client.disablephysicalserver(normalizedservername) print("Server disabled") time.sleep(1) y = 1 x = 0 except KeyboardInterrupt: print("Script Killed by user") sys.exit(0)
Reboot Netscaler via API Call using Curl
This script is under construction. |
- Needs Authentication alerts
- Needs NS Parameters to be set globally
- Needs to confirm API execution via HTTP Status Codes
import subprocess as sub import time # Nitro API calls using Curl uptimeapi = ('curl', '-s', '-k', '-X', 'GET', '-H', 'Content-Type:application/json', '--basic', '--user', 'nsroot:pwd@123', 'http://10.107.88.78/nitro/v1/stat/system?attrs=starttime') rebootapi = ('curl', '-s', '-k', '-X', 'POST', '-H', 'Content-Type:application/vnd.com.citrix.netscaler.reboot+json', '--basic', '--user', 'nsroot:pwd@123', '-d', '{"reboot":{"warm":true}}', 'http://10.107.88.78/nitro/v1/config/reboot') print uptimeapi print rebootapi uptime1 = sub.Popen(uptimeapi, stdout=sub.PIPE) output1, err = uptime1.communicate() print output1 sub.Popen(rebootapi, stdout=sub.PIPE) time.sleep(50) uptime2 = sub.Popen(uptimeapi, stdout=sub.PIPE) output2, err = uptime2.communicate() print output2 if (output1 == output2): print "Reboot unsuccessful" else: print "Reboot successful"
- Under testing version of above with using parameters
import subprocess as sub import time # Netscaler Parameters host = "10.107.88.78" username = "nsroot" passwd = "pwd@123" # Boolean true means Warm reboot, else false boolean = "true" # Nitro API calls using Curl uptimeapi = "curl -s -k -X GET -H 'Content-Type:application/json' --basic --user %s:%s http://%s/nitro/v1/stat/system?attrs=starttime" %(username, passwd, host) rebootapi = "curl -s -k -X POST -H 'Content-Type:application/vnd.com.citrix.netscaler.reboot+json' --basic --user %s:%s -d '{'reboot':{'warm':boolean}}' http://%s/nitro/v1/config/reboot" %(username,passwd,boolean,host) print uptimeapi print rebootapi uptime1 = sub.Popen(uptimeapi.split(), stdout=sub.PIPE) output1, err = uptime1.communicate() print output1 sub.Popen(rebootapi.split(), stdout=sub.PIPE) time.sleep(50) uptime2 = sub.Popen(uptimeapi.split(), stdout=sub.PIPE) output2, err = uptime2.communicate() print output2 if (output1 == output2): print "Reboot unsuccessful" else: print "Reboot successful"
Reboot Netscaler via API Call using SDK
Download & Install Nitro REST SDK: citrix.co.in
Save config & Reboot
import sys from nssrc.com.citrix.netscaler.nitro.exception.nitro_exception import nitro_exception from nssrc.com.citrix.netscaler.nitro.resource.base.base_resource import base_resource from nssrc.com.citrix.netscaler.nitro.service.nitro_service import nitro_service ip = "10.107.88.78" username = "nsroot" password = "pwd@123" try: client = nitro_service(ip,"http") client.set_credential(username,password) client.timeout = 20 client.save_config() client.reboot(True) client.logout() print "Config saved, Rebooting now.." except nitro_exception as e: print("Exception::errorcode="+str(e.errorcode)+",message="+ e.message)
Save Netscaler config, Reboot & Verify
- Need to remove repeating code lines for connection from each function
import sys
import time
from nssrc.com.citrix.netscaler.nitro.exception.nitro_exception import nitro_exception
from nssrc.com.citrix.netscaler.nitro.resource.base.base_resource import base_resource
from nssrc.com.citrix.netscaler.nitro.service.nitro_service import nitro_service
from nssrc.com.citrix.netscaler.nitro.resource.stat.system.system_stats import system_stats
class ns_edit :
def __init__(self, ip, username, password):
self.ip = ip
self.username = username
self.password = password
try:
self.client = nitro_service(self.ip,"http")
self.client.set_credential(self.username,self.password)
except nitro_exception as e:
print("Exception::errorcode="+str(e.errorcode)+",message="+ e.message)
def uptime(self):
try:
node = system_stats.get(self.client)
for i in node:
print "Last reboot time is %s" %i.starttime
except nitro_exception as e:
print("Exception::errorcode="+str(e.errorcode)+",message="+ e.message)
def reboot(self):
try:
self.client.reboot(True)
print "Node rebooted"
except nitro_exception as e:
print("Exception::errorcode="+str(e.errorcode)+",message="+ e.message)
if __name__ == '__main__':
ns = ns_edit("10.107.88.78", "nsroot", "pwd@123")
ns.uptime()
ns.reboot()
time.sleep(60)
ns.uptime()
Python Dice
#!/usr/bin/env python3
from tkinter import *
from random import randint
import sys
root = Tk()
header = Label(root, text="Dice App v1.0", font=("Times", 35))
header.pack()
photo = PhotoImage(file="dice.gif")
label = Label(image=photo)
label.pack(side=RIGHT)
e = Entry(root, width=10, font=("Helvetica", 30, "bold"), justify=CENTER)
e.pack()
e.delete(0, END)
e.insert(0, "Ready!!")
def dice():
e.delete(0, END)
e.insert(0, randint(1,6))
def exit():
print("Bye Bye!!")
sys.exit()
a = Button(root, text="Roll Dice", width=20, command=dice)
a.pack()
f = Button(root, text="Exit App", width=20, fg="Red", command=exit)
f.pack()
root.geometry("400x200")
root.mainloop()
Weather Parcer
- Need to register a account & get API Key from http://openweathermap.org
import socket
import json
s = socket.socket()
addr = socket.getaddrinfo('api.openweathermap.org', 80)
s.connect(addr[0][4])
s.send(b'GET http://api.openweathermap.org/data/2.5/weather?q=Bangalore&appid=84d996853fcc4db149bc40acb09a3ef7_1 HTTP/1.0\r\n\r\n')
html = s.recv(1000)
div = html.split(b'\r\n\r\n')[-1]
data = json.loads(div)
main = data['weather'][0]['main']
print("Main =", main)
desc = data['weather'][0]['description']
print("Description =", desc)
Celcius = data['main']['temp'] - 273.15
rnd = str(round(Celcius,2))
print("Temperature =", rnd, "\xb0C")
pres = data['main']['pressure']
print("Pressure =", pres, "mbar")
hum = data['main']['humidity']
print("Humidity =", hum, "%")
vis = int(data['visibility'])/1000
print("Visibility =", vis, "Km?")
speed = data['wind']['speed']
try:
dir = data['wind']['deg']
except KeyError:
dir = 0
rnd2 = str(round(dir,2))
print("Wind Speed =", speed, "Km/h?")
print("Wind Direction =", rnd2, "Degree")
Plot Temperature using Thingspeak
- Need to register a account & get API Key from http://openweathermap.org
- Need to register a account & get API Key from http://thingspeak.com
Upload Temperature
import socket
import json
from time import strftime,localtime,sleep
while True:
# Fetch Temperature from openweathermap.org
s = socket.socket()
addr = socket.getaddrinfo('api.openweathermap.org', 80)
s.connect(addr[0][4])
s.send(b'GET http://api.openweathermap.org/data/2.5/weather?q=Bangalore&appid=84d996853fcc4db149bc40acb09a3ef7_1 HTTP/1.0\r\n\r\n')
html = s.recv(1000)
div = html.split(b'\r\n\r\n')[-1]
data = json.loads(div.decode('utf-8'))
cel = data['main']['temp'] - 273.15
temp = float(round(cel,2))
# Upload Temperature Data to thingspeak.com
s2 = socket.socket()
addr2 = socket.getaddrinfo('api.thingspeak.com', 80)
s2.connect(addr2[0][4])
s2.send(b'POST https://api.thingspeak.com/update?api_key=XB15HC17CZH6KYMV_1&field1=%f HTTP/1.0\r\n\r\n' % temp)
html2 = s2.recv(1000)
div2 = html2.split(b' ')[1]
# Check if logged data successfully
if int(div2) == 200:
currtime = strftime("%H:%M:%S %d-%b-%Y", localtime())
print("Uploaded Data @", currtime,":", "Temp =", temp, "\xb0C")
else:
print("HTTP Error:", int(div2))
# Wait for 10 minutes
sleep(600)
Bulk Upload Parameters
import socket
import json
from time import strftime,localtime,sleep
import requests
# Fetch Temperature from openweathermap.org
def weather():
s = socket.socket()
addr = socket.getaddrinfo('api.openweathermap.org', 80)
s.connect(addr[0][4])
s.send(b'GET http://api.openweathermap.org/data/2.5/weather?q=Bangalore&units=metric&appid=84d996853fcc4db149bc40acb09a3ef7_1 HTTP/1.0\r\n\r\n')
html = s.recv(1000)
s.close()
div = html.split(b'\r\n\r\n')[-1]
data = json.loads(div.decode('utf-8'))
return data
def thing():
# Call weather function
data = weather()
temp = data['main']['temp']
pres = data['main']['pressure']
hum = data['main']['humidity']
vis = int(data['visibility'])/1000
speed = data['wind']['speed']
try:
dir = data['wind']['deg']
except KeyError:
dir = 0
ctime = strftime("%Y-%m-%d %H:%M:%S +0530", localtime())
print("Data Fetched @",ctime,":","Temp(C)=",temp,"\xb0C","Press(hPa)=",pres,"Hum(%)=",hum,"Vis(Km)=",vis,"Speed(m/s)=",speed,"Dir(deg)=",dir)
# Upload Temperature Data to thingspeak.com
payload = {"write_api_key":"G97NZS1XWXCS798C_1","updates":[{"created_at":ctime,"field1":temp,"field2":pres,"field3":hum,"field4":vis,"field5":speed,"field6":dir}]}
url = 'https://api.thingspeak.com/channels/396314/bulk_update.json'
headers = {'content-type': 'application/json'}
response = requests.post(url, data=json.dumps(payload), headers=headers)
return response
while True:
response = thing()
# Check if logged data successfully
if response.status_code == 202:
print("Data successfully uploaded :", response.status_code)
else:
print("HTTP Error Code:", response.status_code)
# Wait for 10 minutes
sleep(600)
Slack Webhook POST
#!/usr/bin/python
import json
import requests
requests.packages.urllib3.disable_warnings()
# Create the webhook at https://my.slack.com/services/new/incoming-webhook/
webhook_url = 'https://hooks.slack.com/services/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'
slack_data = {'text': "Sev1 Incident: Pool is Down :skull:"}
response = requests.post(
webhook_url, data=json.dumps(slack_data),
headers={'Content-Type': 'application/json'}
)
if response.status_code != 200:
raise ValueError(
'Request to slack returned an error %s, the response is:\n%s'
% (response.status_code, response.text)
)
- References
{{#widget:DISQUS
|id=networkm
|uniqid=Python Scripts
|url=https://aman.awiki.org/wiki/Python_Scripts
}}