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 datetime
  32 import os
  33 import sys
  34 import threading
  35 import time
  36
  37 from gpodder.liblogger import log
  38
  39 from gpodder import util
  40 from gpodder import minidb
  41
  42 # Append gPodder's user agent to mygpoclient's user agent
  43 import mygpoclient
  44 mygpoclient.user_agent += ' ' + gpodder.user_agent
  45
  46 MYGPOCLIENT_REQUIRED = '1.4'
  47
  48 if not hasattr(mygpoclient, 'require_version') or \
  49         not mygpoclient.require_version(MYGPOCLIENT_REQUIRED):
  50     print >>sys.stderr, """
  51     Please upgrade your mygpoclient library.
  52     See http://thpinfo.com/2010/mygpoclient/
  53
  54     Required version:  %s
  55     Installed version: %s
  56     """ % (MYGPOCLIENT_REQUIRED, mygpoclient.__version__)
  57     sys.exit(1)
  58
  59 from mygpoclient import api
  60
  61 from mygpoclient import util as mygpoutil
  62
  63
  64 # Database model classes
  65 class SinceValue(object):
  66     __slots__ = {'host': str, 'device_id': str, 'category': int, 'since': int}
  67
  68     # Possible values for the "category" field
  69     PODCASTS, EPISODES = range(2)
  70
  71     def __init__(self, host, device_id, category, since=0):
  72         self.host = host
  73         self.device_id = device_id
  74         self.category = category
  75         self.since = since
  76
  77 class SubscribeAction(object):
  78     __slots__ = {'action_type': int, 'url': str}
  79
  80     # Possible values for the "action_type" field
  81     ADD, REMOVE = range(2)
  82
  83     def __init__(self, action_type, url):
  84         self.action_type = action_type
  85         self.url = url
  86
  87     @property
  88     def is_add(self):
  89         return self.action_type == self.ADD
  90
  91     @property
  92     def is_remove(self):
  93         return self.action_type == self.REMOVE
  94
  95     @classmethod
  96     def add(cls, url):
  97         return cls(cls.ADD, url)
  98
  99     @classmethod
 100     def remove(cls, url):
 101         return cls(cls.REMOVE, url)
 102
 103     @classmethod
 104     def undo(cls, action):
 105         if action.is_add:
 106             return cls(cls.REMOVE, action.url)
 107         elif action.is_remove:
 108             return cls(cls.ADD, action.url)
 109
 110         raise ValueError('Cannot undo action: %r' % action)
 111
 112 # New entity name for "received" actions
 113 class ReceivedSubscribeAction(SubscribeAction): pass
 114
 115 class UpdateDeviceAction(object):
 116     __slots__ = {'device_id': str, 'caption': str, 'device_type': str}
 117
 118     def __init__(self, device_id, caption, device_type):
 119         self.device_id = device_id
 120         self.caption = caption
 121         self.device_type = device_type
 122
 123 class EpisodeAction(object):
 124     __slots__ = {'podcast_url': str, 'episode_url': str, 'device_id': str,
 125                  'action': str, 'timestamp': int,
 126                  'started': int, 'position': int, 'total': int}
 127
 128     def __init__(self, podcast_url, episode_url, device_id, \
 129             action, timestamp, started, position, total):
 130         self.podcast_url = podcast_url
 131         self.episode_url = episode_url
 132         self.device_id = device_id
 133         self.action = action
 134         self.timestamp = timestamp
 135         self.started = started
 136         self.position = position
 137         self.total = total
 138
 139 # New entity name for "received" actions
 140 class ReceivedEpisodeAction(EpisodeAction): pass
 141
 142 class RewrittenUrl(object):
 143     __slots__ = {'old_url': str, 'new_url': str}
 144
 145     def __init__(self, old_url, new_url):
 146         self.old_url = old_url
 147         self.new_url = new_url
 148 # End Database model classes
 149
 150
 151
 152 # Helper class for displaying changes in the UI
 153 class Change(object):
 154     def __init__(self, action, podcast=None):
 155         self.action = action
 156         self.podcast = podcast
 157
 158     @property
 159     def description(self):
 160         if self.action.is_add:
 161             return _('Add %s') % self.action.url
 162         else:
 163             return _('Remove %s') % self.podcast.title
 164
 165
 166 class MygPoClient(object):
 167     STORE_FILE = 'mygpo.queue.sqlite'
 168     FLUSH_TIMEOUT = 60
 169     FLUSH_RETRIES = 3
 170
 171     def __init__(self, config):
 172         self._store = minidb.Store(os.path.join(gpodder.home, self.STORE_FILE))
 173
 174         self._config = config
 175         self._client = None
 176
 177         # Initialize the _client attribute and register with config
 178         self.on_config_changed()
 179         assert self._client is not None
 180
 181         self._config.add_observer(self.on_config_changed)
 182
 183         self._worker_thread = None
 184         atexit.register(self._at_exit)
 185
 186     def create_device(self):
 187         """Uploads the device changes to the server
 188
 189         This should be called when device settings change
 190         or when the mygpo client functionality is enabled.
 191         """
 192         # Remove all previous device update actions
 193         self._store.remove(self._store.load(UpdateDeviceAction))
 194
 195         # Insert our new update action
 196         action = UpdateDeviceAction(self.device_id, \
 197                 self._config.mygpo_device_caption, \
 198                 self._config.mygpo_device_type)
 199         self._store.save(action)
 200
 201     def get_rewritten_urls(self):
 202         """Returns a list of rewritten URLs for uploads
 203
 204         This should be called regularly. Every object returned
 205         should be merged into the database, and the old_url
 206         should be updated to new_url in every podcdast.
 207         """
 208         rewritten_urls = self._store.load(RewrittenUrl)
 209         self._store.remove(rewritten_urls)
 210         return rewritten_urls
 211
 212     def get_received_actions(self):
 213         """Returns a list of ReceivedSubscribeAction objects
 214
 215         The list might be empty. All these actions have to
 216         be processed. The user should confirm which of these
 217         actions should be taken, the reest should be rejected.
 218
 219         Use confirm_received_actions and reject_received_actions
 220         to return and finalize the actions received by this
 221         method in order to not receive duplicate actions.
 222         """
 223         return self._store.load(ReceivedSubscribeAction)
 224
 225     def confirm_received_actions(self, actions):
 226         """Confirm that a list of actions has been processed
 227
 228         The UI should call this with a list of actions that
 229         have been accepted by the user and processed by the
 230         podcast backend.
 231         """
 232         # Simply remove the received actions from the queue
 233         self._store.remove(actions)
 234
 235     def reject_received_actions(self, actions):
 236         """Reject (undo) a list of ReceivedSubscribeAction objects
 237
 238         The UI should call this with a list of actions that
 239         have been rejected by the user. A reversed set of
 240         actions will be uploaded to the server so that the
 241         state on the server matches the state on the client.
 242         """
 243         # Create "undo" actions for received subscriptions
 244         self._store.save(SubscribeAction.undo(a) for a in actions)
 245         self.flush()
 246
 247         # After we've handled the reverse-actions, clean up
 248         self._store.remove(actions)
 249
 250     @property
 251     def host(self):
 252         return self._config.mygpo_server
 253
 254     @property
 255     def device_id(self):
 256         return self._config.mygpo_device_uid
 257
 258     def can_access_webservice(self):
 259         return self._config.mygpo_enabled and self._config.mygpo_device_uid
 260
 261     def set_subscriptions(self, urls):
 262         if self.can_access_webservice():
 263             log('Uploading (overwriting) subscriptions...')
 264             self._client.put_subscriptions(self.device_id, urls)
 265             log('Subscription upload done.')
 266         else:
 267             raise Exception('Webservice access not enabled')
 268
 269     def _convert_played_episode(self, episode, start, end, total):
 270         return EpisodeAction(episode.channel.url, \
 271                 episode.url, self.device_id, 'play', \
 272                 int(time.time()), start, end, total)
 273
 274     def _convert_episode(self, episode, action):
 275         return EpisodeAction(episode.channel.url, \
 276                 episode.url, self.device_id, action, \
 277                 int(time.time()), None, None, None)
 278
 279     def on_delete(self, episodes):
 280         log('Storing %d episode delete actions', len(episodes), sender=self)
 281         self._store.save(self._convert_episode(e, 'delete') for e in episodes)
 282
 283     def on_download(self, episodes):
 284         log('Storing %d episode download actions', len(episodes), sender=self)
 285         self._store.save(self._convert_episode(e, 'download') for e in episodes)
 286
 287     def on_playback_full(self, episode, start, end, total):
 288         log('Storing full episode playback action', sender=self)
 289         self._store.save(self._convert_played_episode(episode, start, end, total))
 290
 291     def on_playback(self, episodes):
 292         log('Storing %d episode playback actions', len(episodes), sender=self)
 293         self._store.save(self._convert_episode(e, 'play') for e in episodes)
 294
 295     def on_subscribe(self, urls):
 296         # Cancel previously-inserted "remove" actions
 297         self._store.remove(SubscribeAction.remove(url) for url in urls)
 298
 299         # Insert new "add" actions
 300         self._store.save(SubscribeAction.add(url) for url in urls)
 301
 302         self.flush()
 303
 304     def on_unsubscribe(self, urls):
 305         # Cancel previously-inserted "add" actions
 306         self._store.remove(SubscribeAction.add(url) for url in urls)
 307
 308         # Insert new "remove" actions
 309         self._store.save(SubscribeAction.remove(url) for url in urls)
 310
 311         self.flush()
 312
 313     @property
 314     def actions(self):
 315         return self._cache.get('actions', Actions.NONE)
 316
 317     def _at_exit(self):
 318         self._worker_proc(forced=True)
 319         self._store.commit()
 320         self._store.close()
 321
 322     def _worker_proc(self, forced=False):
 323         if not forced:
 324             # Store the current contents of the queue database
 325             self._store.commit()
 326
 327             log('Worker thread waiting for timeout', sender=self)
 328             time.sleep(self.FLUSH_TIMEOUT)
 329
 330         # Only work when enabled, UID set and allowed to work
 331         if self.can_access_webservice() and \
 332                 (self._worker_thread is not None or forced):
 333             self._worker_thread = None
 334
 335             log('Worker thread starting to work...', sender=self)
 336             for retry in range(self.FLUSH_RETRIES):
 337                 must_retry = False
 338
 339                 if retry:
 340                     log('Retrying flush queue...', sender=self)
 341
 342                 # Update the device first, so it can be created if new
 343                 for action in self._store.load(UpdateDeviceAction):
 344                     if self.update_device(action):
 345                         self._store.remove(action)
 346                     else:
 347                         must_retry = True
 348
 349                 # Upload podcast subscription actions
 350                 actions = self._store.load(SubscribeAction)
 351                 if self.synchronize_subscriptions(actions):
 352                     self._store.remove(actions)
 353                 else:
 354                     must_retry = True
 355
 356                 # Upload episode actions
 357                 actions = self._store.load(EpisodeAction)
 358                 if self.synchronize_episodes(actions):
 359                     self._store.remove(actions)
 360                 else:
 361                     must_retry = True
 362
 363                 if not must_retry:
 364                     # No more pending actions. Ready to quit.
 365                     break
 366
 367             log('Worker thread finished.', sender=self)
 368         else:
 369             log('Worker thread may not execute (disabled).', sender=self)
 370
 371         # Store the current contents of the queue database
 372         self._store.commit()
 373
 374     def flush(self, now=False):
 375         if not self.can_access_webservice():
 376             log('Flush requested, but sync disabled.', sender=self)
 377             return
 378
 379         if self._worker_thread is None or now:
 380             if now:
 381                 log('Flushing NOW.', sender=self)
 382             else:
 383                 log('Flush requested.', sender=self)
 384             self._worker_thread = threading.Thread(target=self._worker_proc, args=[now])
 385             self._worker_thread.setDaemon(True)
 386             self._worker_thread.start()
 387         else:
 388             log('Flush requested, already waiting.', sender=self)
 389
 390     def on_config_changed(self, name=None, old_value=None, new_value=None):
 391         if name in ('mygpo_username', 'mygpo_password', 'mygpo_server') \
 392                 or self._client is None:
 393             self._client = api.MygPodderClient(self._config.mygpo_username,
 394                     self._config.mygpo_password, self._config.mygpo_server)
 395             log('Reloading settings.', sender=self)
 396         elif name.startswith('mygpo_device_'):
 397             # Update or create the device
 398             self.create_device()
 399
 400     def synchronize_episodes(self, actions):
 401         log('Starting episode status sync.', sender=self)
 402
 403         def convert_to_api(action):
 404             dt = datetime.datetime.fromtimestamp(action.timestamp)
 405             since = mygpoutil.datetime_to_iso8601(dt)
 406             return api.EpisodeAction(action.podcast_url, \
 407                     action.episode_url, action.action, \
 408                     action.device_id, since, \
 409                     action.started, action.position, action.total)
 410
 411         def convert_from_api(action):
 412             dt = mygpoutil.iso8601_to_datetime(action.timestamp)
 413             since = int(dt.strftime('%s'))
 414             return ReceivedEpisodeAction(action.podcast, \
 415                     action.episode, action.device, \
 416                     action.action, since, \
 417                     action.started, action.position, action.total)
 418
 419         try:
 420             save_since = True
 421
 422             # Load the "since" value from the database
 423             since_o = self._store.get(SinceValue, host=self.host, \
 424                                                   device_id=self.device_id, \
 425                                                   category=SinceValue.EPISODES)
 426
 427             # Use a default since object for the first-time case
 428             if since_o is None:
 429                 since_o = SinceValue(self.host, self.device_id, SinceValue.EPISODES)
 430
 431             # Step 1: Download Episode actions
 432             try:
 433                 changes = self._client.download_episode_actions(since_o.since, \
 434                         device_id=self.device_id)
 435
 436                 received_actions = [convert_from_api(a) for a in changes.actions]
 437                 self._store.save(received_actions)
 438
 439                 # Save the "since" value for later use
 440                 self._store.update(since_o, since=changes.since)
 441             except Exception, e:
 442                 log('Exception while polling for episodes.', sender=self, traceback=True)
 443                 save_since = False
 444
 445             # Step 2: Upload Episode actions
 446
 447             # Convert actions to the mygpoclient format for uploading
 448             episode_actions = [convert_to_api(a) for a in actions]
 449
 450             # Upload the episodes and retrieve the new "since" value
 451             since = self._client.upload_episode_actions(episode_actions)
 452
 453             if save_since:
 454                 # Update the "since" value of the episodes
 455                 self._store.update(since_o, since=since)
 456
 457             # Actions have been uploaded to the server - remove them
 458             self._store.remove(actions)
 459             log('Episode actions have been uploaded to the server.', sender=self)
 460             return True
 461         except Exception, e:
 462             log('Cannot upload episode actions: %s', str(e), sender=self, traceback=True)
 463             return False
 464
 465     def synchronize_subscriptions(self, actions):
 466         log('Starting subscription sync.', sender=self)
 467         try:
 468             # Load the "since" value from the database
 469             since_o = self._store.get(SinceValue, host=self.host, \
 470                                                   device_id=self.device_id, \
 471                                                   category=SinceValue.PODCASTS)
 472
 473             # Use a default since object for the first-time case
 474             if since_o is None:
 475                 since_o = SinceValue(self.host, self.device_id, SinceValue.PODCASTS)
 476
 477             # Step 1: Pull updates from the server and notify the frontend
 478             result = self._client.pull_subscriptions(self.device_id, since_o.since)
 479
 480             # Update the "since" value in the database
 481             self._store.update(since_o, since=result.since)
 482
 483             # Store received actions for later retrieval (and in case we
 484             # have outdated actions in the database, simply remove them)
 485             for url in result.add:
 486                 log('Received add action: %s', url, sender=self)
 487                 self._store.remove(ReceivedSubscribeAction.remove(url))
 488                 self._store.remove(ReceivedSubscribeAction.add(url))
 489                 self._store.save(ReceivedSubscribeAction.add(url))
 490             for url in result.remove:
 491                 log('Received remove action: %s', url, sender=self)
 492                 self._store.remove(ReceivedSubscribeAction.add(url))
 493                 self._store.remove(ReceivedSubscribeAction.remove(url))
 494                 self._store.save(ReceivedSubscribeAction.remove(url))
 495
 496             # Step 2: Push updates to the server and rewrite URLs (if any)
 497             actions = self._store.load(SubscribeAction)
 498
 499             add = [a.url for a in actions if a.is_add]
 500             remove = [a.url for a in actions if a.is_remove]
 501
 502             if add or remove:
 503                 log('Uploading: +%d / -%d', len(add), len(remove), sender=self)
 504                 # Only do a push request if something has changed
 505                 result = self._client.update_subscriptions(self.device_id, add, remove)
 506
 507                 # Update the "since" value in the database
 508                 self._store.update(since_o, since=result.since)
 509
 510                 # Store URL rewrites for later retrieval by GUI
 511                 for old_url, new_url in result.update_urls:
 512                     if new_url:
 513                         log('Rewritten URL: %s', new_url, sender=self)
 514                         self._store.save(RewrittenUrl(old_url, new_url))
 515
 516             # Actions have been uploaded to the server - remove them
 517             self._store.remove(actions)
 518             log('All actions have been uploaded to the server.', sender=self)
 519             return True
 520         except Exception, e:
 521             log('Cannot upload subscriptions: %s', str(e), sender=self, traceback=True)
 522             return False
 523
 524     def update_device(self, action):
 525         try:
 526             log('Uploading device settings...', sender=self)
 527             self._client.update_device_settings(action.device_id, \
 528                     action.caption, action.device_type)
 529             log('Device settings uploaded.', sender=self)
 530             return True
 531         except Exception, e:
 532             log('Cannot update device %s: %s', self.device_id, str(e), sender=self, traceback=True)
 533             return False
 534
 535     def get_devices(self):
 536         result = []
 537         for d in self._client.get_devices():
 538             result.append((d.device_id, d.caption, d.type))
 539         return result
 540
 541     def open_website(self):
 542         util.open_website('http://' + self._config.mygpo_server)
 543