Update PyNMA to hybrid v1.0.

Code refactored, cleaned and PEP8 then released as v1.01 from a v1.0 base.
This commit is contained in:
JackDandy 2015-06-05 01:12:57 +01:00
parent 9ffa5e5cbe
commit bcb70efed2
3 changed files with 97 additions and 70 deletions

View file

@ -20,6 +20,7 @@
* Change handling of general HTTP error response codes to prevent issues
* Add handling for CloudFlare custom HTTP response codes
* Fix to correctly load local libraries instead of system installed libraries
* Update PyNMA to hybrid v1.0
[develop changelog]
* Update Requests library 2.7.0 (ab1f493) to 2.7.0 (8b5e457)

View file

@ -1,4 +1,5 @@
#!/usr/bin/python
from pynma import PyNMA
__version__ = '1.01'
from .pynma import PyNMA

View file

@ -1,137 +1,162 @@
#!/usr/bin/python
from . import __version__
from xml.dom.minidom import parseString
from httplib import HTTPSConnection
from urllib import urlencode
__version__ = "0.1"
try:
from http.client import HTTPSConnection
except ImportError:
from httplib import HTTPSConnection
API_SERVER = 'nma.usk.bz'
ADD_PATH = '/publicapi/notify'
try:
from urllib.parse import urlencode
except ImportError:
from urllib import urlencode
USER_AGENT="PyNMA/v%s"%__version__
def uniq_preserve(seq): # Dave Kirby
# Order preserving
seen = set()
return [x for x in seq if x not in seen and not seen.add(x)]
def uniq(seq):
# Not order preserving
return {}.fromkeys(seq).keys()
class PyNMA(object):
"""PyNMA(apikey=[], developerkey=None)
"""
http://www.notifymyandroid.com/api.jsp
PyNMA(apikey=None, developerkey=None)
takes 2 optional arguments:
- (opt) apykey: might me a string containing 1 key or an array of keys
- (opt) apykey: a string containing 1 key or an array of keys
- (opt) developerkey: where you can store your developer key
"""
def __init__(self, apikey=[], developerkey=None):
def __init__(self, apikey=None, developerkey=None):
self._developerkey = None
self.developerkey(developerkey)
self.api_server = 'https://www.notifymyandroid.com'
self.add_path = '/publicapi/notify'
self.user_agent = 'PyNMA/v%s' % __version__
key = []
if apikey:
if type(apikey) == str:
apikey = [apikey]
self._apikey = uniq(apikey)
key = (apikey, [apikey])[str == type(apikey)]
self._apikey = self.uniq(key)
@staticmethod
def uniq(seq):
# Not order preserving
return list({}.fromkeys(seq).keys())
def addkey(self, key):
"Add a key (register ?)"
if type(key) == str:
if not key in self._apikey:
"""
Add a key (register ?)
"""
if str == type(key):
if key not in self._apikey:
self._apikey.append(key)
elif type(key) == list:
elif list == type(key):
for k in key:
if not k in self._apikey:
if k not in self._apikey:
self._apikey.append(k)
def delkey(self, key):
"Removes a key (unregister ?)"
if type(key) == str:
"""
Removes a key (unregister ?)
"""
if str == type(key):
if key in self._apikey:
self._apikey.remove(key)
elif type(key) == list:
elif list == type(key):
for k in key:
if key in self._apikey:
self._apikey.remove(k)
def developerkey(self, developerkey):
"Sets the developer key (and check it has the good length)"
if type(developerkey) == str and len(developerkey) == 48:
"""
Sets the developer key (and check it has the good length)
"""
if str == type(developerkey) and 48 == len(developerkey):
self._developerkey = developerkey
def push(self, application="", event="", description="", url="", priority=0, batch_mode=False):
"""Pushes a message on the registered API keys.
def push(self, application='', event='', description='', url='', content_type=None, priority=0, batch_mode=False, html=False):
"""
Pushes a message on the registered API keys.
takes 5 arguments:
- (req) application: application name [256]
- (req) event: event name [1000]
- (req) description: description [10000]
- (opt) url: url [512]
- (opt) contenttype: Content Type (act: None (plain text) or text/html)
- (opt) priority: from -2 (lowest) to 2 (highest) (def:0)
- (opt) batch_mode: call API 5 by 5 (def:False)
- (opt) html: shortcut for content_type=text/html
Warning: using batch_mode will return error only if all API keys are bad
cf: http://nma.usk.bz/api.php
http://www.notifymyandroid.com/api.jsp
"""
datas = {
'application': application[:256].encode('utf8'),
'event': event[:1024].encode('utf8'),
'description': description[:10000].encode('utf8'),
'priority': priority
}
datas = {'application': application[:256].encode('utf8'),
'event': event[:1000].encode('utf8'),
'description': description[:10000].encode('utf8'),
'priority': priority}
if url:
datas['url'] = url[:512]
datas['url'] = url[:2000]
if self._developerkey:
datas['developerkey'] = self._developerkey
if 'text/html' == content_type or True == html: # Currently only accepted content type
datas['content-type'] = 'text/html'
results = {}
if not batch_mode:
for key in self._apikey:
datas['apikey'] = key
res = self.callapi('POST', ADD_PATH, datas)
res = self.callapi('POST', self.add_path, datas)
results[key] = res
else:
for i in range(0, len(self._apikey), 5):
datas['apikey'] = ",".join(self._apikey[i:i+5])
res = self.callapi('POST', ADD_PATH, datas)
results[datas['apikey']] = res
datas['apikey'] = ','.join(self._apikey)
res = self.callapi('POST', self.add_path, datas)
results[datas['apikey']] = res
return results
def callapi(self, method, path, args):
headers = { 'User-Agent': USER_AGENT }
if method == "POST":
headers['Content-type'] = "application/x-www-form-urlencoded"
http_handler = HTTPSConnection(API_SERVER)
headers = {'User-Agent': self.user_agent}
if 'POST' == method:
headers['Content-type'] = 'application/x-www-form-urlencoded'
http_handler = HTTPSConnection(self.api_server)
http_handler.request(method, path, urlencode(args), headers)
resp = http_handler.getresponse()
try:
res = self._parse_reponse(resp.read())
except Exception, e:
res = {'type': "pynmaerror",
'code': 600,
'message': str(e)
}
res = self._parse_response(resp.read())
except Exception as e:
res = {'type': 'pynmaerror',
'code': 600,
'message': str(e)}
pass
return res
def _parse_reponse(self, response):
@staticmethod
def _parse_response(response):
root = parseString(response).firstChild
for elem in root.childNodes:
if elem.nodeType == elem.TEXT_NODE: continue
if elem.tagName == 'success':
res = dict(elem.attributes.items())
res['message'] = ""
res['type'] = elem.tagName
if elem.TEXT_NODE == elem.nodeType:
continue
if 'success' == elem.tagName:
res = dict(list(elem.attributes.items()))
res['message'] = ''
res['type'] = elem.tagName
return res
if elem.tagName == 'error':
res = dict(elem.attributes.items())
if 'error' == elem.tagName:
res = dict(list(elem.attributes.items()))
res['message'] = elem.firstChild.nodeValue
res['type'] = elem.tagName
res['type'] = elem.tagName
return res