vcs.maemo.org Git - doneit/blob - src/rtm_backend.py

   1 """
   2 Wrapper for Remember The Milk API
   3 """
   4
   5 import datetime
   6
   7 import toolbox
   8 import rtm_api
   9
  10
  11 def fix_url(rturl):
  12         return "/".join(rturl.split(r"\/"))
  13
  14
  15 class RtMilkManager(object):
  16         """
  17         Interface with rememberthemilk.com
  18
  19         @todo Decide upon an interface that will end up a bit less bloated
  20         @todo Add interface for task tags
  21         @todo Add interface for postponing tasks (have way for UI to specify how many days to postpone?)
  22         @todo Add interface for task recurrence
  23         @todo Add interface for task estimate
  24         @todo Add interface for task location
  25         @todo Add interface for task url
  26         @todo Add undo support
  27         """
  28         API_KEY = '71f471f7c6ecdda6def341967686fe05'
  29         SECRET = '7d3248b085f7efbe'
  30
  31         def __init__(self, username, password, token):
  32                 self._username = username
  33                 self._password = password
  34                 self._token = token
  35
  36                 self._rtm = rtm_api.RTMapi(self._username, self.API_KEY, self.SECRET, token)
  37                 self._token = token
  38                 resp = self._rtm.timelines.create()
  39                 self._timeline = resp.timeline
  40                 self._lists = []
  41
  42         def add_project(self, name):
  43                 rsp = self._rtm.lists.add(
  44                         timeline=self._timeline,
  45                         name=name,
  46                 )
  47                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
  48                 self._lists = []
  49
  50         def set_project_name(self, projId, name):
  51                 rsp = self._rtm.lists.setName(
  52                         timeline=self._timeline,
  53                         list_id=projId,
  54                         name=name,
  55                 )
  56                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
  57                 self._lists = []
  58
  59         def set_project_visibility(self, projId, visibility):
  60                 action = self._rtm.lists.unarchive if visibility else self._rtm.lists.archive
  61                 rsp = action(
  62                         timeline=self._timeline,
  63                         list_id=projId,
  64                 )
  65                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
  66                 self._lists = []
  67
  68         def get_projects(self):
  69                 if len(self._lists) == 0:
  70                         self._populate_projects()
  71
  72                 for list in self._lists:
  73                         yield list
  74
  75         def get_project(self, projId):
  76                 projs = [proj for proj in self.get_projects() if projId == proj["id"]]
  77                 assert len(projs) == 1, "%r: %r / %r" % (projId, projs, self._lists)
  78                 return projs[0]
  79
  80         def lookup_project(self, projName):
  81                 """
  82                 From a project's name, returns the project's details
  83                 """
  84                 todoList = [list for list in self.get_projects() if list["name"] == projName]
  85                 assert len(todoList) == 1, "Wrong number of lists found for %s, in %r" % (projName, todoList)
  86                 return todoList[0]
  87
  88         def get_locations(self):
  89                 rsp = self._rtm.locations.getList()
  90                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
  91                 locations = [
  92                         dict((
  93                                 ("name", t.name),
  94                                 ("id", t.id),
  95                                 ("longitude", t.longitude),
  96                                 ("latitude", t.latitude),
  97                                 ("address", t.address),
  98                         ))
  99                         for t in rsp.locations
 100                 ]
 101                 return locations
 102
 103         def get_tasks_with_details(self, projId):
 104                 for realProjId, taskSeries in self._get_taskseries(projId):
 105                         for task in self._get_tasks(taskSeries):
 106                                 taskId = self._pack_ids(realProjId, taskSeries.id, task.id)
 107                                 rawTaskDetails = {
 108                                         "id": taskId,
 109                                         "projId": projId,
 110                                         "name": taskSeries.name,
 111                                         "url": taskSeries.url,
 112                                         "locationId": taskSeries.location_id,
 113                                         "dueDate": task.due,
 114                                         "isCompleted": task.completed,
 115                                         "completedDate": task.completed,
 116                                         "priority": task.priority,
 117                                         "estimate": task.estimate,
 118                                         "notes": dict((
 119                                                 (note["id"], note)
 120                                                 for note in self._get_notes(taskId, taskSeries.notes)
 121                                         )),
 122                                 }
 123                                 taskDetails = self._parse_task_details(rawTaskDetails)
 124                                 yield taskDetails
 125
 126         def get_task_details(self, taskId):
 127                 projId, rtmSeriesId, rtmTaskId = self._unpack_ids(taskId)
 128                 for task in self.get_tasks_with_details(projId):
 129                         if task["id"] == taskId:
 130                                 return task
 131                 return {}
 132
 133         def add_task(self, projId, taskName):
 134                 rsp = self._rtm.tasks.add(
 135                         timeline=self._timeline,
 136                         list_id=projId,
 137                         name=taskName,
 138                 )
 139                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 140                 seriesId = rsp.list.taskseries.id
 141                 taskId = rsp.list.taskseries.task.id
 142                 name = rsp.list.taskseries.name
 143
 144                 return self._pack_ids(projId, seriesId, taskId)
 145
 146         def set_project(self, taskId, newProjId):
 147                 projId, seriesId, taskId = self._unpack_ids(taskId)
 148                 rsp = self._rtm.tasks.moveTo(
 149                         timeline=self._timeline,
 150                         from_list_id=projId,
 151                         to_list_id=newProjId,
 152                         taskseries_id=seriesId,
 153                         task_id=taskId,
 154                 )
 155                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 156
 157         def set_name(self, taskId, name):
 158                 projId, seriesId, taskId = self._unpack_ids(taskId)
 159                 rsp = self._rtm.tasks.setName(
 160                         timeline=self._timeline,
 161                         list_id=projId,
 162                         taskseries_id=seriesId,
 163                         task_id=taskId,
 164                         name=name,
 165                 )
 166                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 167
 168         def set_duedate(self, taskId, dueDate):
 169                 assert isinstance(dueDate, toolbox.Optional), (
 170                         "Date being set too definitively: %r" % dueDate
 171                 )
 172
 173                 projId, seriesId, taskId = self._unpack_ids(taskId)
 174                 rsp = self._rtm.tasks.setDueDate(
 175                         timeline=self._timeline,
 176                         list_id=projId,
 177                         taskseries_id=seriesId,
 178                         task_id=taskId,
 179                         due=dueDate,
 180                         parse=1,
 181                 )
 182                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 183
 184         def set_priority(self, taskId, priority):
 185                 assert isinstance(priority, toolbox.Optional), (
 186                         "Priority being set too definitively: %r" % priority
 187                 )
 188                 priority = str(priority.get_nothrow("N"))
 189                 projId, seriesId, taskId = self._unpack_ids(taskId)
 190
 191                 rsp = self._rtm.tasks.setPriority(
 192                         timeline=self._timeline,
 193                         list_id=projId,
 194                         taskseries_id=seriesId,
 195                         task_id=taskId,
 196                         priority=priority,
 197                 )
 198                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 199
 200         def complete_task(self, taskId):
 201                 projId, seriesId, taskId = self._unpack_ids(taskId)
 202
 203                 rsp = self._rtm.tasks.complete(
 204                         timeline=self._timeline,
 205                         list_id=projId,
 206                         taskseries_id=seriesId,
 207                         task_id=taskId,
 208                 )
 209                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 210
 211         def add_note(self, taskId, noteTitle, noteBody):
 212                 projId, seriesId, taskId = self._unpack_ids(taskId)
 213
 214                 rsp = self._rtm.tasks.notes.add(
 215                         timeline=self._timeline,
 216                         list_id=projId,
 217                         taskseries_id=seriesId,
 218                         task_id=taskId,
 219                         note_title=noteTitle,
 220                         note_text=noteBody,
 221                 )
 222                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 223
 224         def update_note(self, noteId, noteTitle, noteBody):
 225                 projId, seriesId, taskId, note = self._unpack_ids(noteId)
 226
 227                 rsp = self._rtm.tasks.notes.edit(
 228                         timeline=self._timeline,
 229                         note_id=noteId,
 230                         note_title=noteTitle,
 231                         note_text=noteBody,
 232                 )
 233                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 234
 235         def delete_note(self, noteId):
 236                 projId, seriesId, taskId, noteId = self._unpack_ids(noteId)
 237
 238                 rsp = self._rtm.tasks.notes.delete(
 239                         timeline=self._timeline,
 240                         note_id=noteId,
 241                 )
 242                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 243
 244         @staticmethod
 245         def _pack_ids(*ids):
 246                 """
 247                 >>> RtMilkManager._pack_ids(123, 456)
 248                 '123-456'
 249                 """
 250                 return "-".join((str(id) for id in ids))
 251
 252         @staticmethod
 253         def _unpack_ids(ids):
 254                 """
 255                 >>> RtMilkManager._unpack_ids("123-456")
 256                 ['123', '456']
 257                 """
 258                 return ids.split("-")
 259
 260         def _get_taskseries(self, projId):
 261                 rsp = self._rtm.tasks.getList(
 262                         list_id=projId,
 263                 )
 264                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 265                 # @note Meta-projects return lists for each project (I think)
 266                 rspTasksList = rsp.tasks.list
 267
 268                 if not isinstance(rspTasksList, list):
 269                         rspTasksList = (rspTasksList, )
 270
 271                 for something in rspTasksList:
 272                         realProjId = something.id
 273                         try:
 274                                 something.taskseries
 275                         except AttributeError:
 276                                 continue
 277
 278                         if isinstance(something.taskseries, list):
 279                                 somethingsTaskseries = something.taskseries
 280                         else:
 281                                 somethingsTaskseries = (something.taskseries, )
 282
 283                         for taskSeries in somethingsTaskseries:
 284                                 yield realProjId, taskSeries
 285
 286         def _get_tasks(self, taskSeries):
 287                 if isinstance(taskSeries.task, list):
 288                         tasks = taskSeries.task
 289                 else:
 290                         tasks = (taskSeries.task, )
 291                 for task in tasks:
 292                         yield task
 293
 294         def _parse_task_details(self, rawTaskDetails):
 295                 taskDetails = {}
 296                 taskDetails["id"] = rawTaskDetails["id"]
 297                 taskDetails["projId"] = rawTaskDetails["projId"]
 298                 taskDetails["name"] = rawTaskDetails["name"]
 299                 taskDetails["url"] = fix_url(rawTaskDetails["url"])
 300
 301                 rawLocationId = rawTaskDetails["locationId"]
 302                 if rawLocationId:
 303                         locationId = toolbox.Optional(rawLocationId)
 304                 else:
 305                         locationId = toolbox.Optional()
 306                 taskDetails["locationId"] = locationId
 307
 308                 if rawTaskDetails["dueDate"]:
 309                         dueDate = datetime.datetime.strptime(
 310                                 rawTaskDetails["dueDate"],
 311                                 "%Y-%m-%dT%H:%M:%SZ",
 312                         )
 313                         dueDate = toolbox.Optional(dueDate)
 314                 else:
 315                         dueDate = toolbox.Optional()
 316                 taskDetails["dueDate"] = dueDate
 317
 318                 taskDetails["isCompleted"] = len(rawTaskDetails["isCompleted"]) != 0
 319
 320                 if rawTaskDetails["completedDate"]:
 321                         completedDate = datetime.datetime.strptime(
 322                                 rawTaskDetails["completedDate"],
 323                                 "%Y-%m-%dT%H:%M:%SZ",
 324                         )
 325                         completedDate = toolbox.Optional(completedDate)
 326                 else:
 327                         completedDate = toolbox.Optional()
 328                 taskDetails["completedDate"] = completedDate
 329
 330                 try:
 331                         priority = toolbox.Optional(int(rawTaskDetails["priority"]))
 332                 except ValueError:
 333                         priority = toolbox.Optional()
 334                 taskDetails["priority"] = priority
 335
 336                 if rawTaskDetails["estimate"]:
 337                         estimate = rawTaskDetails["estimate"]
 338                         estimate = toolbox.Optional(estimate)
 339                 else:
 340                         estimate = toolbox.Optional()
 341                 taskDetails["estimate"] = estimate
 342
 343                 taskDetails["notes"] = rawTaskDetails["notes"]
 344
 345                 rawKeys = list(rawTaskDetails.iterkeys())
 346                 rawKeys.sort()
 347                 parsedKeys = list(taskDetails.iterkeys())
 348                 parsedKeys.sort()
 349                 assert rawKeys == parsedKeys, "Missing some, %r != %r" % (rawKeys, parsedKeys)
 350
 351                 return taskDetails
 352
 353         def _get_notes(self, taskId, notes):
 354                 if not notes:
 355                         return
 356                 elif isinstance(notes.note, list):
 357                         notes = notes.note
 358                 else:
 359                         notes = (notes.note, )
 360
 361                 projId, rtmSeriesId, rtmTaskId = self._unpack_ids(taskId)
 362
 363                 for note in notes:
 364                         noteId = self._pack_ids(projId, rtmSeriesId, rtmTaskId, note.id)
 365                         title = note.title
 366                         body = getattr(note, "$t")
 367                         yield {
 368                                 "id": noteId,
 369                                 "title": title,
 370                                 "body": body,
 371                         }
 372
 373         def _populate_projects(self):
 374                 rsp = self._rtm.lists.getList()
 375                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 376                 del self._lists[:]
 377                 self._lists.extend((
 378                         dict((
 379                                 ("name", t.name),
 380                                 ("id", t.id),
 381                                 ("isVisible", not int(t.archived)),
 382                                 ("isMeta", not not int(t.smart)),
 383                         ))
 384                         for t in rsp.lists.list
 385                 ))