src/gpodder/my.py

   1 #!/usr/bin/python
   2 # -*- coding: utf-8 -*-
   3 #
   4 # gPodder - A media aggregator and podcast client
   5 # Copyright (c) 2005-2010 Thomas Perl and the gPodder Team
   6 #
   7 # gPodder is free software; you can redistribute it and/or modify
   8 # it under the terms of the GNU General Public License as published by
   9 # the Free Software Foundation; either version 3 of the License, or
  10 # (at your option) any later version.
  11 #
  12 # gPodder is distributed in the hope that it will be useful,
  13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 # GNU General Public License for more details.
  16 #
  17 # You should have received a copy of the GNU General Public License
  18 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19 #
  20
  21
  22 #
  23 #  my.py -- mygpo Client Abstraction for gPodder
  24 #  Thomas Perl <thp@gpodder.org>; 2010-01-19
  25 #
  26
  27 import gpodder
  28 _ = gpodder.gettext
  29
  30 import atexit
  31 import os
  32 import threading
  33 import time
  34
  35 from gpodder.liblogger import log
  36
  37 from gpodder import util
  38
  39 # Append gPodder's user agent to mygpoclient's user agent
  40 import mygpoclient
  41 mygpoclient.user_agent += ' ' + gpodder.user_agent
  42
  43 from mygpoclient import api
  44
  45 try:
  46     import simplejson as json
  47 except ImportError:
  48     import json
  49
  50
  51 class Change(object):
  52     ADD, REMOVE = range(2)
  53
  54     def __init__(self, url, change, podcast=None):
  55         self.url = url
  56         self.change = change
  57         self.podcast = podcast
  58
  59     @property
  60     def description(self):
  61         if self.change == self.ADD:
  62             return _('Add %s') % self.url
  63         else:
  64             return _('Remove %s') % self.podcast.title
  65
  66
  67 class Actions(object):
  68     NONE = 0
  69
  70     SYNC_PODCASTS, \
  71     UPLOAD_EPISODES, \
  72     UPDATE_DEVICE = (1<<x for x in range(3))
  73
  74 class MygPoClient(object):
  75     CACHE_FILE = 'mygpo.queue.json'
  76     FLUSH_TIMEOUT = 60
  77     FLUSH_RETRIES = 3
  78
  79     def __init__(self, config,
  80             on_rewrite_url=lambda old_url, new_url: None,
  81             on_add_remove_podcasts=lambda add_urls, remove_urls: None,
  82             on_send_full_subscriptions=lambda: None):
  83         self._cache = {'actions': Actions.NONE,
  84                        'add_podcasts': [],
  85                        'remove_podcasts': [],
  86                        'episodes': []}
  87
  88         self._config = config
  89         self._client = None
  90
  91         # Callback for actions that need to be handled by the UI frontend
  92         self._on_rewrite_url = on_rewrite_url
  93         self._on_add_remove_podcasts = on_add_remove_podcasts
  94         self._on_send_full_subscriptions = on_send_full_subscriptions
  95
  96         # Initialize the _client attribute and register with config
  97         self.on_config_changed('mygpo_username')
  98         assert self._client is not None
  99         self._config.add_observer(self.on_config_changed)
 100
 101         # Initialize and load the local queue
 102         self._cache_file = os.path.join(gpodder.home, self.CACHE_FILE)
 103         try:
 104             self._cache = json.loads(open(self._cache_file).read())
 105         except Exception, e:
 106             log('Cannot read cache file: %s', str(e), sender=self)
 107
 108         self._worker_thread = None
 109         atexit.register(self._at_exit)
 110
 111         # Do the initial flush (in case any actions are queued)
 112         self.flush()
 113
 114     def can_access_webservice(self):
 115         return self._config.mygpo_enabled and self._config.mygpo_device_uid
 116
 117     def schedule_podcast_sync(self):
 118         log('Scheduling podcast list sync', sender=self)
 119         self.schedule(Actions.SYNC_PODCASTS)
 120
 121     def request_podcast_lists_in_cache(self):
 122         if 'add_podcasts' not in self._cache:
 123             self._cache['add_podcasts'] = []
 124         if 'remove_podcasts' not in self._cache:
 125             self._cache['remove_podcasts'] = []
 126
 127     def force_fresh_upload(self):
 128         self._on_send_full_subscriptions()
 129
 130     def set_subscriptions(self, urls):
 131         log('Uploading (overwriting) subscriptions...')
 132         self._client.put_subscriptions(self._config.mygpo_device_uid, urls)
 133         log('Subscription upload done.')
 134
 135     def on_subscribe(self, urls):
 136         self.request_podcast_lists_in_cache()
 137         self._cache['add_podcasts'].extend(urls)
 138         for url in urls:
 139             if url in self._cache['remove_podcasts']:
 140                 self._cache['remove_podcasts'].remove(url)
 141         self.schedule(Actions.SYNC_PODCASTS)
 142         self.flush()
 143
 144     def on_unsubscribe(self, urls):
 145         self.request_podcast_lists_in_cache()
 146         self._cache['remove_podcasts'].extend(urls)
 147         for url in urls:
 148             if url in self._cache['add_podcasts']:
 149                 self._cache['add_podcasts'].remove(url)
 150         self.schedule(Actions.SYNC_PODCASTS)
 151         self.flush()
 152
 153     @property
 154     def actions(self):
 155         return self._cache.get('actions', Actions.NONE)
 156
 157     def _at_exit(self):
 158         self._worker_proc(forced=True)
 159
 160     def _worker_proc(self, forced=False):
 161         if not forced:
 162             log('Worker thread waiting for timeout', sender=self)
 163             time.sleep(self.FLUSH_TIMEOUT)
 164
 165         # Only work when enabled, UID set and allowed to work
 166         if self.can_access_webservice() and \
 167                 (self._worker_thread is not None or forced):
 168             self._worker_thread = None
 169             log('Worker thread starting to work...', sender=self)
 170             for retry in range(self.FLUSH_RETRIES):
 171                 if retry:
 172                     log('Retrying flush queue...', sender=self)
 173
 174                 # Update the device first, so it can be created if new
 175                 if self.actions & Actions.UPDATE_DEVICE:
 176                     self.update_device()
 177
 178                 if self.actions & Actions.SYNC_PODCASTS:
 179                     self.synchronize_subscriptions()
 180
 181                 if self.actions & Actions.UPLOAD_EPISODES:
 182                     # TODO: Upload episode actions
 183                     pass
 184
 185                 if not self.actions:
 186                     # No more pending actions. Ready to quit.
 187                     break
 188
 189             log('Flush completed (result: %d)', self.actions, sender=self)
 190             self._dump_cache_to_file()
 191
 192     def _dump_cache_to_file(self):
 193         try:
 194             fp = open(self._cache_file, 'w')
 195             fp.write(json.dumps(self._cache))
 196             fp.close()
 197             # FIXME: Atomic file write would be nice ;)
 198         except Exception, e:
 199             log('Cannot dump cache to file: %s', str(e), sender=self)
 200
 201     def flush(self):
 202         if not self.actions:
 203             return
 204
 205         if self._worker_thread is None:
 206             self._worker_thread = threading.Thread(target=self._worker_proc)
 207             self._worker_thread.setDaemon(True)
 208             self._worker_thread.start()
 209         else:
 210             log('Flush already queued', sender=self)
 211
 212     def schedule(self, action):
 213         if 'actions' not in self._cache:
 214             self._cache['actions'] = 0
 215
 216         self._cache['actions'] |= action
 217         self.flush()
 218
 219     def done(self, action):
 220         if 'actions' not in self._cache:
 221             self._cache['actions'] = 0
 222
 223         if action == Actions.SYNC_PODCASTS:
 224             self._cache['add_podcasts'] = []
 225             self._cache['remove_podcasts'] = []
 226
 227         self._cache['actions'] &= ~action
 228
 229     def on_config_changed(self, name=None, old_value=None, new_value=None):
 230         if name in ('mygpo_username', 'mygpo_password', 'mygpo_server'):
 231             self._client = api.MygPodderClient(self._config.mygpo_username,
 232                     self._config.mygpo_password, self._config.mygpo_server)
 233             log('Reloading settings.', sender=self)
 234         elif name.startswith('mygpo_device_'):
 235             self.schedule(Actions.UPDATE_DEVICE)
 236             if name == 'mygpo_device_uid':
 237                 # Reset everything because we have a new device ID
 238                 threading.Thread(target=self.force_fresh_upload).start()
 239                 self._cache['podcasts_since'] = 0
 240
 241     def synchronize_subscriptions(self):
 242         try:
 243             device_id = self._config.mygpo_device_uid
 244             since = self._cache.get('podcasts_since', 0)
 245
 246             # Step 1: Pull updates from the server and notify the frontend
 247             result = self._client.pull_subscriptions(device_id, since)
 248             self._cache['podcasts_since'] = result.since
 249             if result.add or result.remove:
 250                 log('Changes from server: add %d, remove %d', \
 251                         len(result.add), \
 252                         len(result.remove), \
 253                         sender=self)
 254                 self._on_add_remove_podcasts(result.add, result.remove)
 255
 256             # Step 2: Push updates to the server and rewrite URLs (if any)
 257             add = list(set(self._cache.get('add_podcasts', [])))
 258             remove = list(set(self._cache.get('remove_podcasts', [])))
 259             if add or remove:
 260                 # Only do a push request if something has changed
 261                 result = self._client.update_subscriptions(device_id, add, remove)
 262                 self._cache['podcasts_since'] = result.since
 263
 264                 for old_url, new_url in result.update_urls:
 265                     if new_url:
 266                         log('URL %s rewritten: %s', old_url, new_url, sender=self)
 267                         self._on_rewrite_url(old_url, new_url)
 268
 269             self.done(Actions.SYNC_PODCASTS)
 270             return True
 271         except Exception, e:
 272             log('Cannot upload subscriptions: %s', str(e), sender=self, traceback=True)
 273             return False
 274
 275     def update_device(self):
 276         try:
 277             log('Uploading device settings...', sender=self)
 278             uid = self._config.mygpo_device_uid
 279             caption = self._config.mygpo_device_caption
 280             device_type = self._config.mygpo_device_type
 281             self._client.update_device_settings(uid, caption, device_type)
 282             log('Device settings uploaded.', sender=self)
 283             self.done(Actions.UPDATE_DEVICE)
 284             return True
 285         except Exception, e:
 286             log('Cannot update device %s: %s', uid, str(e), sender=self, traceback=True)
 287             return False
 288
 289     def get_devices(self):
 290         result = []
 291         for d in self._client.get_devices():
 292             result.append((d.device_id, d.caption, d.type))
 293         return result
 294
 295     def open_website(self):
 296         util.open_website('http://' + self._config.mygpo_server)
 297