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 taskSeries in self._get_taskseries(projId):
  81                         for task in self._get_tasks(taskSeries):
  82                                 taskId = self._pack_ids(projId, 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 taskSeries in self._get_taskseries(projId):
 101                         curSeriesId = taskSeries.id
 102                         if seriesId != curSeriesId:
 103                                 continue
 104                         for task in self._get_tasks(taskSeries):
 105                                 curTaskId = task.id
 106                                 if task.id != curTaskId:
 107                                         continue
 108                                 return {
 109                                         "id": taskId,
 110                                         "projId": projId,
 111                                         "name": taskSeries.name,
 112                                         "url": fix_url(taskSeries.url),
 113                                         "locationId": taskSeries.location_id,
 114                                         "due": task.due,
 115                                         "isCompleted": task.completed,
 116                                         "completedDate": task.completed,
 117                                         "priority": task.priority,
 118                                         "estimate": task.estimate,
 119                                         "notes": list(self._get_notes(taskSeries.notes)),
 120                                 }
 121                 return {}
 122
 123         def add_task(self, projId, taskName):
 124                 rsp = self._rtm.tasks.add(
 125                         timeline=self._timeline,
 126                         list_id=projId,
 127                         name=taskName,
 128                 )
 129                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 130                 seriesId = rsp.list.taskseries.id
 131                 taskId = rsp.list.taskseries.task.id
 132                 name = rsp.list.taskseries.name
 133
 134                 return self._pack_ids(projId, seriesId, taskId)
 135
 136         def set_project(self, taskId, newProjId):
 137                 projId, seriesId, taskId = self._unpack_ids(taskId)
 138                 rsp = self._rtm.tasks.moveTo(
 139                         timeline=self._timeline,
 140                         from_list_id=projId,
 141                         to_list_id=newProjId,
 142                         taskseries_id=seriesId,
 143                         task_id=taskId,
 144                 )
 145                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 146
 147         def set_name(self, taskId, name):
 148                 projId, seriesId, taskId = self._unpack_ids(taskId)
 149                 rsp = self._rtm.tasks.setName(
 150                         timeline=self._timeline,
 151                         list_id=projId,
 152                         taskseries_id=seriesId,
 153                         task_id=taskId,
 154                         name=name,
 155                 )
 156                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 157
 158         def set_duedate(self, taskId, dueDate):
 159                 projId, seriesId, taskId = self._unpack_ids(taskId)
 160                 rsp = self._rtm.tasks.setDueDate(
 161                         timeline=self._timeline,
 162                         list_id=projId,
 163                         taskseries_id=seriesId,
 164                         task_id=taskId,
 165                         due=dueDate,
 166                         parse=1,
 167                 )
 168                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 169
 170         def set_priority(self, taskId, priority):
 171                 if priority is None:
 172                         priority = "N"
 173                 else:
 174                         priority = str(priority)
 175                 projId, seriesId, taskId = self._unpack_ids(taskId)
 176
 177                 rsp = self._rtm.tasks.setPriority(
 178                         timeline=self._timeline,
 179                         list_id=projId,
 180                         taskseries_id=seriesId,
 181                         task_id=taskId,
 182                         priority=priority,
 183                 )
 184                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 185
 186         def complete_task(self, taskId):
 187                 projId, seriesId, taskId = self._unpack_ids(taskId)
 188
 189                 rsp = self._rtm.tasks.complete(
 190                         timeline=self._timeline,
 191                         list_id=projId,
 192                         taskseries_id=seriesId,
 193                         task_id=taskId,
 194                 )
 195                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 196
 197         @staticmethod
 198         def _pack_ids(*ids):
 199                 """
 200                 >>> RtMilkManager._pack_ids(123, 456)
 201                 '123-456'
 202                 """
 203                 return "-".join((str(id) for id in ids))
 204
 205         @staticmethod
 206         def _unpack_ids(ids):
 207                 """
 208                 >>> RtMilkManager._unpack_ids("123-456")
 209                 ['123', '456']
 210                 """
 211                 return ids.split("-")
 212
 213         def _get_taskseries(self, projId):
 214                 rsp = self._rtm.tasks.getList(
 215                         list_id=projId,
 216                 )
 217                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 218                 # @note Meta-projects return lists for each project (I think)
 219                 rspTasksList = rsp.tasks.list
 220
 221                 if not isinstance(rspTasksList, list):
 222                         rspTasksList = (rspTasksList, )
 223
 224                 for something in rspTasksList:
 225                         try:
 226                                 something.taskseries
 227                         except AttributeError:
 228                                 continue
 229
 230                         if isinstance(something.taskseries, list):
 231                                 somethingsTaskseries = something.taskseries
 232                         else:
 233                                 somethingsTaskseries = (something.taskseries, )
 234
 235                         for taskSeries in somethingsTaskseries:
 236                                 yield taskSeries
 237
 238         def _get_tasks(self, taskSeries):
 239                 if isinstance(taskSeries.task, list):
 240                         tasks = taskSeries.task
 241                 else:
 242                         tasks = (taskSeries.task, )
 243                 for task in tasks:
 244                         yield task
 245
 246         def _get_notes(self, notes):
 247                 if not notes:
 248                         return
 249                 elif isinstance(notes.note, list):
 250                         notes = notes.note
 251                 else:
 252                         notes = (notes.note, )
 253
 254                 for note in notes:
 255                         id = note.id
 256                         title = note.title
 257                         body = getattr(note, "$t")
 258                         yield {
 259                                 "id": id,
 260                                 "title": title,
 261                                 "body": body,
 262                         }
 263
 264         def _populate_projects(self):
 265                 rsp = self._rtm.lists.getList()
 266                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 267                 del self._lists[:]
 268                 self._lists.extend((
 269                         dict((
 270                                 ("name", t.name),
 271                                 ("id", t.id),
 272                                 ("isVisible", not int(t.archived)),
 273                                 ("isMeta", not not int(t.smart)),
 274                         ))
 275                         for t in rsp.lists.list
 276                 ))