stov/lib_stov/yt_noapi.py

#
#        This file is part of stov, written by Helmut Pozimski 2012-2021.
#
#       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 logging
import subprocess
import urllib.error
import urllib.parse
import urllib.request

import lxml.html

from lib_stov import configuration
from lib_stov import stov_exceptions
from lib_stov import youtubedl_wrapper

LOGGER = logging.getLogger("stov")


class YtChannel:
    """Stores the relevant attributes of a youtube channel."""

    def __init__(self, _type, title, videos=None):
        if videos is None:
            videos = []
        self.type = _type
        self.title = title
        self.videos = videos


class YtVideo:
    """Stores the relevant attributes of a single youtube video."""

    def __init__(self, title, ytid):
        self.title = title
        self.video_id = ytid


class Connector:
    """This class will retrieve all the necessary data from youtube using
    youtube-dl, thus bypassing the API.
    """

    def __init__(self, subscription_type, name, search=""):
        """Populates the object with all necessary data."""
        self._type = subscription_type
        self._name = name
        self._search = search
        self._conf = configuration.Conf.get_instance()
        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 == "user":
            self._url = "https://www.youtube.com/user/%s" \
                        % urllib.parse.quote(self._name)
        elif self._type == "channel":
            self._url = "https://www.youtube.com/channel/%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)
        LOGGER.debug(_("Constructed URL for subscription: %s"), self._url)

    def _fetch_title(self):
        """Retrieves the title of the HTML page to use as a title for the
        subscription."""
        try:
            LOGGER.debug(_("Opening URL: %s to fetch title"), self._url)
            data = urllib.request.urlopen(self._url)
        except urllib.error.HTTPError as err:
            if err.code == 404 and self._type == "user":
                self._type = "channel"
                self._construct_url()
                try:
                    data = urllib.request.urlopen(self._url)
                except urllib.error.HTTPError as exc:
                    raise stov_exceptions.ChannelNotFound() from exc
                else:
                    self._parse_title(data)
            else:
                raise stov_exceptions.ChannelNotFound()
        else:
            self._parse_title(data)

    def _parse_title(self, data):
        """ Parses the title from a HTML document
        :param data: HTTP connection to the document
        :type data: http.client.HTTPResponse
        """
        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 == "user":
            self._title += _(" search %s") % self._search

    def _fetch_videos(self, existing_videos):
        """Retrieves all the relevant videos in a subscription."""
        videos_list = []
        if self._type == "user" and self._search:
            video_ids = youtubedl_wrapper.get_ids(self._url)
        elif self._type == "playlist":
            video_ids = youtubedl_wrapper.get_ids(self._url, reverse=True)
        else:
            video_ids = youtubedl_wrapper.get_ids(self._url)
        LOGGER.debug("Got video IDs: %s", video_ids)
        if video_ids:
            for video_id in video_ids:
                video_exists = False
                if existing_videos:
                    for existing_video in existing_videos:
                        if video_id == existing_video.site_id:
                            video_exists = True
                            break
                if not video_exists:
                    try:
                        video_title = youtubedl_wrapper.get_title(
                            self.construct_video_url(video_id))
                    except subprocess.CalledProcessError as exc:
                        raise stov_exceptions.YoutubeDlCallFailed() from exc
                    else:
                        videos_list.append(YtVideo(
                            video_title,
                            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(self._type, self._title, videos)
        return channel

    @staticmethod
    def construct_video_url(ytid):
        """
        Resturns the URL to a specified youtube video

        :param ytid: Youtube ID of the video
        :type ytid: str
        :return: Video URL
        :rtype: str
        """
        url = "https://www.youtube.com/watch?v=%s" % ytid
        return url