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

   1 """
   2 Wrapper for Remember The Milk API
   3 """
   4
   5
   6 import rtmapi
   7
   8
   9 def fix_url(rturl):
  10         return "/".join(rturl.split(r"\/"))
  11
  12
  13 class RtMilkManager(object):
  14         """
  15         Interface with rememberthemilk.com
  16
  17         @todo Decide upon an interface that will end up a bit less bloated
  18         @todo Add interface for task tags
  19         @todo Add interface for postponing tasks (have way for UI to specify how many days to postpone?)
  20         @todo Add interface for task recurrence
  21         @todo Add interface for task estimate
  22         @todo Add interface for task location
  23         @todo Add interface for task url
  24         @todo Add interface for task notes
  25         @todo Add undo support
  26         """
  27         API_KEY = '71f471f7c6ecdda6def341967686fe05'
  28         SECRET = '7d3248b085f7efbe'
  29
  30         def __init__(self, username, password, token):
  31                 self._username = username
  32                 self._password = password
  33                 self._token = token
  34
  35                 self._rtm = rtmapi.RTMapi(self._username, self.API_KEY, self.SECRET, token)
  36                 self._token = token
  37                 resp = self._rtm.timelines.create()
  38                 self._timeline = resp.timeline
  39                 self._lists = []
  40
  41         def get_projects(self):
  42                 if len(self._lists) == 0:
  43                         self._populate_projects()
  44
  45                 for list in self._lists:
  46                         yield list
  47
  48         def get_project(self, projId):
  49                 proj = [proj for proj in self.get_projects() if projId == proj["id"]]
  50                 assert len(proj) == 1, "%r / %r" % (proj, self._lists)
  51                 return proj[0]
  52
  53         def get_project_names(self):
  54                 return (list["name"] for list in self.get_projects)
  55
  56         def lookup_project(self, projName):
  57                 """
  58                 From a project's name, returns the project's details
  59                 """
  60                 todoList = [list for list in self.get_projects() if list["name"] == projName]
  61                 assert len(todoList) == 1, "Wrong number of lists found for %s, in %r" % (projName, todoList)
  62                 return todoList[0]
  63
  64         def get_locations(self):
  65                 rsp = self._rtm.locations.getList()
  66                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
  67                 locations = [
  68                         dict((
  69                                 ("name", t.name),
  70                                 ("id", t.id),
  71                                 ("longitude", t.longitude),
  72                                 ("latitude", t.latitude),
  73                                 ("address", t.address),
  74                         ))
  75                         for t in rsp.locations
  76                 ]
  77                 return locations
  78
  79         def get_tasks_with_details(self, projId):
  80                 for realProjId, taskSeries in self._get_taskseries(projId):
  81                         for task in self._get_tasks(taskSeries):
  82                                 taskId = self._pack_ids(realProjId, taskSeries.id, task.id)
  83                                 priority = task.priority if task.priority != "N" else ""
  84                                 yield {
  85                                         "id": taskId,
  86                                         "projId": projId,
  87                                         "name": taskSeries.name,
  88                                         "url": fix_url(taskSeries.url),
  89                                         "locationId": taskSeries.location_id,
  90                                         "dueDate": task.due,
  91                                         "isCompleted": len(task.completed) != 0,
  92                                         "completedDate": task.completed,
  93                                         "priority": priority,
  94                                         "estimate": task.estimate,
  95                                         "notes": list(self._get_notes(taskSeries.notes)),
  96                                 }
  97
  98         def get_task_details(self, taskId):
  99                 projId, seriesId, taskId = self._unpack_ids(taskId)
 100                 for realProjId, taskSeries in self._get_taskseries(projId):
 101                         assert projId == realProjId, "%s != %s, looks like we let leak a metalist id when packing a task id" % (projId, realProjId)
 102                         curSeriesId = taskSeries.id
 103                         if seriesId != curSeriesId:
 104                                 continue
 105                         for task in self._get_tasks(taskSeries):
 106                                 curTaskId = task.id
 107                                 if task.id != curTaskId:
 108                                         continue
 109                                 return {
 110                                         "id": taskId,
 111                                         "projId": realProjId,
 112                                         "name": taskSeries.name,
 113                                         "url": fix_url(taskSeries.url),
 114                                         "locationId": taskSeries.location_id,
 115                                         "due": task.due,
 116                                         "isCompleted": task.completed,
 117                                         "completedDate": task.completed,
 118                                         "priority": task.priority,
 119                                         "estimate": task.estimate,
 120                                         "notes": list(self._get_notes(taskSeries.notes)),
 121                                 }
 122                 return {}
 123
 124         def add_task(self, projId, taskName):
 125                 rsp = self._rtm.tasks.add(
 126                         timeline=self._timeline,
 127                         list_id=projId,
 128                         name=taskName,
 129                 )
 130                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 131                 seriesId = rsp.list.taskseries.id
 132                 taskId = rsp.list.taskseries.task.id
 133                 name = rsp.list.taskseries.name
 134
 135                 return self._pack_ids(projId, seriesId, taskId)
 136
 137         def set_project(self, taskId, newProjId):
 138                 projId, seriesId, taskId = self._unpack_ids(taskId)
 139                 rsp = self._rtm.tasks.moveTo(
 140                         timeline=self._timeline,
 141                         from_list_id=projId,
 142                         to_list_id=newProjId,
 143                         taskseries_id=seriesId,
 144                         task_id=taskId,
 145                 )
 146                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 147
 148         def set_name(self, taskId, name):
 149                 projId, seriesId, taskId = self._unpack_ids(taskId)
 150                 rsp = self._rtm.tasks.setName(
 151                         timeline=self._timeline,
 152                         list_id=projId,
 153                         taskseries_id=seriesId,
 154                         task_id=taskId,
 155                         name=name,
 156                 )
 157                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 158
 159         def set_duedate(self, taskId, dueDate):
 160                 projId, seriesId, taskId = self._unpack_ids(taskId)
 161                 rsp = self._rtm.tasks.setDueDate(
 162                         timeline=self._timeline,
 163                         list_id=projId,
 164                         taskseries_id=seriesId,
 165                         task_id=taskId,
 166                         due=dueDate,
 167                         parse=1,
 168                 )
 169                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 170
 171         def set_priority(self, taskId, priority):
 172                 if priority is None:
 173                         priority = "N"
 174                 else:
 175                         priority = str(priority)
 176                 projId, seriesId, taskId = self._unpack_ids(taskId)
 177
 178                 rsp = self._rtm.tasks.setPriority(
 179                         timeline=self._timeline,
 180                         list_id=projId,
 181                         taskseries_id=seriesId,
 182                         task_id=taskId,
 183                         priority=priority,
 184                 )
 185                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 186
 187         def complete_task(self, taskId):
 188                 projId, seriesId, taskId = self._unpack_ids(taskId)
 189
 190                 rsp = self._rtm.tasks.complete(
 191                         timeline=self._timeline,
 192                         list_id=projId,
 193                         taskseries_id=seriesId,
 194                         task_id=taskId,
 195                 )
 196                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 197
 198         @staticmethod
 199         def _pack_ids(*ids):
 200                 """
 201                 >>> RtMilkManager._pack_ids(123, 456)
 202                 '123-456'
 203                 """
 204                 return "-".join((str(id) for id in ids))
 205
 206         @staticmethod
 207         def _unpack_ids(ids):
 208                 """
 209                 >>> RtMilkManager._unpack_ids("123-456")
 210                 ['123', '456']
 211                 """
 212                 return ids.split("-")
 213
 214         def _get_taskseries(self, projId):
 215                 rsp = self._rtm.tasks.getList(
 216                         list_id=projId,
 217                 )
 218                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 219                 # @note Meta-projects return lists for each project (I think)
 220                 rspTasksList = rsp.tasks.list
 221
 222                 if not isinstance(rspTasksList, list):
 223                         rspTasksList = (rspTasksList, )
 224
 225                 for something in rspTasksList:
 226                         realProjId = something.id
 227                         try:
 228                                 something.taskseries
 229                         except AttributeError:
 230                                 continue
 231
 232                         if isinstance(something.taskseries, list):
 233                                 somethingsTaskseries = something.taskseries
 234                         else:
 235                                 somethingsTaskseries = (something.taskseries, )
 236
 237                         for taskSeries in somethingsTaskseries:
 238                                 yield realProjId, taskSeries
 239
 240         def _get_tasks(self, taskSeries):
 241                 if isinstance(taskSeries.task, list):
 242                         tasks = taskSeries.task
 243                 else:
 244                         tasks = (taskSeries.task, )
 245                 for task in tasks:
 246                         yield task
 247
 248         def _get_notes(self, notes):
 249                 if not notes:
 250                         return
 251                 elif isinstance(notes.note, list):
 252                         notes = notes.note
 253                 else:
 254                         notes = (notes.note, )
 255
 256                 for note in notes:
 257                         id = note.id
 258                         title = note.title
 259                         body = getattr(note, "$t")
 260                         yield {
 261                                 "id": id,
 262                                 "title": title,
 263                                 "body": body,
 264                         }
 265
 266         def _populate_projects(self):
 267                 rsp = self._rtm.lists.getList()
 268                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 269                 del self._lists[:]
 270                 self._lists.extend((
 271                         dict((
 272                                 ("name", t.name),
 273                                 ("id", t.id),
 274                                 ("isVisible", not int(t.archived)),
 275                                 ("isMeta", not not int(t.smart)),
 276                         ))
 277                         for t in rsp.lists.list
 278                 ))