Жизнь замечательных грибов
Здесь собраны статьи, наработки, исправление ошибок и решение проблем связанных с разработкой на языке Python
Ошибка возникла при попытке установить модуль pysnmp во FreeBSD:
|
1 |
pip install pysnmp |
Решение:
— ставим gcc, убеждаемся что /tmp смонтирован с флагом exec. Помогло. Теперь другая ошибка:
src/_fastmath.c:36:11: fatal error: ‘gmp.h’ file not found
Решение: копируем gmp.h из /usr/include/ в /usr/local/include/
Задача: считать текущий статус порта, и в зависимости от него или выключить порт, или включить.
Решение: скрипт на python
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 |
# coding=utf-8 from pysnmp.hlapi import * from pysnmp.entity.rfc3413.oneliner import cmdgen from pysnmp.proto import rfc1902 def GetPortStatus(ip,comm,oid): rez=None try: errorIndication, errorStatus, errorIndex, varBinds = next( getCmd(SnmpEngine(), CommunityData(comm), UdpTransportTarget( (ip, 161), timeout=2.0, retries=0 ), ContextData(), ObjectType(ObjectIdentity(oid))) ) if errorIndication: print(errorIndication) return rez elif errorStatus: print('%s at %s' % (errorStatus.prettyPrint(), errorIndex and varBinds[int(errorIndex) - 1][0] or '?')) return rez else: for varBind in varBinds: if str(varBind).find("up")>0: rez=1; if str(varBind).find("down")>0: rez=2; if rez==None: zx=str(varBind).split("="); rez=zx[1].replace(" ",""); except: rez=None return rez def SetPortStatus(ip,comm,oid,status): rez=None try: cmdGen = cmdgen.CommandGenerator() errorIndication, errorStatus, errorIndex, varBinds = cmdGen.setCmd( cmdgen.CommunityData(comm,mpModel=1), cmdgen.UdpTransportTarget((ip, 161)), (oid, rfc1902.Integer(status)), ) # Check for errors and print out results if errorIndication: print(errorIndication) else: if errorStatus: print('%s at %s' % ( errorStatus.prettyPrint(), errorIndex and varBinds[int(errorIndex)-1] or '?' ) ) rez=None else: for name, val in varBinds: print('%s = %s' % (name.prettyPrint(), val.prettyPrint())) rez=True except: rez=None return rez #снимает статус порта #dlink # oid=1.3.6.1.2.1.2.2.1.8.x # x - номер порта rz=GetPortStatus('12.1.14.6','XFiles','1.3.6.1.2.1.2.2.1.8.9') print "Dlink: ",rz #snr rz=GetPortStatus('72.1.14.201','X-Files','1.3.6.1.2.1.2.2.1.8.3') print "SNR: ",rz #qsw rz=GetPortStatus('72.1.114.201','X-Files','1.3.6.1.2.1.2.2.1.8.3') print "QSW: ",rz #устанавливаем статус порта #up=1,down=2 #oid=1.3.6.1.2.1.2.2.1.7.9.x # x - номер порта #Dlink rz=SetPortStatus('72.8.114.4','X-Files','1.3.6.1.2.1.2.2.1.7.9',1) print rz; #SNR rz=SetPortStatus('72.7.114.196','X-Files','1.3.6.1.2.1.2.2.1.7.8',1) print rz; #QSR rz=SetPortStatus('12.1.114.100','X-Files','1.3.6.1.2.1.2.2.1.7.3',1) print rz; |
Есть несколько модулей, которые помогают работать с snmp на python. Гугл в помощь какие. Наиболее популярный: pysnmp. В Ubuntu ставится как:
|
1 |
pip install pysnmp |
Напишем на Python аналог команды из консоли, которая получает список мак адресов на портах свичей Dlink:
|
1 |
snmpwalk -v2c -c XFiles 172.18.114.6 1.3.6.1.2.1.17.7.1.2.2.1.2 |
Код:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
from pysnmp.entity.rfc3413.oneliner import cmdgen cmdGen = cmdgen.CommandGenerator() errorIndication, errorStatus, errorIndex, varBindTable = cmdGen.nextCmd( cmdgen.CommunityData('XFiles'), cmdgen.UdpTransportTarget(('172.18.114.6', 161)), '1.3.6.1.2.1.17.7.1.2.2.1.2' ) if errorIndication: print(errorIndication) else: if errorStatus: print('%s at %s' % ( errorStatus.prettyPrint(), errorIndex and varBindTable[-1][int(errorIndex)-1] or '?' ) ) else: for varBindTableRow in varBindTable: for name, val in varBindTableRow: print('%s = %s' % (name.prettyPrint(), val.prettyPrint())) |
На удивление, на такую простую вещь как создание потоков, потратил пару часов, из-за обилия в интернетиках примеров с ошибками в коде. Потому привожу рабочий пример:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 |
#!/usr/local/bin/python # coding=utf-8 import threading import time debug=True #количество одновременно запускаемых потоков thread_count=10; def putlog(txt): if (debug==True): print txt def worker(th): putlog("--стартовал поток "+str(th)); i=0 while i<config.thread_count: putlog("--поток "+str(th)+", шаг "+str(i)) time.sleep(1) i=i+1 def main(): putlog("-скрипт стартовал!"); #запускаем потоки for i in xrange(thread_count): thread = threading.Thread(target=worker, args=(i,)).start() #ждем пока потоки завершатся while threading.active_count() >1: putlog("!ждем завершения потоков") time.sleep(1) putlog("-скрипт закончил работу!"); main() |
Результат работы скрипта:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
-скрипт стартовал! --стартовал поток 0 --поток 0, шаг 0 --стартовал поток 1 --поток 1, шаг 0 --стартовал поток 2 !ждем завершения потоков --поток 2, шаг 0 --поток 0, шаг 1 !ждем завершения потоков --поток 1, шаг 1 --поток 2, шаг 1 --поток 0, шаг 2 !ждем завершения потоков--поток 1, шаг 2 --поток 2, шаг 2 !ждем завершения потоков -скрипт закончил работу |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart msg = MIMEMultipart('alternative') html_body="Внимание! У вас есть задача (№"+str(id)+") в sbss, которая уже длительное время находится в статусе ожидания! Через 3 дня она будет автоматически закрыта.. <br/><a href=https://oeirfoeriferoife.ru>"+name+"</a>".encode('utf-8') msg.set_charset("utf-8") msg['Subject'] = "Уведомление о длительной задаче №"+str(id); msg['From'] = sbss_from_email msg['To'] = m_arr["email"] part = MIMEText(html_body, 'html', 'UTF-8') msg.attach(part) s = smtplib.SMTP(sbss_from_server) s.login(sbss_from_email,sbss_from_pass); s.set_debuglevel(1); s.sendmail(sbss_from_email, m_arr["email"], msg.as_string()) s.quit() |