gpodder/src/gpodder/my.py

# -*- coding: utf-8 -*-
#
# gPodder - A media aggregator and podcast client
# Copyright (c) 2005-2018 The gPodder Team
#
# gPodder is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# gPodder is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#


#
#  my.py -- mygpo Client Abstraction for gPodder
#  Thomas Perl <thp@gpodder.org>; 2010-01-19
#

import atexit
import calendar
import datetime
import logging
import os
import sys
import time

# Append gPodder's user agent to mygpoclient's user agent
import mygpoclient
from mygpoclient import api, public
from mygpoclient import util as mygpoutil

import gpodder
from gpodder import minidb, util

_ = gpodder.gettext


logger = logging.getLogger(__name__)


mygpoclient.user_agent += ' ' + gpodder.user_agent

# 2013-02-08: We should update this to 1.7 once we use the new features
MYGPOCLIENT_REQUIRED = '1.4'

if (not hasattr(mygpoclient, 'require_version')
        or not mygpoclient.require_version(MYGPOCLIENT_REQUIRED)):
    print("""
    Please upgrade your mygpoclient library.
    See http://thp.io/2010/mygpoclient/

    Required version:  %s
    Installed version: %s
    """ % (MYGPOCLIENT_REQUIRED, mygpoclient.__version__), file=sys.stderr)
    sys.exit(1)

try:
    from mygpoclient.simple import MissingCredentials

except ImportError:
    # if MissingCredentials does not yet exist in the installed version of
    # mygpoclient, we use an object that can never be raised/caught
    MissingCredentials = object()


EPISODE_ACTIONS_BATCH_SIZE = 100


# Database model classes
class SinceValue(object):
    __slots__ = {'host': str, 'device_id': str, 'category': int, 'since': int}

    # Possible values for the "category" field
    PODCASTS, EPISODES = list(range(2))

    def __init__(self, host, device_id, category, since=0):
        self.host = host
        self.device_id = device_id
        self.category = category
        self.since = since


class SubscribeAction(object):
    __slots__ = {'action_type': int, 'url': str}

    # Possible values for the "action_type" field
    ADD, REMOVE = list(range(2))

    def __init__(self, action_type, url):
        self.action_type = action_type
        self.url = url

    @property
    def is_add(self):
        return self.action_type == self.ADD

    @property
    def is_remove(self):
        return self.action_type == self.REMOVE

    @classmethod
    def add(cls, url):
        return cls(cls.ADD, url)

    @classmethod
    def remove(cls, url):
        return cls(cls.REMOVE, url)

    @classmethod
    def undo(cls, action):
        if action.is_add:
            return cls(cls.REMOVE, action.url)
        elif action.is_remove:
            return cls(cls.ADD, action.url)

        raise ValueError('Cannot undo action: %r' % action)


# New entity name for "received" actions
class ReceivedSubscribeAction(SubscribeAction): pass


class UpdateDeviceAction(object):
    __slots__ = {'device_id': str, 'caption': str, 'device_type': str}

    def __init__(self, device_id, caption, device_type):
        self.device_id = device_id
        self.caption = caption
        self.device_type = device_type


class EpisodeAction(object):
    __slots__ = {'podcast_url': str, 'episode_url': str, 'device_id': str,
                 'action': str, 'timestamp': int,
                 'started': int, 'position': int, 'total': int}

    def __init__(self, podcast_url, episode_url, device_id,
            action, timestamp, started, position, total):
        self.podcast_url = podcast_url
        self.episode_url = episode_url
        self.device_id = device_id
        self.action = action
        self.timestamp = timestamp
        self.started = started
        self.position = position
        self.total = total


# New entity name for "received" actions
class ReceivedEpisodeAction(EpisodeAction): pass


class RewrittenUrl(object):
    __slots__ = {'old_url': str, 'new_url': str}

    def __init__(self, old_url, new_url):
        self.old_url = old_url
        self.new_url = new_url
# End Database model classes


# Helper class for displaying changes in the UI
class Change(object):
    def __init__(self, action, podcast=None):
        self.action = action
        self.podcast = podcast

    @property
    def description(self):
        if self.action.is_add:
            return _('Add %s') % self.action.url
        else:
            return _('Remove %s') % self.podcast.title


class MygPoClient(object):
    STORE_FILE = 'gpodder.net'
    FLUSH_TIMEOUT = 60
    FLUSH_RETRIES = 3

    def __init__(self, config):
        self._store = minidb.Store(os.path.join(gpodder.home, self.STORE_FILE))

        self._config = config
        self._client = None

        # Initialize the _client attribute and register with config
        self.on_config_changed()
        assert self._client is not None

        self._config.add_observer(self.on_config_changed)

        self._worker_thread = None
        atexit.register(self._at_exit)

    def create_device(self):
        """Uploads the device changes to the server

        This should be called when device settings change
        or when the mygpo client functionality is enabled.
        """
        # Remove all previous device update actions
        self._store.remove(self._store.load(UpdateDeviceAction))

        # Insert our new update action
        action = UpdateDeviceAction(self.device_id,
                self._config.mygpo.device.caption,
                self._config.mygpo.device.type)
        self._store.save(action)

    def get_rewritten_urls(self):
        """Returns a list of rewritten URLs for uploads

        This should be called regularly. Every object returned
        should be merged into the database, and the old_url
        should be updated to new_url in every podcdast.
        """
        rewritten_urls = self._store.load(RewrittenUrl)
        self._store.remove(rewritten_urls)
        return rewritten_urls

    def process_episode_actions(self, find_episode, on_updated=None):
        """Process received episode actions

        The parameter "find_episode" should be a function accepting
        two parameters (podcast_url and episode_url). It will be used
        to get an episode object that needs to be updated. It should
        return None if the requested episode does not exist.

        The optional callback "on_updated" should accept a single
        parameter (the episode object) and will be called whenever
        the episode data is changed in some way.
        """
        logger.debug('Processing received episode actions...')
        for action in self._store.load(ReceivedEpisodeAction):
            if action.action not in ('play', 'delete'):
                # Ignore all other action types for now
                continue

            episode = find_episode(action.podcast_url, action.episode_url)

            if episode is None:
                # The episode does not exist on this client
                continue

            if action.action == 'play':
                logger.debug('Play action for %s', episode.url)
                episode.mark(is_played=True)

                if (action.timestamp > episode.current_position_updated
                        and action.position is not None):
                    logger.debug('Updating position for %s', episode.url)
                    episode.current_position = action.position
                    episode.current_position_updated = action.timestamp

                if action.total:
                    logger.debug('Updating total time for %s', episode.url)
                    episode.total_time = action.total

                episode.save()
                if on_updated is not None:
                    on_updated(episode)
            elif action.action == 'delete':
                if not episode.was_downloaded(and_exists=True):
                    # Set the episode to a "deleted" state
                    logger.debug('Marking as deleted: %s', episode.url)
                    episode.delete_from_disk()
                    episode.save()
                    if on_updated is not None:
                        on_updated(episode)

        # Remove all received episode actions
        self._store.delete(ReceivedEpisodeAction)
        self._store.commit()
        logger.debug('Received episode actions processed.')

    def get_received_actions(self):
        """Returns a list of ReceivedSubscribeAction objects

        The list might be empty. All these actions have to
        be processed. The user should confirm which of these
        actions should be taken, the rest should be rejected.

        Use confirm_received_actions and reject_received_actions
        to return and finalize the actions received by this
        method in order to not receive duplicate actions.
        """
        return self._store.load(ReceivedSubscribeAction)

    def confirm_received_actions(self, actions):
        """Confirm that a list of actions has been processed

        The UI should call this with a list of actions that
        have been accepted by the user and processed by the
        podcast backend.
        """
        # Simply remove the received actions from the queue
        self._store.remove(actions)

    def reject_received_actions(self, actions):
        """Reject (undo) a list of ReceivedSubscribeAction objects

        The UI should call this with a list of actions that
        have been rejected by the user. A reversed set of
        actions will be uploaded to the server so that the
        state on the server matches the state on the client.
        """
        # Create "undo" actions for received subscriptions
        self._store.save(SubscribeAction.undo(a) for a in actions)
        self.flush()

        # After we've handled the reverse-actions, clean up
        self._store.remove(actions)

    @property
    def host(self):
        return self._config.mygpo.server

    @property
    def device_id(self):
        return self._config.mygpo.device.uid

    def can_access_webservice(self):
        return self._config.mygpo.enabled and \
               self._config.mygpo.username and \
               self._config.mygpo.device.uid

    def set_subscriptions(self, urls):
        if self.can_access_webservice():
            logger.debug('Uploading (overwriting) subscriptions...')
            self._client.put_subscriptions(self.device_id, urls)
            logger.debug('Subscription upload done.')
        else:
            raise Exception('Webservice access not enabled')

    def _convert_played_episode(self, episode, start, end, total):
        return EpisodeAction(episode.channel.url,
                episode.url, self.device_id, 'play',
                int(time.time()), start, end, total)

    def _convert_episode(self, episode, action):
        return EpisodeAction(episode.channel.url,
                episode.url, self.device_id, action,
                int(time.time()), None, None, None)

    def on_delete(self, episodes):
        logger.debug('Storing %d episode delete actions', len(episodes))
        self._store.save(self._convert_episode(e, 'delete') for e in episodes)

    def on_download(self, episodes):
        logger.debug('Storing %d episode download actions', len(episodes))
        self._store.save(self._convert_episode(e, 'download') for e in episodes)

    def on_playback_full(self, episode, start, end, total):
        logger.debug('Storing full episode playback action')
        self._store.save(self._convert_played_episode(episode, start, end, total))

    def on_playback(self, episodes):
        logger.debug('Storing %d episode playback actions', len(episodes))
        self._store.save(self._convert_episode(e, 'play') for e in episodes)

    def on_subscribe(self, urls):
        # Cancel previously-inserted "remove" actions
        self._store.remove(SubscribeAction.remove(url) for url in urls)

        # Insert new "add" actions
        self._store.save(SubscribeAction.add(url) for url in urls)

        self.flush()

    def on_unsubscribe(self, urls):
        # Cancel previously-inserted "add" actions
        self._store.remove(SubscribeAction.add(url) for url in urls)

        # Insert new "remove" actions
        self._store.save(SubscribeAction.remove(url) for url in urls)

        self.flush()

    def _at_exit(self):
        self._worker_proc(forced=True)
        self._store.commit()
        self._store.close()

    def _worker_proc(self, forced=False):
        if not forced:
            # Store the current contents of the queue database
            self._store.commit()

            logger.debug('Worker thread waiting for timeout')
            time.sleep(self.FLUSH_TIMEOUT)

        # Only work when enabled, UID set and allowed to work
        if self.can_access_webservice() and \
                (self._worker_thread is not None or forced):
            self._worker_thread = None

            logger.debug('Worker thread starting to work...')
            for retry in range(self.FLUSH_RETRIES):
                must_retry = False

                if retry:
                    logger.debug('Retrying flush queue...')

                # Update the device first, so it can be created if new
                for action in self._store.load(UpdateDeviceAction):
                    if self.update_device(action):
                        self._store.remove(action)
                    else:
                        must_retry = True

                # Upload podcast subscription actions
                actions = self._store.load(SubscribeAction)
                if self.synchronize_subscriptions(actions):
                    self._store.remove(actions)
                else:
                    must_retry = True

                # Upload episode actions
                actions = self._store.load(EpisodeAction)
                if self.synchronize_episodes(actions):
                    self._store.remove(actions)
                else:
                    must_retry = True

                if not must_retry or not self.can_access_webservice():
                    # No more pending actions, or no longer enabled.
                    # Ready to quit.
                    break

            logger.debug('Worker thread finished.')
        else:
            logger.info('Worker thread may not execute (disabled).')

        # Store the current contents of the queue database
        self._store.commit()

    def flush(self, now=False):
        if not self.can_access_webservice():
            logger.warning('Flush requested, but sync disabled.')
            return

        if self._worker_thread is None or now:
            if now:
                logger.debug('Flushing NOW.')
            else:
                logger.debug('Flush requested.')
            self._worker_thread = util.run_in_background(lambda: self._worker_proc(now), True)
        else:
            logger.debug('Flush requested, already waiting.')

    def on_config_changed(self, name=None, old_value=None, new_value=None):
        if name in ('mygpo.username', 'mygpo.password', 'mygpo.server') \
                or self._client is None:
            self._client = api.MygPodderClient(self._config.mygpo.username,
                    self._config.mygpo.password, self._config.mygpo.server)
            logger.info('Reloading settings.')
        elif name.startswith('mygpo.device.'):
            # Update or create the device
            self.create_device()

    def synchronize_episodes(self, actions):
        logger.debug('Starting episode status sync.')

        def convert_to_api(action):
            dt = datetime.datetime.utcfromtimestamp(action.timestamp)
            action_ts = mygpoutil.datetime_to_iso8601(dt)
            return api.EpisodeAction(action.podcast_url,
                    action.episode_url, action.action,
                    action.device_id, action_ts,
                    action.started, action.position, action.total)

        def convert_from_api(action):
            dt = mygpoutil.iso8601_to_datetime(action.timestamp)
            action_ts = calendar.timegm(dt.timetuple())
            return ReceivedEpisodeAction(action.podcast,
                    action.episode, action.device,
                    action.action, action_ts,
                    action.started, action.position, action.total)

        try:
            # Load the "since" value from the database
            since_o = self._store.get(SinceValue, host=self.host,
                                      device_id=self.device_id,
                                      category=SinceValue.EPISODES)

            # Use a default since object for the first-time case
            if since_o is None:
                since_o = SinceValue(self.host, self.device_id, SinceValue.EPISODES)

            # Step 1: Download Episode actions
            try:
                changes = self._client.download_episode_actions(since_o.since)

                received_actions = [convert_from_api(a) for a in changes.actions]
                logger.debug('Received %d episode actions', len(received_actions))
                self._store.save(received_actions)

                # Save the "since" value for later use
                self._store.update(since_o, since=changes.since)

            except (MissingCredentials, mygpoclient.http.Unauthorized):
                # handle outside
                raise

            except Exception as e:
                logger.warning('Exception while polling for episodes.', exc_info=True)

            # Step 2: Upload Episode actions

            # Uploads are done in batches; uploading can resume if only parts
            # be uploaded; avoids empty uploads as well
            for lower in range(0, len(actions), EPISODE_ACTIONS_BATCH_SIZE):
                batch = actions[lower:(lower + EPISODE_ACTIONS_BATCH_SIZE)]

                # Convert actions to the mygpoclient format for uploading
                episode_actions = [convert_to_api(a) for a in batch]

                # Upload the episode actions
                self._client.upload_episode_actions(episode_actions)

                # Actions have been uploaded to the server - remove them
                self._store.remove(batch)

            logger.debug('Episode actions have been uploaded to the server.')
            return True

        except (MissingCredentials, mygpoclient.http.Unauthorized):
            logger.warning('Invalid credentials. Disabling gpodder.net.')
            self._config.mygpo.enabled = False
            return False

        except Exception as e:
            logger.error('Cannot upload episode actions: %s', str(e), exc_info=True)
            return False

    def synchronize_subscriptions(self, actions):
        logger.debug('Starting subscription sync.')
        try:
            # Load the "since" value from the database
            since_o = self._store.get(SinceValue, host=self.host,
                                      device_id=self.device_id,
                                      category=SinceValue.PODCASTS)

            # Use a default since object for the first-time case
            if since_o is None:
                since_o = SinceValue(self.host, self.device_id, SinceValue.PODCASTS)

            # Step 1: Pull updates from the server and notify the frontend
            result = self._client.pull_subscriptions(self.device_id, since_o.since)

            # Update the "since" value in the database
            self._store.update(since_o, since=result.since)

            # Store received actions for later retrieval (and in case we
            # have outdated actions in the database, simply remove them)
            for url in result.add:
                logger.debug('Received add action: %s', url)
                self._store.remove(ReceivedSubscribeAction.remove(url))
                self._store.remove(ReceivedSubscribeAction.add(url))
                self._store.save(ReceivedSubscribeAction.add(url))
            for url in result.remove:
                logger.debug('Received remove action: %s', url)
                self._store.remove(ReceivedSubscribeAction.add(url))
                self._store.remove(ReceivedSubscribeAction.remove(url))
                self._store.save(ReceivedSubscribeAction.remove(url))

            # Step 2: Push updates to the server and rewrite URLs (if any)
            actions = self._store.load(SubscribeAction)

            add = [a.url for a in actions if a.is_add]
            remove = [a.url for a in actions if a.is_remove]

            if add or remove:
                logger.debug('Uploading: +%d / -%d', len(add), len(remove))
                # Only do a push request if something has changed
                result = self._client.update_subscriptions(self.device_id, add, remove)

                # Update the "since" value in the database
                self._store.update(since_o, since=result.since)

                # Store URL rewrites for later retrieval by GUI
                for old_url, new_url in result.update_urls:
                    if new_url:
                        logger.debug('Rewritten URL: %s', new_url)
                        self._store.save(RewrittenUrl(old_url, new_url))

            # Actions have been uploaded to the server - remove them
            self._store.remove(actions)
            logger.debug('All actions have been uploaded to the server.')
            return True

        except (MissingCredentials, mygpoclient.http.Unauthorized):
            logger.warning('Invalid credentials. Disabling gpodder.net.')
            self._config.mygpo.enabled = False
            return False

        except Exception as e:
            logger.error('Cannot upload subscriptions: %s', str(e), exc_info=True)
            return False

    def update_device(self, action):
        try:
            logger.debug('Uploading device settings...')
            self._client.update_device_settings(action.device_id,
                    action.caption, action.device_type)
            logger.debug('Device settings uploaded.')
            return True

        except (MissingCredentials, mygpoclient.http.Unauthorized):
            logger.warning('Invalid credentials. Disabling gpodder.net.')
            self._config.mygpo.enabled = False
            return False

        except Exception as e:
            logger.error('Cannot update device %s: %s', self.device_id,
                str(e), exc_info=True)
            return False

    def get_devices(self):
        result = []

        try:
            devices = self._client.get_devices()

        except (MissingCredentials, mygpoclient.http.Unauthorized):
            logger.warning('Invalid credentials. Disabling gpodder.net.')
            self._config.mygpo.enabled = False
            raise

        for d in devices:
            result.append((d.device_id, d.caption, d.type))
        return result

    def open_website(self):
        util.open_website('http://' + self._config.mygpo.server)

    def get_download_user_subscriptions_url(self):
        OPML_URL = self._client.locator.subscriptions_uri()
        url = util.url_add_authentication(OPML_URL,
                self._config.mygpo.username,
                self._config.mygpo.password)
        return url


class Directory(object):
    def __init__(self):
        self.client = public.PublicClient()

    def toplist(self):
        return [(p.title or p.url, p.url)
                for p in self.client.get_toplist()
                if p.url]

    def search(self, query):
        return [(p.title or p.url, p.url)
                for p in self.client.search_podcasts(query)
                if p.url]