# # This file is part of stov, written by Helmut Pozimski 2012-2017. # # 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 . # -*- coding: utf8 -*- """This module provides all classes and methods that were provided by the youtubeAPI module earlier.""" import subprocess import urllib.parse import urllib.request import urllib.error import logging import lxml.html from lib_stov import stov_exceptions from lib_stov import youtubedl_wrapper LOGGER = logging.getLogger("stov") class YtChannel(object): """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(object): """Stores the relevant attributes of a single youtube video.""" def __init__(self, title, ytid): self.title = title self.video_id = 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 == "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: raise stov_exceptions.ChannelNotFound() 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._conf, self._url, self._search) elif self._type == "playlist": video_ids = youtubedl_wrapper.get_ids(self._conf, self._url, reverse=True) else: video_ids = youtubedl_wrapper.get_ids(self._conf, 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._conf, self.construct_video_url(video_id)) except subprocess.CalledProcessError: raise stov_exceptions.YoutubeDlCallFailed() 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 @staticmethod def get_quality_parameter(config): """Determines which itag value results from codec and resolution settings and returns it :param config: configuration object :type config: lib_stov.configuration.Conf :return: itag value :rtype: str """ LOGGER.debug(_("Trying to determine the itag value for youtube-dl from" " your quality and codec settings.")) itag_value = 0 if config.values["videocodec"] == "flv": if config.values["maxresolution"] == "240p": itag_value = 5 elif config.values["maxresolution"] == "270p": itag_value = 6 elif config.values["maxresolution"] == "360p": itag_value = 34 elif config.values["maxresolution"] == "480p": itag_value = 35 elif config.values["videocodec"] == "webm": if config.values["maxresolution"] == "360p": itag_value = 43 elif config.values["maxresolution"] == "480p": itag_value = 44 elif config.values["maxresolution"] == "720p": itag_value = 45 elif config.values["maxresolution"] == "1080p": itag_value = 46 elif config.values["videocodec"] == "mp4": if config.values["maxresolution"] == "360p": itag_value = 18 elif config.values["maxresolution"] == "720p": itag_value = 22 elif config.values["maxresolution"] == "1080p": itag_value = 37 elif config.values["maxresolution"] == "3072p": itag_value = 38 if itag_value: LOGGER.debug(_("Found value: %s."), itag_value) return str(itag_value) + "/" + config.values["videocodec"] else: LOGGER.debug(_("Could not determine an itag value " "from the configuration")) return "38"