stov/lib_stov/noapi.py

#
#        This file is part of stov, written by Helmut Pozimski 2012-2015.
#
#       stov 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, version 2 of the License.
#
#       stov 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 stov.  If not, see <http://www.gnu.org/licenses/>.


# -*- coding: utf8 -*-

"""This module provides all classes and methods that were provided by the
youtubeAPI module earlier."""

import subprocess
import sys
import lxml.html
import urllib.parse
import urllib.request

from lib_stov import stov_exceptions


class YtChannel(object):
    """Stores the relevant attributes of a youtube channel."""
    def __init__(self):
        self.title = ""
        self.videos = []


class YtVideo(object):
    """Stores the relevant attributes of a single youtube video."""
    def __init__(self, title, description, ytid):
        self.title = title
        self.description = description
        self.ytid = ytid


class Connector(object):
    """This class will retrieve all the necessary data from youtube using
    youtube-dl, thus bypassing the API.
    """
    def __init__(self, subscription_type, name, conf, search=""):
        """Populates the object with all necessary data."""
        self._type = subscription_type
        self._name = name
        self._search = search
        self._conf = conf
        self._title = ""
        self._url = ""
        self._construct_url()

    def _construct_url(self):
        """Constructs the URL to request from youtube-dl according to the
        subscription type and the given parameters.
        """
        if self._type == "channel":
            self._url = "https://www.youtube.com/user/%s" \
                        % urllib.parse.quote(self._name)
        elif self._type == "search":
            self._url = "https://www.youtube.com/results?search_query=%s"\
                        % urllib.parse.quote(self._search)
        elif self._type == "playlist":
            self._url = "https://www.youtube.com/playlist?list=%s" \
                        % urllib.parse.quote(self._name)

    def _fetch_title(self):
        """Retrieves the title of the HTML page to use as a title for the
        subscription."""
        data = urllib.request.urlopen(self._url)
        parsed_html = lxml.html.parse(data)
        data.close()
        i = 0
        for item in parsed_html.iter("title"):
            if i == 0:
                self._title = item.text_content().strip().replace("\n", "")
            i += 1
        if self._search != "" and self._type == "channel":
            self._title += _(" search %s") % self._search

    def _fetch_videos(self, existing_videos):
        """Retrieves all the relevant videos in a subscription."""
        videos_list = []
        if self._conf.outputlevel == "verbose":
            stderr = sys.stderr
        else:
            stderr = open("/dev/null", "w")
        if self._type == "channel" and self._search != "":
            try:
                video_ids = subprocess.check_output([
                    self._conf.values["youtube-dl"],
                    "--max-downloads",
                    self._conf.values["maxvideos"],
                    "--match-title",
                    self._search,
                    "--get-id",
                    self._url], stderr=stderr).strip()
            except subprocess.CalledProcessError as error_message:
                video_ids = error_message.output.strip()
        else:
            try:
                video_ids = subprocess.check_output([
                    self._conf.values["youtube-dl"], "--max-downloads",
                    self._conf.values["maxvideos"], "--get-id",
                    self._url], stderr=stderr).strip()
            except subprocess.CalledProcessError as error_message:
                video_ids = error_message.output.strip()
        if len(video_ids) >= 1:
            for video_id in video_ids.split("\n"):
                video_exists = False
                if existing_videos:
                    for existing_video in existing_videos:
                        if video_id == existing_video.ytid:
                            video_exists = True
                            break
                if not video_exists:
                    try:
                        video_title = subprocess.check_output([
                            self._conf.values["youtube-dl"], "--get-title",
                            "https://www.youtube.com/watch?v=%s"
                            % video_id], stderr=stderr).strip()
                        video_description = subprocess.check_output([
                            self._conf.values["youtube-dl"],
                            "--get-description",
                            "https://www.youtube.com/watch?v=%s"
                            % video_id], stderr=stderr).strip()
                    except subprocess.CalledProcessError:
                        raise stov_exceptions.YoutubeDlCallFailed()
                    else:
                        videos_list.append(YtVideo(
                            video_title,
                            video_description,
                            video_id))
        return videos_list

    def parse_api_data(self, existing_videos):
        """This method calls all necessary methods to retrieve the data
                and assembles them into a Channel object. The naming of this
                method was set according to the method in youtubeAPI to be
                compatible.
        """
        self._fetch_title()
        videos = self._fetch_videos(existing_videos)
        channel = YtChannel()
        channel.title = self._title
        channel.videos = videos
        return channel