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         API_KEY = '71f471f7c6ecdda6def341967686fe05'
  18         SECRET = '7d3248b085f7efbe'
  19
  20         def __init__(self, username, password, token):
  21                 self._username = username
  22                 self._password = password
  23                 self._token = token
  24
  25                 self._rtm = rtmapi.RTMapi(self._username, self.API_KEY, self.SECRET, token)
  26                 self._token = token
  27                 resp = self._rtm.timelines.create()
  28                 self._timeline = resp.timeline
  29                 self._lists = []
  30
  31         def get_projects(self):
  32                 if len(self._lists) == 0:
  33                         self._populate_projects()
  34
  35                 for list in self._lists:
  36                         yield list
  37
  38         def get_project(self, projId):
  39                 proj = [proj for proj in self.get_projects() if projId == proj["id"]]
  40                 assert len(proj) == 1, "%r / %r" % (proj, self._lists)
  41                 return proj[0]
  42
  43         def get_project_names(self):
  44                 return (list["name"] for list in self.get_projects)
  45
  46         def lookup_project(self, projName):
  47                 """
  48                 From a project's name, returns the project's details
  49                 """
  50                 todoList = [list for list in self.get_projects() if list["name"] == projName]
  51                 assert len(todoList) == 1, "Wrong number of lists found for %s, in %r" % (projName, todoList)
  52                 return todoList[0]
  53
  54         def get_locations(self):
  55                 rsp = self._rtm.locations.getList()
  56                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
  57                 locations = [
  58                         dict((
  59                                 ("name", t.name),
  60                                 ("id", t.id),
  61                                 ("longitude", t.longitude),
  62                                 ("latitude", t.latitude),
  63                                 ("address", t.address),
  64                         ))
  65                         for t in rsp.locations
  66                 ]
  67                 return locations
  68
  69         def get_tasks_with_details(self, projId):
  70                 for taskSeries in self._get_taskseries(projId):
  71                         for task in self._get_tasks(taskSeries):
  72                                 taskId = self._pack_ids(projId, taskSeries.id, task.id)
  73                                 priority = task.priority if task.priority != "N" else ""
  74                                 yield {
  75                                         "id": taskId,
  76                                         "projId": projId,
  77                                         "name": taskSeries.name,
  78                                         "url": fix_url(taskSeries.url),
  79                                         "locationId": taskSeries.location_id,
  80                                         "dueDate": task.due,
  81                                         "isCompleted": len(task.completed) != 0,
  82                                         "completedDate": task.completed,
  83                                         "priority": priority,
  84                                         "estimate": task.estimate,
  85                                         "notes": list(self._get_notes(taskSeries.notes)),
  86                                 }
  87
  88         def get_task_details(self, taskId):
  89                 projId, seriesId, taskId = self._unpack_ids(taskId)
  90                 for taskSeries in self._get_taskseries(projId):
  91                         curSeriesId = taskSeries.id
  92                         if seriesId != curSeriesId:
  93                                 continue
  94                         for task in self._get_tasks(taskSeries):
  95                                 curTaskId = task.id
  96                                 if task.id != curTaskId:
  97                                         continue
  98                                 return {
  99                                         "id": taskId,
 100                                         "projId": projId,
 101                                         "name": taskSeries.name,
 102                                         "url": fix_url(taskSeries.url),
 103                                         "locationId": taskSeries.location_id,
 104                                         "due": task.due,
 105                                         "isCompleted": task.completed,
 106                                         "completedDate": task.completed,
 107                                         "priority": task.priority,
 108                                         "estimate": task.estimate,
 109                                         "notes": list(self._get_notes(taskSeries.notes)),
 110                                 }
 111                 return {}
 112
 113         def add_task(self, projId, taskName):
 114                 rsp = self._rtm.tasks.add(
 115                         timeline=self._timeline,
 116                         list_id=projId,
 117                         name=taskName,
 118                 )
 119                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 120                 seriesId = rsp.list.taskseries.id
 121                 taskId = rsp.list.taskseries.task.id
 122                 name = rsp.list.taskseries.name
 123
 124                 return self._pack_ids(projId, seriesId, taskId)
 125
 126         def set_project(self, taskId, newProjId):
 127                 projId, seriesId, taskId = self._unpack_ids(taskId)
 128                 rsp = self._rtm.tasks.moveTo(
 129                         timeline=self._timeline,
 130                         from_list_id=projId,
 131                         to_list_id=newProjId,
 132                         taskseries_id=seriesId,
 133                         task_id=taskId,
 134                 )
 135                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 136
 137         def set_name(self, taskId, name):
 138                 projId, seriesId, taskId = self._unpack_ids(taskId)
 139                 rsp = self._rtm.tasks.setName(
 140                         timeline=self._timeline,
 141                         list_id=projId,
 142                         taskseries_id=seriesId,
 143                         task_id=taskId,
 144                         name=name,
 145                 )
 146                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 147
 148         def set_duedate(self, taskId, dueDate):
 149                 projId, seriesId, taskId = self._unpack_ids(taskId)
 150                 rsp = self._rtm.tasks.setDueDate(
 151                         timeline=self._timeline,
 152                         list_id=projId,
 153                         taskseries_id=seriesId,
 154                         task_id=taskId,
 155                         due=dueDate,
 156                         parse=1,
 157                 )
 158                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 159
 160         def set_priority(self, taskId, priority):
 161                 if priority is None:
 162                         priority = "N"
 163                 else:
 164                         priority = str(priority)
 165                 projId, seriesId, taskId = self._unpack_ids(taskId)
 166
 167                 rsp = self._rtm.tasks.setPriority(
 168                         timeline=self._timeline,
 169                         list_id=projId,
 170                         taskseries_id=seriesId,
 171                         task_id=taskId,
 172                         priority=priority,
 173                 )
 174                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 175
 176         def complete_task(self, taskId):
 177                 projId, seriesId, taskId = self._unpack_ids(taskId)
 178
 179                 rsp = self._rtm.tasks.complete(
 180                         timeline=self._timeline,
 181                         list_id=projId,
 182                         taskseries_id=seriesId,
 183                         task_id=taskId,
 184                 )
 185                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 186
 187         @staticmethod
 188         def _pack_ids(*ids):
 189                 """
 190                 >>> RtMilkManager._pack_ids(123, 456)
 191                 '123-456'
 192                 """
 193                 return "-".join((str(id) for id in ids))
 194
 195         @staticmethod
 196         def _unpack_ids(ids):
 197                 """
 198                 >>> RtMilkManager._unpack_ids("123-456")
 199                 ['123', '456']
 200                 """
 201                 return ids.split("-")
 202
 203         def _get_taskseries(self, projId):
 204                 rsp = self._rtm.tasks.getList(
 205                         list_id=projId,
 206                 )
 207                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 208                 # @note Meta-projects return lists for each project (I think)
 209                 rspTasksList = rsp.tasks.list
 210
 211                 if not isinstance(rspTasksList, list):
 212                         rspTasksList = (rspTasksList, )
 213
 214                 for something in rspTasksList:
 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 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 _get_notes(self, notes):
 237                 if not notes:
 238                         return
 239                 elif isinstance(notes.note, list):
 240                         notes = notes.note
 241                 else:
 242                         notes = (notes.note, )
 243
 244                 for note in notes:
 245                         id = note.id
 246                         title = note.title
 247                         body = getattr(note, "$t")
 248                         yield {
 249                                 "id": id,
 250                                 "title": title,
 251                                 "body": body,
 252                         }
 253
 254         def _populate_projects(self):
 255                 rsp = self._rtm.lists.getList()
 256                 assert rsp.stat == "ok", "Bad response: %r" % (rsp, )
 257                 del self._lists[:]
 258                 self._lists.extend((
 259                         dict((
 260                                 ("name", t.name),
 261                                 ("id", t.id),
 262                                 ("isVisible", not int(t.archived)),
 263                                 ("isMeta", not not int(t.smart)),
 264                         ))
 265                         for t in rsp.lists.list
 266                 ))