-
Notifications
You must be signed in to change notification settings - Fork 0
/
trellogrove.py
275 lines (223 loc) · 8.44 KB
/
trellogrove.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
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
""":mod:`trellogrove` --- Trello IRC Notibot
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
"""
import json
import logging
import os.path
import urllib
import urlparse
from google.appengine.api.urlfetch import fetch
from google.appengine.ext.db import Model, StringProperty, put
from google.appengine.ext.deferred import defer
from jinja2 import Environment, FileSystemLoader
from webapp2 import RequestHandler, WSGIApplication
class Setting(Model):
"""Settings pair."""
name = StringProperty(required=True, indexed=True)
value = StringProperty(multiline=True)
def has_settings_complete():
"""Returns ``True`` only if the all settings are completely filled."""
settings = get_settings()
return (settings.get('trello.app_key') and
settings.get('trello.oauth_token') and
settings.get('webhook_url'))
def get_settings(name=None):
"""Gets the current settings as dictionary."""
pairs = Setting.all()
if name:
return pairs.filter('name =', name).get().value
return dict((pair.name, pair.value) for pair in pairs if pair.name)
def update_settings(settings):
"""Updates the settings."""
pairs = Setting.all().filter('name IN', settings.keys())
objects = []
for pair in pairs:
pair.value = settings.pop(pair.name)
objects.append(pair)
for key, value in settings.iteritems():
pair = Setting(name=key, value=value)
objects.append(pair)
put(objects)
class Action(dict):
"""The rich dictionary to represent Trello actions."""
@classmethod
def all(cls, since=None):
"""Polls all actions from Trello."""
boards = fetch('https://trello.com/1/members/me/boards?' + cls.sign())
result = []
board_url = 'https://trello.com/1/boards/{0}/actions?{1}{2}'
since_query = 'since=' + since + '&' if since else ''
logger = logging.getLogger(__name__ + '.Action.all')
logger.debug(boards.content)
for board in json.loads(boards.content):
url = board_url.format(board['id'], since_query, cls.sign())
actions = json.loads(fetch(url).content)
result.extend(map(cls, actions))
result.sort(key=lambda d: d['date'])
return result
@staticmethod
def sign():
"""Generates a signing query."""
fmt = 'key={0[trello.app_key]}&token={0[trello.oauth_token]}'
return fmt.format(get_settings())
@property
def id(self):
"""The action id."""
return self['id']
@property
def url(self):
"""The url of the action."""
return 'https://trello.com/1/actions/' + str(self.id)
@property
def data(self):
"""The data dictionary."""
return self['data']
@property
def user(self):
"""The agent."""
return self['memberCreator']['fullName']
@property
def card(self):
"""The card name."""
return self.data.get('card', {}).get('name')
@property
def link_url(self):
if 'card' in self.data:
fmt = 'https://trello.com/card/-/{0.data[board][id]}/' \
'{0.data[card][idShort]}'.format(self)
else:
fmt = 'https://trello.com/board/-/{0.data[board][id]}'
return str(fmt.format(self))
@property
def board(self):
"""The board name."""
return self.data['board']['name']
def is_change(self):
"""Returns whether the action is about change of card."""
return self['type'] in ('changeCard', 'updateCard')
def is_move(self):
"""Returns whether the action is about moving position of card."""
if not self.is_change():
return False
return 'listBefore' in self.data.get('old', {})
def is_create(self):
"""Returns whether the noti is about new card."""
return self['type'] == 'createdCard'
def is_close(self):
"""Returns whether the noti is about closing of card."""
if not self.is_change():
return False
return self.data.get('old', {}).get('closed', False)
def is_comment(self):
"""Returns whether the noti is about new comment."""
return self['type'] == 'commentCard'
def is_attachment(self):
"""Returns whether the action is about new attachment."""
return self['type'] == 'addAttachmentToCard'
def __unicode__(self):
if self.is_close():
msg = u'{0.user} closed card "{0.card}" in "{0.board}".'
elif self.is_move():
msg = u'{0.user} moved card "{0.card}" from ' \
u'"{0.data[listBefore][name]}" to ' \
u'"{0.data[listAfter][name]}" ({0.board} board).'
elif self.is_create():
msg = u'{0.user} created card "{0.card}" into ' \
u'"{0.data[list][name]}" ({0.board} board)'
elif self.is_comment():
msg = u'{0.user} commented to card "{0.card}" ({0.board} board)'
elif self.is_attachment():
msg = u'{0.user} attach a file {{{0.data[attachment][url]}}} ' \
u'to card "{0.card}" ({0.board} board)'
else:
msg = u'{0.user} {0[type]} {0.card} ({0.board} board)'
logger = logging.getLogger(__name__ + '.Action.__unicode__')
logger.warn(repr(self))
return msg.format(self)
def __repr__(self):
r = super(Action, self).__repr__()
return 'Action({0})'.format(r)
#: (:class:`jinja2.Environment`) The Jinja2 environment.
jinja_env = Environment(
loader=FileSystemLoader(
os.path.join(os.path.dirname(__file__), 'templates')
)
)
class BaseHandler(RequestHandler):
"""The common base request handler."""
def render_template(self, template_name, **context):
"""Finds the template and renders it with the given ``context``."""
template = jinja_env.get_template(template_name)
self.response.out.write(template.render(**context))
class SettingPage(BaseHandler):
"""Initial settings page."""
def get(self):
auth_url = urlparse.urljoin(self.request.url, '/trello-oauth')
self.render_template('setting.html',
settings=get_settings(),
auth_url=auth_url,
all_filled=has_settings_complete())
def post(self):
update_settings({
'trello.app_key': self.request.POST['trello.app_key'],
'webhook_url': self.request.POST['webhook_url']
})
return self.get()
class TrelloOAuthPage(BaseHandler):
"""Saves Trello OAuth token."""
def get(self):
self.render_template('trello_oauth.html')
def post(self):
update_settings({
'trello.oauth_token': self.request.POST['token']
})
self.redirect('/')
def poll():
"""Does polling from Trello and posts messages to IRC."""
logger = logging.getLogger(__name__ + '.poll')
settings = get_settings()
latest = settings.get('trello.latest_date')
actions = Action.all(since=latest)
if actions:
latest = max(action['date'] for action in actions)
update_settings({'trello.latest_date': latest})
webhook_url = settings['webhook_url']
for noti in actions:
try:
message = unicode(noti)
link_url = noti.link_url
except Exception as e:
logger.exception(e)
else:
post(message, link_url, webhook_url)
def post(message, link_url, webhook_url):
"""Posts a message to the IRC channel."""
logger = logging.getLogger(__name__ + '.post')
payload = {
'service': 'Trello',
'message': message.encode('utf-8') + ' \xe2\x80\x94 ' + link_url,
'url': link_url,
'icon_url': 'https://trello.com/favicon.ico'
}
logger.info(webhook_url)
logger.info('%r', payload)
response = fetch(
webhook_url,
payload=urllib.urlencode(payload),
method='POST'
)
level = logging.INFO if response.status_code == 200 else logging.WARNING
logger.log(level, '%d: %s', response.status_code, response.content)
class PollPage(BaseHandler):
"""Does polling notifcations from Trello."""
def get(self):
if has_settings_complete():
defer(poll)
else:
self.error(500)
#: (:class:`webapp2.WSGIApplication`) The WSGI callable entrypoint.
app = WSGIApplication([
('/', SettingPage),
('/trello-oauth', TrelloOAuthPage),
('/poll', PollPage)
], debug=True)