git.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 interface for task notes
  27         @todo Add undo support
  28         """
  29         API_KEY = '71f471f7c6ecdda6def341967686fe05'
  30         SECRET = '7d3248b085f7efbe'
  31
  32         def __init__(self, username, password, token):
  33                 self._username = username
  34                 self._password = password
  35                 self._token = token
  36
  37                 self._rtm = rtm_api.RTMapi(self._username, self.API_KEY, self.SECRET, token)
  38                 self._token = token
  39                 resp = self._rtm.timelines.create()
  40                 self._timeline = resp.timeline
  41                 self._lists = []
  42
  43         def get_projects(self):
  44                 if len(self._lists) == 0:
  45                         self._populate_projects()
  46
  47                 for list in self._lists:
  48                         yield list
  49
  50         def get_project(self, projId):
  51                 projs = [proj for proj in self.get_projects() if projId == proj["id"]]
  52                 assert len(projs) == 1, "%r: %r / %r" % (projId, projs, self._lists)
  53                 return projs[0]
  54
  55         def get_project_names(self):
  56                 return (list["name"] for list in self.get_projects)
  57
  58         def lookup_project(self, projName):
  59                 """
  60                 From a project's name, returns the project's details
  61                 """
  62                 todoList = [list for list in self.get_projects() if list["name"] == projName]
  63                 assert len(todoList) == 1, "Wrong number of lists found for %s, in %r" % (projName, todoList)
  64                 return todoList[0]
  65
  66         def get_locations(self):
  67                 rsp = self._rtm.locations.getList()
  68                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
  69                 locations = [
  70                         dict((
  71                                 ("name", t.name),
  72                                 ("id", t.id),
  73                                 ("longitude", t.longitude),
  74                                 ("latitude", t.latitude),
  75                                 ("address", t.address),
  76                         ))
  77                         for t in rsp.locations
  78                 ]
  79                 return locations
  80
  81         def get_tasks_with_details(self, projId):
  82                 for realProjId, taskSeries in self._get_taskseries(projId):
  83                         for task in self._get_tasks(taskSeries):
  84                                 taskId = self._pack_ids(realProjId, taskSeries.id, task.id)
  85                                 rawTaskDetails = {
  86                                         "id": taskId,
  87                                         "projId": projId,
  88                                         "name": taskSeries.name,
  89                                         "url": taskSeries.url,
  90                                         "locationId": taskSeries.location_id,
  91                                         "dueDate": task.due,
  92                                         "isCompleted": task.completed,
  93                                         "completedDate": task.completed,
  94                                         "priority": task.priority,
  95                                         "estimate": task.estimate,
  96                                         "notes": list(self._get_notes(taskId, taskSeries.notes)),
  97                                 }
  98                                 taskDetails = self._parse_task_details(rawTaskDetails)
  99                                 yield taskDetails
 100
 101         def get_task_details(self, taskId):
 102                 projId, rtmSeriesId, rtmTaskId = self._unpack_ids(taskId)
 103                 for task in self.get_tasks_with_details(projId):
 104                         if task["id"] == taskId:
 105                                 return task
 106                 return {}
 107
 108         def add_task(self, projId, taskName):
 109                 rsp = self._rtm.tasks.add(
 110                         timeline=self._timeline,
 111                         list_id=projId,
 112                         name=taskName,
 113                 )
 114                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 115                 seriesId = rsp.list.taskseries.id
 116                 taskId = rsp.list.taskseries.task.id
 117                 name = rsp.list.taskseries.name
 118
 119                 return self._pack_ids(projId, seriesId, taskId)
 120
 121         def set_project(self, taskId, newProjId):
 122                 projId, seriesId, taskId = self._unpack_ids(taskId)
 123                 rsp = self._rtm.tasks.moveTo(
 124                         timeline=self._timeline,
 125                         from_list_id=projId,
 126                         to_list_id=newProjId,
 127                         taskseries_id=seriesId,
 128                         task_id=taskId,
 129                 )
 130                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 131
 132         def set_name(self, taskId, name):
 133                 projId, seriesId, taskId = self._unpack_ids(taskId)
 134                 rsp = self._rtm.tasks.setName(
 135                         timeline=self._timeline,
 136                         list_id=projId,
 137                         taskseries_id=seriesId,
 138                         task_id=taskId,
 139                         name=name,
 140                 )
 141                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 142
 143         def set_duedate(self, taskId, dueDate):
 144                 assert isinstance(dueDate, toolbox.Optional), (
 145                         "Date being set too definitively: %r" % dueDate
 146                 )
 147
 148                 projId, seriesId, taskId = self._unpack_ids(taskId)
 149                 rsp = self._rtm.tasks.setDueDate(
 150                         timeline=self._timeline,
 151                         list_id=projId,
 152                         taskseries_id=seriesId,
 153                         task_id=taskId,
 154                         due=dueDate,
 155                         parse=1,
 156                 )
 157                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 158
 159         def set_priority(self, taskId, priority):
 160                 assert isinstance(priority, toolbox.Optional), (
 161                         "Priority being set too definitively: %r" % priority
 162                 )
 163                 priority = str(priority.get_nothrow("N"))
 164                 projId, seriesId, taskId = self._unpack_ids(taskId)
 165
 166                 rsp = self._rtm.tasks.setPriority(
 167                         timeline=self._timeline,
 168                         list_id=projId,
 169                         taskseries_id=seriesId,
 170                         task_id=taskId,
 171                         priority=priority,
 172                 )
 173                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 174
 175         def complete_task(self, taskId):
 176                 projId, seriesId, taskId = self._unpack_ids(taskId)
 177
 178                 rsp = self._rtm.tasks.complete(
 179                         timeline=self._timeline,
 180                         list_id=projId,
 181                         taskseries_id=seriesId,
 182                         task_id=taskId,
 183                 )
 184                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 185
 186         @staticmethod
 187         def _pack_ids(*ids):
 188                 """
 189                 >>> RtMilkManager._pack_ids(123, 456)
 190                 '123-456'
 191                 """
 192                 return "-".join((str(id) for id in ids))
 193
 194         @staticmethod
 195         def _unpack_ids(ids):
 196                 """
 197                 >>> RtMilkManager._unpack_ids("123-456")
 198                 ['123', '456']
 199                 """
 200                 return ids.split("-")
 201
 202         def _get_taskseries(self, projId):
 203                 rsp = self._rtm.tasks.getList(
 204                         list_id=projId,
 205                 )
 206                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 207                 # @note Meta-projects return lists for each project (I think)
 208                 rspTasksList = rsp.tasks.list
 209
 210                 if not isinstance(rspTasksList, list):
 211                         rspTasksList = (rspTasksList, )
 212
 213                 for something in rspTasksList:
 214                         realProjId = something.id
 215                         try:
 216                                 something.taskseries
 217                         except AttributeError:
 218                                 continue
 219
 220                         if isinstance(something.taskseries, list):
 221                                 somethingsTaskseries = something.taskseries
 222                         else:
 223                                 somethingsTaskseries = (something.taskseries, )
 224
 225                         for taskSeries in somethingsTaskseries:
 226                                 yield realProjId, taskSeries
 227
 228         def _get_tasks(self, taskSeries):
 229                 if isinstance(taskSeries.task, list):
 230                         tasks = taskSeries.task
 231                 else:
 232                         tasks = (taskSeries.task, )
 233                 for task in tasks:
 234                         yield task
 235
 236         def _parse_task_details(self, rawTaskDetails):
 237                 taskDetails = {}
 238                 taskDetails["id"] = rawTaskDetails["id"]
 239                 taskDetails["projId"] = rawTaskDetails["projId"]
 240                 taskDetails["name"] = rawTaskDetails["name"]
 241                 taskDetails["url"] = fix_url(rawTaskDetails["url"])
 242
 243                 rawLocationId = rawTaskDetails["locationId"]
 244                 if rawLocationId:
 245                         locationId = toolbox.Optional(rawLocationId)
 246                 else:
 247                         locationId = toolbox.Optional()
 248                 taskDetails["locationId"] = locationId
 249
 250                 if rawTaskDetails["dueDate"]:
 251                         dueDate = datetime.datetime.strptime(
 252                                 rawTaskDetails["dueDate"],
 253                                 "%Y-%m-%dT%H:%M:%SZ",
 254                         )
 255                         dueDate = toolbox.Optional(dueDate)
 256                 else:
 257                         dueDate = toolbox.Optional()
 258                 taskDetails["dueDate"] = dueDate
 259
 260                 taskDetails["isCompleted"] = len(rawTaskDetails["isCompleted"]) != 0
 261
 262                 if rawTaskDetails["completedDate"]:
 263                         completedDate = datetime.datetime.strptime(
 264                                 rawTaskDetails["completedDate"],
 265                                 "%Y-%m-%dT%H:%M:%SZ",
 266                         )
 267                         completedDate = toolbox.Optional(completedDate)
 268                 else:
 269                         completedDate = toolbox.Optional()
 270                 taskDetails["completedDate"] = completedDate
 271
 272                 try:
 273                         priority = toolbox.Optional(int(rawTaskDetails["priority"]))
 274                 except ValueError:
 275                         priority = toolbox.Optional()
 276                 taskDetails["priority"] = priority
 277
 278                 if rawTaskDetails["estimate"]:
 279                         estimate = rawTaskDetails["estimate"]
 280                         estimate = toolbox.Optional(estimate)
 281                 else:
 282                         estimate = toolbox.Optional()
 283                 taskDetails["estimate"] = estimate
 284
 285                 taskDetails["notes"] = rawTaskDetails["notes"]
 286
 287                 rawKeys = list(rawTaskDetails.iterkeys())
 288                 rawKeys.sort()
 289                 parsedKeys = list(taskDetails.iterkeys())
 290                 parsedKeys.sort()
 291                 assert rawKeys == parsedKeys, "Missing some, %r != %r" % (rawKeys, parsedKeys)
 292
 293                 return taskDetails
 294
 295         def _get_notes(self, taskId, notes):
 296                 if not notes:
 297                         return
 298                 elif isinstance(notes.note, list):
 299                         notes = notes.note
 300                 else:
 301                         notes = (notes.note, )
 302
 303                 projId, rtmSeriesId, rtmTaskId = self._unpack_ids(taskId)
 304
 305                 for note in notes:
 306                         noteId = self._pack_ids(projId, rtmSeriesId, rtmTaskId, note.id)
 307                         title = note.title
 308                         body = getattr(note, "$t")
 309                         yield {
 310                                 "id": noteId,
 311                                 "title": title,
 312                                 "body": body,
 313                         }
 314
 315         def _populate_projects(self):
 316                 rsp = self._rtm.lists.getList()
 317                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 318                 del self._lists[:]
 319                 self._lists.extend((
 320                         dict((
 321                                 ("name", t.name),
 322                                 ("id", t.id),
 323                                 ("isVisible", not int(t.archived)),
 324                                 ("isMeta", not not int(t.smart)),
 325                         ))
 326                         for t in rsp.lists.list
 327                 ))