<s> from __future__ import division <EOL> import base64 <EOL> import hmac <EOL> import re <EOL> import os . path <EOL> from binascii import unhexlify <EOL> from collections import namedtuple <EOL> from copy import deepcopy <EOL> from hashlib import sha256 <EOL> from io import BytesIO <EOL> from math import ceil <EOL> from . flvconcat import FLVTagConcat <EOL> from . segmented import ( SegmentedStreamReader , <EOL> SegmentedStreamWriter , <EOL> SegmentedStreamWorker ) <EOL> from . stream import Stream <EOL> from . wrappers import StreamIOIterWrapper <EOL> from . . cache import Cache <EOL> from . . compat import parse_qsl , urljoin , urlparse , urlunparse , bytes , range <EOL> from . . exceptions import StreamError <EOL> from . . utils import absolute_url , swfdecompress <EOL> from . . packages . flashmedia import F4V , F4VError <EOL> from . . packages . flashmedia . box import Box <EOL> from . . packages . flashmedia . tag import ScriptData , Tag , TAG_TYPE_SCRIPT <EOL> AKAMAIHD_PV_KEY = unhexlify ( <EOL> b"<STR_LIT>" ) <EOL> HDCORE_VERSION = "<STR_LIT>" <EOL> FRAGMENT_URL = "<STR_LIT>" <EOL> Fragment = namedtuple ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> class HDSStreamWriter ( SegmentedStreamWriter ) : <EOL> def __init__ ( self , reader , * args , ** kwargs ) : <EOL> options = reader . stream . session . options <EOL> kwargs [ "<STR_LIT>" ] = options . get ( "<STR_LIT>" ) <EOL> kwargs [ "<STR_LIT>" ] = options . get ( "<STR_LIT>" ) <EOL> kwargs [ "<STR_LIT>" ] = options . get ( "<STR_LIT>" ) <EOL> SegmentedStreamWriter . __init__ ( self , reader , * args , ** kwargs ) <EOL> duration , tags = None , [ ] <EOL> if self . stream . metadata : <EOL> duration = self . stream . metadata . value . get ( "<STR_LIT>" ) <EOL> tags = [ Tag ( TAG_TYPE_SCRIPT , timestamp = <NUM_LIT:0> , <EOL> data = self . stream . metadata ) ] <EOL> self . concater = FLVTagConcat ( tags = tags , <EOL> duration = duration , <EOL> flatten_timestamps = True ) <EOL> def fetch ( self , fragment , retries = None ) : <EOL> if self . closed or not retries : <EOL> return <EOL> try : <EOL> return self . session . http . get ( fragment . url , <EOL> stream = True , <EOL> timeout = self . timeout , <EOL> exception = StreamError , <EOL> ** self . stream . request_params ) <EOL> except StreamError as err : <EOL> self . logger . error ( "<STR_LIT>" , <EOL> fragment . segment , fragment . fragment , err ) <EOL> return self . fetch ( fragment , retries - <NUM_LIT:1> ) <EOL> def write ( self , fragment , res , chunk_size = <NUM_LIT> ) : <EOL> fd = StreamIOIterWrapper ( res . iter_content ( chunk_size ) ) <EOL> self . convert_fragment ( fragment , fd ) <EOL> def convert_fragment ( self , fragment , fd ) : <EOL> mdat = None <EOL> try : <EOL> f4v = F4V ( fd , raw_payload = True ) <EOL> for box in f4v : <EOL> if box . type == "<STR_LIT>" : <EOL> mdat = box . payload . data <EOL> break <EOL> except F4VError as err : <EOL> self . logger . error ( "<STR_LIT>" , <EOL> fragment . segment , fragment . fragment , err ) <EOL> return <EOL> if not mdat : <EOL> self . logger . error ( "<STR_LIT>" , <EOL> fragment . segment , fragment . fragment ) <EOL> return <EOL> try : <EOL> for chunk in self . concater . iter_chunks ( buf = mdat , skip_header = True ) : <EOL> self . reader . buffer . write ( chunk ) <EOL> if self . closed : <EOL> break <EOL> else : <EOL> self . logger . debug ( "<STR_LIT>" , <EOL> fragment . segment , fragment . fragment ) <EOL> except IOError as err : <EOL> if "<STR_LIT>" in str ( err ) : <EOL> self . logger . error ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . close ( ) <EOL> return <EOL> self . logger . error ( "<STR_LIT>" , <EOL> fragment . segment , fragment . fragment , err ) <EOL> class HDSStreamWorker ( SegmentedStreamWorker ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> SegmentedStreamWorker . __init__ ( self , * args , ** kwargs ) <EOL> self . bootstrap = self . stream . bootstrap <EOL> self . current_segment = - <NUM_LIT:1> <EOL> self . current_fragment = - <NUM_LIT:1> <EOL> self . first_fragment = <NUM_LIT:1> <EOL> self . last_fragment = - <NUM_LIT:1> <EOL> self . end_fragment = None <EOL> self . bootstrap_minimal_reload_time = <NUM_LIT> <EOL> self . bootstrap_reload_time = self . bootstrap_minimal_reload_time <EOL> self . invalid_fragments = set ( ) <EOL> self . live_edge = self . session . options . get ( "<STR_LIT>" ) <EOL> self . update_bootstrap ( ) <EOL> def update_bootstrap ( self ) : <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> if isinstance ( self . bootstrap , Box ) : <EOL> bootstrap = self . bootstrap <EOL> else : <EOL> bootstrap = self . fetch_bootstrap ( self . bootstrap ) <EOL> self . live = bootstrap . payload . live <EOL> self . profile = bootstrap . payload . profile <EOL> self . timestamp = bootstrap . payload . current_media_time <EOL> self . identifier = bootstrap . payload . movie_identifier <EOL> self . time_scale = bootstrap . payload . time_scale <EOL> self . segmentruntable = bootstrap . payload . segment_run_table_entries [ <NUM_LIT:0> ] <EOL> self . fragmentruntable = bootstrap . payload . fragment_run_table_entries [ <NUM_LIT:0> ] <EOL> self . first_fragment , last_fragment = self . fragment_count ( ) <EOL> fragment_duration = self . fragment_duration ( last_fragment ) <EOL> if last_fragment != self . last_fragment : <EOL> bootstrap_changed = True <EOL> self . last_fragment = last_fragment <EOL> else : <EOL> bootstrap_changed = False <EOL> if self . current_fragment < <NUM_LIT:0> : <EOL> if self . live : <EOL> current_fragment = last_fragment <EOL> fragment_buffer = int ( ceil ( self . live_edge / fragment_duration ) ) <EOL> current_fragment = max ( self . first_fragment , <EOL> current_fragment - ( fragment_buffer - <NUM_LIT:1> ) ) <EOL> self . logger . debug ( "<STR_LIT>" , <EOL> self . live_edge , fragment_buffer ) <EOL> self . writer . concater . duration = None <EOL> else : <EOL> current_fragment = self . first_fragment <EOL> self . current_fragment = current_fragment <EOL> self . logger . debug ( "<STR_LIT>" , self . timestamp / self . time_scale ) <EOL> self . logger . debug ( "<STR_LIT>" , self . current_segment ) <EOL> self . logger . debug ( "<STR_LIT>" , self . current_fragment ) <EOL> self . logger . debug ( "<STR_LIT>" , self . first_fragment ) <EOL> self . logger . debug ( "<STR_LIT>" , self . last_fragment ) <EOL> self . logger . debug ( "<STR_LIT>" , self . end_fragment ) <EOL> self . bootstrap_reload_time = fragment_duration <EOL> if self . live and not bootstrap_changed : <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> self . bootstrap_reload_time /= <NUM_LIT:2> <EOL> self . bootstrap_reload_time = max ( self . bootstrap_reload_time , <EOL> self . bootstrap_minimal_reload_time ) <EOL> def fetch_bootstrap ( self , url ) : <EOL> res = self . session . http . get ( url , <EOL> exception = StreamError , <EOL> ** self . stream . request_params ) <EOL> return Box . deserialize ( BytesIO ( res . content ) ) <EOL> def fragment_url ( self , segment , fragment ) : <EOL> url = absolute_url ( self . stream . baseurl , self . stream . url ) <EOL> return FRAGMENT_URL . format ( url = url , <EOL> segment = segment , <EOL> fragment = fragment , <EOL> identifier = "<STR_LIT>" , <EOL> quality = "<STR_LIT>" ) <EOL> def fragment_count ( self ) : <EOL> table = self . fragmentruntable . payload . fragment_run_entry_table <EOL> first_fragment , end_fragment = None , None <EOL> for i , fragmentrun in enumerate ( table ) : <EOL> if fragmentrun . discontinuity_indicator is not None : <EOL> if fragmentrun . discontinuity_indicator == <NUM_LIT:0> : <EOL> break <EOL> elif fragmentrun . discontinuity_indicator > <NUM_LIT:0> : <EOL> continue <EOL> if first_fragment is None : <EOL> first_fragment = fragmentrun . first_fragment <EOL> end_fragment = fragmentrun . first_fragment <EOL> fragment_duration = ( fragmentrun . first_fragment_timestamp + <EOL> fragmentrun . fragment_duration ) <EOL> if self . timestamp > fragment_duration : <EOL> offset = ( ( self . timestamp - fragment_duration ) / <EOL> fragmentrun . fragment_duration ) <EOL> end_fragment += int ( offset ) <EOL> if first_fragment is None : <EOL> first_fragment = <NUM_LIT:1> <EOL> if end_fragment is None : <EOL> end_fragment = <NUM_LIT:1> <EOL> return first_fragment , end_fragment <EOL> def fragment_duration ( self , fragment ) : <EOL> fragment_duration = <NUM_LIT:0> <EOL> table = self . fragmentruntable . payload . fragment_run_entry_table <EOL> time_scale = self . fragmentruntable . payload . time_scale <EOL> for i , fragmentrun in enumerate ( table ) : <EOL> if fragmentrun . discontinuity_indicator is not None : <EOL> self . invalid_fragments . add ( fragmentrun . first_fragment ) <EOL> if fragmentrun . discontinuity_indicator == <NUM_LIT:0> : <EOL> if i > <NUM_LIT:0> : <EOL> prev = table [ i - <NUM_LIT:1> ] <EOL> self . end_fragment = prev . first_fragment <EOL> break <EOL> elif fragmentrun . discontinuity_indicator > <NUM_LIT:0> : <EOL> continue <EOL> if fragment >= fragmentrun . first_fragment : <EOL> fragment_duration = fragmentrun . fragment_duration / time_scale <EOL> return fragment_duration <EOL> def segment_from_fragment ( self , fragment ) : <EOL> table = self . segmentruntable . payload . segment_run_entry_table <EOL> for segment , start , end in self . iter_segment_table ( table ) : <EOL> if fragment >= ( start + <NUM_LIT:1> ) and fragment <= ( end + <NUM_LIT:1> ) : <EOL> break <EOL> else : <EOL> segment = <NUM_LIT:1> <EOL> return segment <EOL> def iter_segment_table ( self , table ) : <EOL> if table [ <NUM_LIT:0> ] . first_segment == <NUM_LIT:1> : <EOL> prev_frag = self . first_fragment - <NUM_LIT:1> <EOL> for segmentrun in table : <EOL> start = prev_frag + <NUM_LIT:1> <EOL> end = prev_frag + segmentrun . fragments_per_segment <EOL> yield segmentrun . first_segment , start , end <EOL> prev_frag = end <EOL> else : <EOL> prev_frag = self . last_fragment + <NUM_LIT:1> <EOL> for segmentrun in reversed ( table ) : <EOL> start = prev_frag - segmentrun . fragments_per_segment <EOL> end = prev_frag - <NUM_LIT:1> <EOL> yield segmentrun . first_segment , start , end <EOL> prev_frag = start <EOL> def valid_fragment ( self , fragment ) : <EOL> return fragment not in self . invalid_fragments <EOL> def iter_segments ( self ) : <EOL> while not self . closed : <EOL> fragments = range ( self . current_fragment , self . last_fragment + <NUM_LIT:1> ) <EOL> fragments = filter ( self . valid_fragment , fragments ) <EOL> for fragment in fragments : <EOL> self . current_fragment = fragment + <NUM_LIT:1> <EOL> self . current_segment = self . segment_from_fragment ( fragment ) <EOL> fragment_duration = int ( self . fragment_duration ( fragment ) * <NUM_LIT:1000> ) <EOL> fragment_url = self . fragment_url ( self . current_segment , fragment ) <EOL> fragment = Fragment ( self . current_segment , fragment , <EOL> fragment_duration , fragment_url ) <EOL> self . logger . debug ( "<STR_LIT>" , <EOL> fragment . segment , fragment . fragment ) <EOL> yield fragment <EOL> stream_end = self . end_fragment and fragment . fragment >= self . end_fragment <EOL> if self . closed or stream_end : <EOL> return <EOL> if self . wait ( self . bootstrap_reload_time ) : <EOL> try : <EOL> self . update_bootstrap ( ) <EOL> except StreamError as err : <EOL> self . logger . warning ( "<STR_LIT>" , err ) <EOL> class HDSStreamReader ( SegmentedStreamReader ) : <EOL> __worker__ = HDSStreamWorker <EOL> __writer__ = HDSStreamWriter <EOL> def __init__ ( self , stream , * args , ** kwargs ) : <EOL> SegmentedStreamReader . __init__ ( self , stream , * args , ** kwargs ) <EOL> self . logger = stream . session . logger . new_module ( "<STR_LIT>" ) <EOL> class HDSStream ( Stream ) : <EOL> """<STR_LIT>""" <EOL> __shortname__ = "<STR_LIT>" <EOL> def __init__ ( self , session , baseurl , url , bootstrap , metadata = None , <EOL> timeout = <NUM_LIT> , ** request_params ) : <EOL> Stream . __init__ ( self , session ) <EOL> self . baseurl = baseurl <EOL> self . url = url <EOL> self . bootstrap = bootstrap <EOL> self . metadata = metadata <EOL> self . timeout = timeout <EOL> self . request_params = deepcopy ( request_params ) <EOL> parsed = urlparse ( self . url ) <EOL> if parsed . query : <EOL> params = parse_qsl ( parsed . query ) <EOL> if params : <EOL> if not self . request_params . get ( "<STR_LIT>" ) : <EOL> self . request_params [ "<STR_LIT>" ] = { } <EOL> self . request_params [ "<STR_LIT>" ] . update ( params ) <EOL> self . url = urlunparse ( <EOL> ( parsed . scheme , parsed . netloc , parsed . path , None , None , None ) <EOL> ) <EOL> def __repr__ ( self ) : <EOL> return ( "<STR_LIT>" <EOL> "<STR_LIT>" ) . format ( self . baseurl , <EOL> self . url , <EOL> self . bootstrap , <EOL> self . metadata , <EOL> self . timeout ) <EOL> def __json__ ( self ) : <EOL> if isinstance ( self . bootstrap , Box ) : <EOL> bootstrap = base64 . b64encode ( self . bootstrap . serialize ( ) ) <EOL> else : <EOL> bootstrap = self . bootstrap <EOL> if isinstance ( self . metadata , ScriptData ) : <EOL> metadata = self . metadata . __dict__ <EOL> else : <EOL> metadata = self . metadata <EOL> return dict ( type = HDSStream . shortname ( ) , baseurl = self . baseurl , <EOL> url = self . url , bootstrap = bootstrap , metadata = metadata ) <EOL> def open ( self ) : <EOL> reader = HDSStreamReader ( self ) <EOL> reader . open ( ) <EOL> return reader <EOL> @ classmethod <EOL> def parse_manifest ( cls , session , url , timeout = <NUM_LIT> , pvswf = None , <EOL> ** request_params ) : <EOL> """<STR_LIT>""" <EOL> if not request_params : <EOL> request_params = { } <EOL> request_params [ "<STR_LIT>" ] = request_params . get ( "<STR_LIT>" , { } ) <EOL> request_params [ "<STR_LIT>" ] = request_params . get ( "<STR_LIT>" , { } ) <EOL> request_params . pop ( "<STR_LIT>" , None ) <EOL> request_params . pop ( "<STR_LIT>" , None ) <EOL> request_params . pop ( "<STR_LIT>" , None ) <EOL> request_params . pop ( "<STR_LIT:url>" , None ) <EOL> if "<STR_LIT>" in url : <EOL> request_params [ "<STR_LIT>" ] [ "<STR_LIT>" ] = HDCORE_VERSION <EOL> res = session . http . get ( url , exception = IOError , ** request_params ) <EOL> manifest = session . http . xml ( res , "<STR_LIT>" , ignore_ns = True , <EOL> exception = IOError ) <EOL> parsed = urlparse ( url ) <EOL> baseurl = manifest . findtext ( "<STR_LIT>" ) <EOL> baseheight = manifest . findtext ( "<STR_LIT>" ) <EOL> bootstraps = { } <EOL> streams = { } <EOL> if not baseurl : <EOL> baseurl = urljoin ( url , os . path . dirname ( parsed . path ) ) <EOL> if not baseurl . endswith ( "<STR_LIT:/>" ) : <EOL> baseurl += "<STR_LIT:/>" <EOL> for bootstrap in manifest . findall ( "<STR_LIT>" ) : <EOL> name = bootstrap . attrib . get ( "<STR_LIT:id>" ) or "<STR_LIT>" <EOL> url = bootstrap . attrib . get ( "<STR_LIT:url>" ) <EOL> if url : <EOL> box = absolute_url ( baseurl , url ) <EOL> else : <EOL> data = base64 . b64decode ( bytes ( bootstrap . text , "<STR_LIT:utf8>" ) ) <EOL> box = Box . deserialize ( BytesIO ( data ) ) <EOL> bootstraps [ name ] = box <EOL> pvtoken = manifest . findtext ( "<STR_LIT>" ) <EOL> if pvtoken : <EOL> if not pvswf : <EOL> raise IOError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> params = cls . _pv_params ( session , pvswf , pvtoken , ** request_params ) <EOL> request_params [ "<STR_LIT>" ] . update ( params ) <EOL> for media in manifest . findall ( "<STR_LIT>" ) : <EOL> url = media . attrib . get ( "<STR_LIT:url>" ) <EOL> bootstrapid = media . attrib . get ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> href = media . attrib . get ( "<STR_LIT>" ) <EOL> if url and bootstrapid : <EOL> bootstrap = bootstraps . get ( bootstrapid ) <EOL> if not bootstrap : <EOL> continue <EOL> bitrate = media . attrib . get ( "<STR_LIT>" ) <EOL> streamid = media . attrib . get ( "<STR_LIT>" ) <EOL> height = media . attrib . get ( "<STR_LIT>" ) <EOL> if height : <EOL> quality = height + "<STR_LIT:p>" <EOL> elif bitrate : <EOL> quality = bitrate + "<STR_LIT:k>" <EOL> elif streamid : <EOL> quality = streamid <EOL> elif baseheight : <EOL> quality = baseheight + "<STR_LIT:p>" <EOL> else : <EOL> quality = "<STR_LIT>" <EOL> metadata = media . findtext ( "<STR_LIT>" ) <EOL> if metadata : <EOL> metadata = base64 . b64decode ( bytes ( metadata , "<STR_LIT:utf8>" ) ) <EOL> metadata = ScriptData . deserialize ( BytesIO ( metadata ) ) <EOL> else : <EOL> metadata = None <EOL> stream = HDSStream ( session , baseurl , url , bootstrap , <EOL> metadata = metadata , timeout = timeout , <EOL> ** request_params ) <EOL> streams [ quality ] = stream <EOL> elif href : <EOL> url = absolute_url ( baseurl , href ) <EOL> child_streams = cls . parse_manifest ( session , url , <EOL> timeout = timeout , <EOL> ** request_params ) <EOL> for name , stream in child_streams . items ( ) : <EOL> bitrate = media . attrib . get ( "<STR_LIT>" ) <EOL> if bitrate and not re . match ( "<STR_LIT>" , name ) : <EOL> name = bitrate + "<STR_LIT:k>" <EOL> streams [ name ] = stream <EOL> return streams <EOL> @ classmethod <EOL> def _pv_params ( cls , session , pvswf , pv , ** request_params ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> data , hdntl = pv . split ( "<STR_LIT:;>" ) <EOL> except ValueError : <EOL> data = pv <EOL> hdntl = "<STR_LIT>" <EOL> cache = Cache ( filename = "<STR_LIT>" ) <EOL> key = "<STR_LIT>" + pvswf <EOL> cached = cache . get ( key ) <EOL> request_params = deepcopy ( request_params ) <EOL> headers = request_params . pop ( "<STR_LIT>" , { } ) <EOL> if cached : <EOL> headers [ "<STR_LIT>" ] = cached [ "<STR_LIT>" ] <EOL> swf = session . http . get ( pvswf , headers = headers , ** request_params ) <EOL> if cached and swf . status_code == <NUM_LIT> : <EOL> hash = cached [ "<STR_LIT>" ] <EOL> else : <EOL> hash = sha256 ( ) <EOL> hash . update ( swfdecompress ( swf . content ) ) <EOL> hash = base64 . b64encode ( hash . digest ( ) ) . decode ( "<STR_LIT:ascii>" ) <EOL> modified = swf . headers . get ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> if len ( modified ) < <NUM_LIT> : <EOL> cache . set ( key , dict ( hash = hash , modified = modified ) ) <EOL> msg = "<STR_LIT>" . format ( data , hash ) <EOL> auth = hmac . new ( AKAMAIHD_PV_KEY , msg . encode ( "<STR_LIT:ascii>" ) , sha256 ) <EOL> pvtoken = "<STR_LIT>" . format ( msg , auth . hexdigest ( ) ) <EOL> params = [ ( "<STR_LIT>" , pvtoken ) ] <EOL> params . extend ( parse_qsl ( hdntl , keep_blank_values = True ) ) <EOL> return params </s>
