PlayStream/sources/YouTubeVideoUrl.py

# -*- coding: UTF-8 -*-
# This video extraction code based on youtube-dl: https://github.com/rg3/youtube-dl

import codecs
import json
import re

from urllib import urlencode
from urllib2 import urlopen, URLError
import sys
import ssl
if "_create_unverified_context" in dir(ssl):
    ssl._create_default_https_context = ssl._create_unverified_context

#from Components.config import config

#from . import sslContext
sslContext = None
if sys.version_info >= (2, 7, 9):
    try:
        import ssl
        sslContext = ssl._create_unverified_context()
    except:
        pass
from jsinterp import JSInterpreter
from swfinterp import SWFInterpreter


PRIORITY_VIDEO_FORMAT = []
maxResolution =  '22'


def createPriorityFormats():
    global PRIORITY_VIDEO_FORMAT,maxResolution
    PRIORITY_VIDEO_FORMAT = []
    use_format = False
    for itag_value in ['38', '37', '96', '22', '95', '120',
                           '35', '94', '18', '93', '5', '92', '132', '17']:
        if itag_value == maxResolution: #config.plugins.YouTube.maxResolution.value:
            use_format = True
        if use_format:
            PRIORITY_VIDEO_FORMAT.append(itag_value)

createPriorityFormats()

IGNORE_VIDEO_FORMAT = [
    '43',  # webm
                '44',  # webm
                '45',  # webm
                '46',  # webm
                '100',  # webm
                '101',  # webm
                '102'  # webm
]


def uppercase_escape(s):
    unicode_escape = codecs.getdecoder('unicode_escape')
    return re.sub(
            r'\\U[0-9a-fA-F]{8}',
                lambda m: unicode_escape(m.group(0))[0],
                s)


def compat_urllib_parse_unquote(string, encoding='utf-8', errors='replace'):
    if string == '':
        return string
    res = string.split('%')
    if len(res) == 1:
        return string
    if encoding is None:
        encoding = 'utf-8'
    if errors is None:
        errors = 'replace'
    # pct_sequence: contiguous sequence of percent-encoded bytes, decoded
    pct_sequence = b''
    string = res[0]
    for item in res[1:]:
        try:
            if not item:
                raise ValueError
            pct_sequence += item[:2].decode('hex')
            rest = item[2:]
            if not rest:
                # This segment was just a single percent-encoded character.
                # May be part of a sequence of code units, so delay decoding.
                # (Stored in pct_sequence).
                continue
        except ValueError:
            rest = '%' + item
        # Encountered non-percent-encoded characters. Flush the current
        # pct_sequence.
        string += pct_sequence.decode(encoding, errors) + rest
        pct_sequence = b''
    if pct_sequence:
        # Flush the final pct_sequence
        string += pct_sequence.decode(encoding, errors)
    return string


def _parse_qsl(qs, keep_blank_values=False, strict_parsing=False,
               encoding='utf-8', errors='replace'):
    qs, _coerce_result = qs, unicode
    pairs = [s2 for s1 in qs.split('&') for s2 in s1.split(';')]
    r = []
    for name_value in pairs:
        if not name_value and not strict_parsing:
            continue
        nv = name_value.split('=', 1)
        if len(nv) != 2:
            if strict_parsing:
                raise ValueError("bad query field: %r" % (name_value,))
            # Handle case of a control-name with no equal sign
            if keep_blank_values:
                nv.append('')
            else:
                continue
        if len(nv[1]) or keep_blank_values:
            name = nv[0].replace('+', ' ')
            name = compat_urllib_parse_unquote(
                            name, encoding=encoding, errors=errors)
            name = _coerce_result(name)
            value = nv[1].replace('+', ' ')
            value = compat_urllib_parse_unquote(
                            value, encoding=encoding, errors=errors)
            value = _coerce_result(value)
            r.append((name, value))
    return r


def compat_parse_qs(qs, keep_blank_values=False, strict_parsing=False,
                    encoding='utf-8', errors='replace'):
    parsed_result = {}
    pairs = _parse_qsl(qs, keep_blank_values, strict_parsing,
                           encoding=encoding, errors=errors)
    for name, value in pairs:
        if name in parsed_result:
            parsed_result[name].append(value)
        else:
            parsed_result[name] = [value]
    return parsed_result


class YouTubeVideoUrl():

    def _download_webpage(self, url):
        """ Returns a tuple (page content as string, URL handle) """
        try:
            if sslContext:
                urlh = urlopen(url, context = sslContext)
            else:
                urlh = urlopen(url)
        except URLError, e:
            #raise Exception(e.reason)
            return ""
        return urlh.read()

    def _search_regex(self, pattern, string):
        """
        Perform a regex search on the given string, using a single or a list of
        patterns returning the first matching group.
        """
        mobj = re.search(pattern, string, 0)
        if mobj:
            # return the first matching group
            return next(g for g in mobj.groups() if g is not None)
        else:
            raise Exception('Unable extract pattern from string!')

    def _decrypt_signature(self, s, player_url):
        """Turn the encrypted s field into a working signature"""

        if player_url is None:
            raise Exception('Cannot decrypt signature without player_url!')

        if player_url[:2] == '//':
            player_url = 'https:' + player_url
        try:
            func = self._extract_signature_function(player_url)
            return func(s)
        except:
            raise Exception('Signature extraction failed!')

    def _extract_signature_function(self, player_url):
        id_m = re.match(
                    r'.*?-(?P<id>[a-zA-Z0-9_-]+)(?:/watch_as3|/html5player(?:-new)?|/base)?\.(?P<ext>[a-z]+)$',
                        player_url)
        if not id_m:
            raise Exception('Cannot identify player %r!' % player_url)
        player_type = id_m.group('ext')
        code = self._download_webpage(player_url)
        if player_type == 'js':
            return self._parse_sig_js(code)
        elif player_type == 'swf':
            return self._parse_sig_swf(code)
        else:
            raise Exception('Invalid player type %r!' % player_type)

    def _parse_sig_js(self, jscode):
        funcname = self._search_regex(r'\.sig\|\|([a-zA-Z0-9$]+)\(', jscode)
        jsi = JSInterpreter(jscode)
        initial_function = jsi.extract_function(funcname)
        return lambda s: initial_function([s])

    def _parse_sig_swf(self, file_contents):
        swfi = SWFInterpreter(file_contents)
        TARGET_CLASSNAME = 'SignatureDecipher'
        searched_class = swfi.extract_class(TARGET_CLASSNAME)
        initial_function = swfi.extract_function(searched_class, 'decipher')
        return lambda s: initial_function([s])

    def _extract_from_m3u8(self, manifest_url):
        url_map = {}

        def _get_urls(_manifest):
            lines = _manifest.split('\n')
            urls = filter(lambda l: l and not l.startswith('#'), lines)
            return urls

        manifest = self._download_webpage(manifest_url)
        formats_urls = _get_urls(manifest)
        for format_url in formats_urls:
            itag = self._search_regex(r'itag/(\d+?)/', format_url)
            url_map[itag] = format_url
        return url_map

    def _get_ytplayer_config(self, webpage):
        # User data may contain arbitrary character sequences that may affect
        # JSON extraction with regex, e.g. when '};' is contained the second
        # regex won't capture the whole JSON. Yet working around by trying more
        # concrete regex first keeping in mind proper quoted string handling
        # to be implemented in future that will replace this workaround (see
        # https://github.com/rg3/youtube-dl/issues/7468,
        # https://github.com/rg3/youtube-dl/pull/7599)
        patterns = [
                    r';ytplayer\.config\s*=\s*({.+?});ytplayer',
                        r';ytplayer\.config\s*=\s*({.+?});',
                ]
        for pattern in patterns:
            config = self._search_regex(pattern, webpage)
            if config:
                return json.loads(uppercase_escape(config))

    def extract(self, video_id):
        url = 'https://www.youtube.com/watch?v=%s&gl=US&hl=en&has_verified=1&bpctr=9999999999' % video_id

        # Get video webpage
        video_webpage = self._download_webpage(url)
        if not video_webpage:
            #raise Exception('Video webpage not found!')
            return ""

        # Attempt to extract SWF player URL
        mobj = re.search(r'swfConfig.*?"(https?:\\/\\/.*?watch.*?-.*?\.swf)"', video_webpage)
        if mobj is not None:
            player_url = re.sub(r'\\(.)', r'\1', mobj.group(1))
        else:
            player_url = None

        # Get video info
        embed_webpage = None
        if re.search(r'player-age-gate-content">', video_webpage) is not None:
            age_gate = True
            # We simulate the access to the video from www.youtube.com/v/{video_id}
            # this can be viewed without login into Youtube
            url = 'https://www.youtube.com/embed/%s' % video_id
            embed_webpage = self._download_webpage(url)
            data = urlencode({
                            'video_id': video_id,
                                'eurl': 'https://youtube.googleapis.com/v/' + video_id,
                                'sts': self._search_regex(r'"sts"\s*:\s*(\d+)', embed_webpage),
                        })
            video_info_url = 'https://www.youtube.com/get_video_info?' + data
            video_info_webpage = self._download_webpage(video_info_url)
            video_info = compat_parse_qs(video_info_webpage)
        else:
            age_gate = False
            video_info = None
            # Try looking directly into the video webpage
            ytplayer_config = self._get_ytplayer_config(video_webpage)
            if ytplayer_config:
                args = ytplayer_config['args']
                if args.get('url_encoded_fmt_stream_map'):
                    # Convert to the same format returned by compat_parse_qs
                    video_info = dict((k, [v]) for k, v in args.items())

            if not video_info:
                # We also try looking in get_video_info since it may contain different dashmpd
                # URL that points to a DASH manifest with possibly different itag set (some itags
                # are missing from DASH manifest pointed by webpage's dashmpd, some - from DASH
                # manifest pointed by get_video_info's dashmpd).
                # The general idea is to take a union of itags of both DASH manifests (for example
                # video with such 'manifest behavior' see https://github.com/rg3/youtube-dl/issues/6093)
                for el_type in ['&el=info', '&el=embedded', '&el=detailpage', '&el=vevo', '']:
                    video_info_url = (
                                            'https://www.youtube.com/get_video_info?&video_id=%s%s&ps=default&eurl=&gl=US&hl=en'
                                                % (video_id, el_type))
                    video_info_webpage = self._download_webpage(video_info_url)
                    video_info = compat_parse_qs(video_info_webpage)
                    if 'token' in video_info:
                        break
        if 'token' not in video_info:
            if 'reason' in video_info:
                print '[YouTubeVideoUrl] %s' % video_info['reason'][0]
            else:
                print '[YouTubeVideoUrl] "token" parameter not in video info for unknown reason'

        # Start extracting information
        if 'conn' in video_info and video_info['conn'][0][:4] == 'rtmp':
            url = video_info['conn'][0]
        elif len(video_info.get('url_encoded_fmt_stream_map', [''])[0]) >= 1 or \
                     len(video_info.get('adaptive_fmts', [''])[0]) >= 1:
            encoded_url_map = video_info.get('url_encoded_fmt_stream_map', [''])[0] + \
                            ',' + video_info.get('adaptive_fmts', [''])[0]
            if 'rtmpe%3Dyes' in encoded_url_map:
                raise Exception('rtmpe downloads are not supported, see https://github.com/rg3/youtube-dl/issues/343')

            # Find the best format from our format priority map
            encoded_url_map = encoded_url_map.split(',')
            url_map_str = None
            # If format changed in config, recreate priority list
            if PRIORITY_VIDEO_FORMAT[0] != maxResolution: #config.plugins.YouTube.maxResolution.value:
                createPriorityFormats()
            for our_format in PRIORITY_VIDEO_FORMAT:
                our_format = 'itag=' + our_format
                for encoded_url in encoded_url_map:
                    if our_format in encoded_url and 'url=' in encoded_url:
                        url_map_str = encoded_url
                        break
                if url_map_str:
                    break
            # If anything not found, used first in the list if it not in ignore map
            if not url_map_str:
                for encoded_url in encoded_url_map:
                    if 'url=' in encoded_url:
                        url_map_str = encoded_url
                        for ignore_format in IGNORE_VIDEO_FORMAT:
                            ignore_format = 'itag=' + ignore_format
                            if ignore_format in encoded_url:
                                url_map_str = None
                                break
                    if url_map_str:
                        break
            if not url_map_str:
                url_map_str = encoded_url_map[0]

            url_data = compat_parse_qs(url_map_str)
            url = url_data['url'][0]
            if 'sig' in url_data:
                url += '&signature=' + url_data['sig'][0]
            elif 's' in url_data:
                encrypted_sig = url_data['s'][0]
                ASSETS_RE = r'"assets":.+?"js":\s*("[^"]+")'

                jsplayer_url_json = self._search_regex(ASSETS_RE,
                                                                       embed_webpage if age_gate else video_webpage)
                if not jsplayer_url_json and not age_gate:
                    # We need the embed website after all
                    if embed_webpage is None:
                        embed_url = 'https://www.youtube.com/embed/%s' % video_id
                        embed_webpage = self._download_webpage(embed_url)
                    jsplayer_url_json = self._search_regex(ASSETS_RE, embed_webpage)

                player_url = json.loads(jsplayer_url_json)
                if player_url is None:
                    player_url_json = self._search_regex(
                                            r'ytplayer\.config.*?"url"\s*:\s*("[^"]+")',
                                                video_webpage)
                    player_url = json.loads(player_url_json)

                signature = self._decrypt_signature(encrypted_sig, player_url)
                url += '&signature=' + signature
            if 'ratebypass' not in url:
                url += '&ratebypass=yes'
        elif video_info.get('hlsvp'):
            url = None
            manifest_url = video_info['hlsvp'][0]
            url_map = self._extract_from_m3u8(manifest_url)

            # Find the best format from our format priority map
            for our_format in PRIORITY_VIDEO_FORMAT:
                if url_map.get(our_format):
                    url = url_map[our_format]
                    break
            # If anything not found, used first in the list if it not in ignore map
            if not url:
                for url_map_key in url_map.keys():
                    if url_map_key not in IGNORE_VIDEO_FORMAT:
                        url = url_map[url_map_key]
                        break
            if not url:
                url = url_map.values()[0]
        else:
            #raise Exception('No supported formats found in video info!')
            return ""

        return str(url)

if __name__ == "__main__":

    #yt = YouTubeVideoUrl()
    if len(sys.argv)>1:
        video_id= sys.argv[1]
    else:
        video_id = "2rlTF6HiMGg"
    e = YouTubeVideoUrl().extract(video_id)
    print e