-
Notifications
You must be signed in to change notification settings - Fork 2
/
deploypage.py
163 lines (133 loc) · 5.47 KB
/
deploypage.py
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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
from lxml import etree
import dateutil.parser
from threading import Timer
from datetime import datetime
import math
import pytz
class DeployPage:
#: mwclient.Site mwcon: Connection to MediaWiki server that hosts the deployment calendar
mwcon = None
page = ""
logger = None
update_interval = 1
notify_callback = None
notify_timer = None
update_timer = None
deploy_items = {}
page_url = ''
def __init__(self, mwcon, page, logger, update_interval = 15):
""" Create a DeployPage object
:param mwclient.Site mwcon: Connection to MediaWiki server that hosts :param page
:param string page: Title of page that hosts the deployment calendar
:param int update_interval: Number of minutes between requests for the deployment page
"""
self.mwcon = mwcon
self.page = page
self.logger = logger
self.update_interval = update_interval
# Things I hate about the MW API right here...
# This is getting the full URL of the deployments page so we can create
# nice links in IRC messages
page_url_result = mwcon.api('query', **{'titles':'Deployments', 'prop':'info','inprop':'url'})
self.page_url = page_url_result['query']['pages'][page_url_result['query']['pages'].keys()[0]]['fullurl']
def start(self, notify_callback):
"""Start all the various timers"""
self.notify_callback = notify_callback
self._reparse_on_timer()
def stop(self):
if self.notify_timer:
self.notify_timer.cancel()
if self.update_timer:
self.update_timer.cancel()
def reparse(self, set_timer = False):
deploy_items = {}
def stringify_children(node):
from lxml.etree import tostring
from itertools import chain
parts = ([node.text] +
list(chain(*(stringify_children(c) for c in node.getchildren()))) +
[node.tail])
# filter removes possible Nones in texts and tails
return ''.join(filter(None, parts))
self.logger.debug("Collecting new deployment information from the server")
tree = etree.fromstring(self._get_page_html(), etree.HTMLParser())
for item in tree.xpath('//tr[@class="deploycal-item"]'):
id = item.get('id')
times = item.xpath('td/span[@class="deploycal-time-utc"]/time')
start_time = dateutil.parser.parse(times[0].get('datetime'))
end_time = dateutil.parser.parse(times[1].get('datetime'))
window = stringify_children(item.xpath('td/span[@class="deploycal-window"]')[0]) \
.replace("\n", " ") \
.strip()
owners = map(
lambda x: x.text,
item.xpath('td/span[@class="ircnick-container"]/span[@class="ircnick"]')
)
item_obj = DeployItem(id, '%s#%s' % (self.page_url, id), start_time, end_time, window, owners)
if start_time in deploy_items:
deploy_items[start_time].append(item_obj)
else:
deploy_items[start_time] = [item_obj]
self.logger.debug("Got %s items" % len(deploy_items))
self.deploy_items = deploy_items
if set_timer:
self._set_deploy_timer()
return deploy_items
def get_events(self):
return self.deploy_items
def get_current_events(self):
pass
def get_next_events(self):
"""What are the first set of DeployEvents in the future"""
ctime = datetime.now(pytz.utc)
nexttime = None
for time in sorted(self.deploy_items.keys()):
if ctime < time:
nexttime = time
break
if nexttime:
return self.deploy_items[nexttime]
else:
return []
def _get_page_html(self):
try:
return self.mwcon.parse(self.mwcon.pages[self.page].edit())['text']['*']
except Exception as ex:
self.logger.error("Could not fetch page due to exception: " + repr(ex))
return ""
def _reparse_on_timer(self):
self.reparse(set_timer = True)
if self.update_timer:
self.update_timer.cancel()
self.update_timer = Timer(self.update_interval * 60, self._reparse_on_timer)
self.update_timer.start()
def _set_deploy_timer(self):
next_events = self.get_next_events()
if len(next_events) > 0:
td = math.floor((next_events[0].start - datetime.now(pytz.utc)).total_seconds()) + 5
if self.notify_timer:
self.notify_timer.cancel()
self.logger.debug( "Setting deploy timer to %s for %s" % (td, next_events[0]))
self.notify_timer = Timer(td, self._on_deploy_timer, [next_events])
self.notify_timer.start()
def _on_deploy_timer(self, events):
self.logger.info('Deploy timer kicked. Attempting to notify.')
self.logger.debug("Num events: %s" % len(events))
self.notify_callback(events)
self._set_deploy_timer()
class DeployItem:
def __init__(self, id, url, start, end, window, owners):
self.id = id
self.url = url
self.start = start
self.end = end
self.window = window
self.owners = owners
def __repr__(self):
return "%s: (%s -> %s) %s; %s" % (
self.id,
self.start,
self.end,
self.window,
", ".join(self.owners)
)