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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
#| /usr/bin/python2
import requests
import time
__all__ = ["init", "get_sounds", "Client", "InitError", "RequestError"]
BASE_URL = "https://api.pushover.net/1/"
MESSAGE_URL = BASE_URL + "messages.json"
USER_URL = BASE_URL + "users/validate.json"
SOUND_URL = BASE_URL + "sounds.json"
RECEIPT_URL = BASE_URL + "receipts/"
SOUNDS = None
TOKEN = None
def get_sounds():
global SOUNDS
if not SOUNDS:
request = Request("get", SOUND_URL, {})
SOUNDS = request.answer["sounds"]
return SOUNDS
def init(token, sound=False):
global TOKEN
TOKEN = token
if sound:
return get_sounds()
class InitError(Exception):
def __str__(self):
return "Init the pushover module by calling the init function"
class RequestError(Exception):
def __init__(self, errors):
Exception.__init__(self)
self.errors = errors
def __str__(self):
return "\n==> " + "\n==> ".join(self.errors)
class Request:
def __init__(self, request_type, url, payload):
if not TOKEN:
raise InitError
payload["token"] = TOKEN
request = getattr(requests, request_type)(url, params=payload)
self.answer = request.json()
if 400 <= request.status_code < 500:
raise RequestError(self.answer["errors"])
def __str__(self):
print self.answer
class MessageRequest(Request):
def __init__(self, payload):
Request.__init__(self, "post", MESSAGE_URL, payload)
self.receipt = None
if payload.get("priority", 0) == 2:
self.receipt = self.answer["receipt"]
self.parameters = ["expired", "called_back", "expired"]
for parameter in self.parameters:
setattr(self, parameter, False)
setattr(self, parameter + "_at", 0)
def poll(self):
if (self.receipt and not any(getattr(self, parameter)
for parameter in self.parameters)):
request = Request("get", RECEIPT_URL + self.receipt + ".json", {})
for parameter in self.parameters:
setattr(self, parameter, request.answer[parameter])
setattr(self, parameter + "_at",
request.answer[parameter + "_at"])
return request
class Client:
def __init__(self, user, device=None):
self.user = user
self.device = device
self.devices = []
def verify(self, device=None):
payload = {"user": self.user}
device = device or self.device
if device:
payload["device"] = device
try:
request = Request("post", USER_URL, payload)
except RequestError:
return False
self.devices = request.answer["devices"]
return True
def send_message(self, message, **kwords):
valid_keywords = ["title", "priority", "sound", "callback",
"timestamp", "url", "url_title", "device"]
payload = {"message": message, "user": self.user}
if self.device:
payload["device"] = self.device
for key, value in kwords.iteritems():
if key not in valid_keywords:
raise ValueError("{0}: invalid message parameter".format(key))
if key == "timestamp" and value:
payload[key] = time.time()
elif key == "sound":
if not SOUNDS:
get_sounds()
if value not in SOUNDS:
raise ValueError("{0}: invalid sound".format(value))
else:
payload[key] = value
elif value:
payload[key] = value
return MessageRequest(payload)
if __name__ == "__main__":
from argparse import ArgumentParser
parser = ArgumentParser(description="Send a message to pushover.")
parser.add_argument("--token", help="Pushover application token",
required=True)
parser.add_argument("--client", "-c", help="Pushover client ID",
required=True)
parser.add_argument("message", help="message to send")
parser.add_argument("--title", "-t", help="message title")
parser.add_argument("--priority", "-p", help="message priority")
parser.add_argument("--url", help="additional url")
parser.add_argument("--url-title", help="additional url title")
args = parser.parse_args()
init(args.token)
Client(args.client).send_message(args.message, title=args.title,
priority=args.priority, url=args.url,
url_title=args.url_title, timestamp=True)
|
