123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158 |
- #
- # 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)
- except subprocess.CalledProcessError as error_message:
- video_ids = error_message.output
- 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
- video_ids = video_ids.decode(sys.stdout.encoding).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)
- video_description = subprocess.check_output([
- self._conf.values["youtube-dl"],
- "--get-description",
- "https://www.youtube.com/watch?v=%s"
- % video_id], stderr=stderr)
- except subprocess.CalledProcessError:
- raise stov_exceptions.YoutubeDlCallFailed()
- else:
- video_title = video_title.decode(sys.stdout.encoding)
- video_description = video_description.decode(
- sys.stdout.encoding)
- 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
|