<s> import re <EOL> from collections import namedtuple <EOL> from functools import partial <EOL> from random import randint <EOL> from time import sleep <EOL> from livestreamer . compat import urlparse , urljoin , range <EOL> from livestreamer . exceptions import StreamError , PluginError , NoStreamsError <EOL> from livestreamer . plugin import Plugin , PluginOptions <EOL> from livestreamer . plugin . api import http , validate <EOL> from livestreamer . stream import RTMPStream , HLSStream , HTTPStream , Stream <EOL> from livestreamer . stream . flvconcat import FLVTagConcat <EOL> from livestreamer . stream . segmented import ( <EOL> SegmentedStreamReader , SegmentedStreamWriter , SegmentedStreamWorker <EOL> ) <EOL> try : <EOL> import librtmp <EOL> HAS_LIBRTMP = True <EOL> except ImportError : <EOL> HAS_LIBRTMP = False <EOL> _url_re = re . compile ( """<STR_LIT>""" , re . VERBOSE ) <EOL> _channel_id_re = re . compile ( "<STR_LIT>" ) <EOL> HLS_PLAYLIST_URL = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> RECORDED_URL = "<STR_LIT>" <EOL> RTMP_URL = "<STR_LIT>" <EOL> SWF_URL = "<STR_LIT>" <EOL> _module_info_schema = validate . Schema ( <EOL> list , <EOL> validate . length ( <NUM_LIT:1> ) , <EOL> validate . get ( <NUM_LIT:0> ) , <EOL> dict <EOL> ) <EOL> _amf3_array = validate . Schema ( <EOL> validate . any ( <EOL> validate . all ( <EOL> { int : object } , <EOL> validate . transform ( lambda a : list ( a . values ( ) ) ) , <EOL> ) , <EOL> list <EOL> ) <EOL> ) <EOL> _recorded_schema = validate . Schema ( { <EOL> validate . optional ( "<STR_LIT>" ) : validate . all ( <EOL> _amf3_array , <EOL> [ { <EOL> "<STR_LIT:name>" : validate . text , <EOL> "<STR_LIT>" : validate . all ( <EOL> _amf3_array , <EOL> [ { <EOL> "<STR_LIT>" : validate . text , <EOL> "<STR_LIT>" : float , <EOL> } ] , <EOL> ) , <EOL> validate . optional ( "<STR_LIT:url>" ) : validate . text , <EOL> } ] <EOL> ) <EOL> } ) <EOL> _stream_schema = validate . Schema ( <EOL> validate . any ( { <EOL> "<STR_LIT:name>" : validate . text , <EOL> "<STR_LIT:url>" : validate . text , <EOL> "<STR_LIT>" : validate . all ( <EOL> _amf3_array , <EOL> [ { <EOL> "<STR_LIT>" : validate . any ( int , float ) , <EOL> "<STR_LIT>" : { validate . text : validate . text } , <EOL> "<STR_LIT>" : validate . any ( int , float ) , <EOL> "<STR_LIT>" : validate . any ( int , float ) , <EOL> "<STR_LIT>" : validate . any ( int , float ) , <EOL> "<STR_LIT>" : validate . text , <EOL> validate . optional ( "<STR_LIT>" ) : validate . any ( int , float ) , <EOL> validate . optional ( "<STR_LIT>" ) : validate . any ( int , float ) , <EOL> validate . optional ( "<STR_LIT:description>" ) : validate . text , <EOL> validate . optional ( "<STR_LIT>" ) : bool <EOL> } ] , <EOL> ) <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : validate . text , <EOL> "<STR_LIT>" : validate . text <EOL> } ) <EOL> ) <EOL> _channel_schema = validate . Schema ( { <EOL> validate . optional ( "<STR_LIT>" ) : validate . any ( <EOL> validate . all ( <EOL> _amf3_array , <EOL> [ _stream_schema ] , <EOL> ) , <EOL> "<STR_LIT>" <EOL> ) <EOL> } ) <EOL> Chunk = namedtuple ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> if HAS_LIBRTMP : <EOL> from io import BytesIO <EOL> from time import time <EOL> from librtmp . rtmp import RTMPTimeoutError , PACKET_TYPE_INVOKE <EOL> from livestreamer . packages . flashmedia . types import AMF0Value <EOL> def decode_amf ( body ) : <EOL> def generator ( ) : <EOL> fd = BytesIO ( body ) <EOL> while True : <EOL> try : <EOL> yield AMF0Value . read ( fd ) <EOL> except IOError : <EOL> break <EOL> return list ( generator ( ) ) <EOL> class FlashmediaRTMP ( librtmp . RTMP ) : <EOL> """<STR_LIT>""" <EOL> def process_packets ( self , transaction_id = None , invoked_method = None , <EOL> timeout = None ) : <EOL> start = time ( ) <EOL> while self . connected and transaction_id not in self . _invoke_results : <EOL> if timeout and ( time ( ) - start ) >= timeout : <EOL> raise RTMPTimeoutError ( "<STR_LIT>" ) <EOL> packet = self . read_packet ( ) <EOL> if packet . type == PACKET_TYPE_INVOKE : <EOL> try : <EOL> decoded = decode_amf ( packet . body ) <EOL> except IOError : <EOL> continue <EOL> try : <EOL> method , transaction_id_ , obj = decoded [ : <NUM_LIT:3> ] <EOL> args = decoded [ <NUM_LIT:3> : ] <EOL> except ValueError : <EOL> continue <EOL> if method == "<STR_LIT>" : <EOL> if len ( args ) > <NUM_LIT:0> : <EOL> result = args [ <NUM_LIT:0> ] <EOL> else : <EOL> result = None <EOL> self . _invoke_results [ transaction_id_ ] = result <EOL> else : <EOL> handler = self . _invoke_handlers . get ( method ) <EOL> if handler : <EOL> res = handler ( * args ) <EOL> if res is not None : <EOL> self . call ( "<STR_LIT>" , res , <EOL> transaction_id = transaction_id_ ) <EOL> if method == invoked_method : <EOL> self . _invoke_args [ invoked_method ] = args <EOL> break <EOL> if transaction_id_ == <NUM_LIT:1.0> : <EOL> self . _connect_result = packet <EOL> else : <EOL> self . handle_packet ( packet ) <EOL> else : <EOL> self . handle_packet ( packet ) <EOL> if transaction_id : <EOL> result = self . _invoke_results . pop ( transaction_id , None ) <EOL> return result <EOL> if invoked_method : <EOL> args = self . _invoke_args . pop ( invoked_method , None ) <EOL> return args <EOL> def create_ums_connection ( app , media_id , page_url , password , <EOL> exception = PluginError ) : <EOL> url = RTMP_URL . format ( randint ( <NUM_LIT:0> , <NUM_LIT> ) , media_id ) <EOL> params = { <EOL> "<STR_LIT>" : app , <EOL> "<STR_LIT>" : str ( media_id ) , <EOL> "<STR_LIT:password>" : password <EOL> } <EOL> conn = FlashmediaRTMP ( url , <EOL> swfurl = SWF_URL , <EOL> pageurl = page_url , <EOL> connect_data = params ) <EOL> try : <EOL> conn . connect ( ) <EOL> except librtmp . RTMPError : <EOL> raise exception ( "<STR_LIT>" ) <EOL> return conn <EOL> class UHSStreamWriter ( SegmentedStreamWriter ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> SegmentedStreamWriter . __init__ ( self , * args , ** kwargs ) <EOL> self . concater = FLVTagConcat ( flatten_timestamps = True , <EOL> sync_headers = True ) <EOL> def fetch ( self , chunk , retries = None ) : <EOL> if not retries or self . closed : <EOL> return <EOL> try : <EOL> params = { } <EOL> if chunk . offset : <EOL> params [ "<STR_LIT:start>" ] = chunk . offset <EOL> return http . get ( chunk . url , <EOL> timeout = self . timeout , <EOL> params = params , <EOL> exception = StreamError ) <EOL> except StreamError as err : <EOL> self . logger . error ( "<STR_LIT>" , chunk . num , err ) <EOL> return self . fetch ( chunk , retries - <NUM_LIT:1> ) <EOL> def write ( self , chunk , res , chunk_size = <NUM_LIT> ) : <EOL> try : <EOL> for data in self . concater . iter_chunks ( buf = res . content , <EOL> skip_header = not chunk . offset ) : <EOL> self . reader . buffer . write ( data ) <EOL> if self . closed : <EOL> break <EOL> else : <EOL> self . logger . debug ( "<STR_LIT>" , chunk . num ) <EOL> except IOError as err : <EOL> self . logger . error ( "<STR_LIT>" , chunk . num , err ) <EOL> class UHSStreamWorker ( SegmentedStreamWorker ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> SegmentedStreamWorker . __init__ ( self , * args , ** kwargs ) <EOL> self . chunk_ranges = { } <EOL> self . chunk_id = None <EOL> self . chunk_id_max = None <EOL> self . chunks = [ ] <EOL> self . filename_format = "<STR_LIT>" <EOL> self . module_info_reload_time = <NUM_LIT:2> <EOL> self . process_module_info ( ) <EOL> def fetch_module_info ( self ) : <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> conn = create_ums_connection ( "<STR_LIT>" , <EOL> self . stream . channel_id , <EOL> self . stream . page_url , <EOL> self . stream . password , <EOL> exception = StreamError ) <EOL> try : <EOL> result = conn . process_packets ( invoked_method = "<STR_LIT>" , <EOL> timeout = <NUM_LIT:10> ) <EOL> except ( IOError , librtmp . RTMPError ) as err : <EOL> raise StreamError ( "<STR_LIT>" . format ( err ) ) <EOL> finally : <EOL> conn . close ( ) <EOL> result = _module_info_schema . validate ( result ) <EOL> return _channel_schema . validate ( result , "<STR_LIT>" ) <EOL> def process_module_info ( self ) : <EOL> if self . closed : <EOL> return <EOL> try : <EOL> result = self . fetch_module_info ( ) <EOL> except PluginError as err : <EOL> self . logger . error ( "<STR_LIT>" , err ) <EOL> return <EOL> providers = result . get ( "<STR_LIT>" ) <EOL> if not providers or providers == "<STR_LIT>" : <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> self . close ( ) <EOL> return <EOL> for provider in providers : <EOL> if provider . get ( "<STR_LIT:name>" ) == self . stream . provider : <EOL> break <EOL> else : <EOL> return <EOL> try : <EOL> stream = provider [ "<STR_LIT>" ] [ self . stream . stream_index ] <EOL> except IndexError : <EOL> self . logger . error ( "<STR_LIT>" ) <EOL> return <EOL> filename_format = stream [ "<STR_LIT>" ] . replace ( "<STR_LIT:%>" , "<STR_LIT:%s>" ) <EOL> filename_format = urljoin ( provider [ "<STR_LIT:url>" ] , filename_format ) <EOL> self . filename_format = filename_format <EOL> self . update_chunk_info ( stream ) <EOL> def update_chunk_info ( self , result ) : <EOL> chunk_range = result [ "<STR_LIT>" ] <EOL> if not chunk_range : <EOL> return <EOL> chunk_id = int ( result [ "<STR_LIT>" ] ) <EOL> chunk_offset = int ( result [ "<STR_LIT>" ] ) <EOL> chunk_range = dict ( map ( partial ( map , int ) , chunk_range . items ( ) ) ) <EOL> self . chunk_ranges . update ( chunk_range ) <EOL> self . chunk_id_min = sorted ( chunk_range ) [ <NUM_LIT:0> ] <EOL> self . chunk_id_max = int ( result [ "<STR_LIT>" ] ) <EOL> self . chunks = [ Chunk ( i , self . format_chunk_url ( i ) , <EOL> not self . chunk_id and i == chunk_id and chunk_offset ) <EOL> for i in range ( self . chunk_id_min , self . chunk_id_max + <NUM_LIT:1> ) ] <EOL> if self . chunk_id is None and self . chunks : <EOL> self . chunk_id = chunk_id <EOL> def format_chunk_url ( self , chunk_id ) : <EOL> chunk_hash = "<STR_LIT>" <EOL> for chunk_start in sorted ( self . chunk_ranges ) : <EOL> if chunk_id >= chunk_start : <EOL> chunk_hash = self . chunk_ranges [ chunk_start ] <EOL> return self . filename_format % ( chunk_id , chunk_hash ) <EOL> def valid_chunk ( self , chunk ) : <EOL> return self . chunk_id and chunk . num >= self . chunk_id <EOL> def iter_segments ( self ) : <EOL> while not self . closed : <EOL> for chunk in filter ( self . valid_chunk , self . chunks ) : <EOL> self . logger . debug ( "<STR_LIT>" , chunk . num ) <EOL> yield chunk <EOL> if self . closed : <EOL> return <EOL> self . chunk_id = chunk . num + <NUM_LIT:1> <EOL> if self . wait ( self . module_info_reload_time ) : <EOL> try : <EOL> self . process_module_info ( ) <EOL> except StreamError as err : <EOL> self . logger . warning ( "<STR_LIT>" , err ) <EOL> class UHSStreamReader ( SegmentedStreamReader ) : <EOL> __worker__ = UHSStreamWorker <EOL> __writer__ = UHSStreamWriter <EOL> def __init__ ( self , stream , * args , ** kwargs ) : <EOL> self . logger = stream . session . logger . new_module ( "<STR_LIT>" ) <EOL> SegmentedStreamReader . __init__ ( self , stream , * args , ** kwargs ) <EOL> class UHSStream ( Stream ) : <EOL> __shortname__ = "<STR_LIT>" <EOL> def __init__ ( self , session , channel_id , page_url , provider , <EOL> stream_index , password = "<STR_LIT>" ) : <EOL> Stream . __init__ ( self , session ) <EOL> self . channel_id = channel_id <EOL> self . page_url = page_url <EOL> self . provider = provider <EOL> self . stream_index = stream_index <EOL> self . password = password <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" . format ( <EOL> self . channel_id , self . page_url , self . provider , <EOL> self . stream_index , self . password <EOL> ) <EOL> def __json__ ( self ) : <EOL> json = Stream . __json__ ( self ) <EOL> json . update ( { <EOL> "<STR_LIT>" : self . channel_id , <EOL> "<STR_LIT>" : self . page_url , <EOL> "<STR_LIT>" : self . provider , <EOL> "<STR_LIT>" : self . stream_index , <EOL> "<STR_LIT:password>" : self . password <EOL> } ) <EOL> return json <EOL> def open ( self ) : <EOL> reader = UHSStreamReader ( self ) <EOL> reader . open ( ) <EOL> return reader <EOL> class UStreamTV ( Plugin ) : <EOL> options = PluginOptions ( { <EOL> "<STR_LIT:password>" : "<STR_LIT>" <EOL> } ) <EOL> @ classmethod <EOL> def can_handle_url ( cls , url ) : <EOL> return _url_re . match ( url ) <EOL> @ classmethod <EOL> def stream_weight ( cls , stream ) : <EOL> match = re . match ( "<STR_LIT>" , stream ) <EOL> if match : <EOL> weight , group = Plugin . stream_weight ( match . group ( <NUM_LIT:1> ) ) <EOL> weight -= <NUM_LIT:1> <EOL> group = "<STR_LIT>" <EOL> elif stream == "<STR_LIT>" : <EOL> weight , group = <NUM_LIT> , "<STR_LIT>" <EOL> else : <EOL> weight , group = Plugin . stream_weight ( stream ) <EOL> return weight , group <EOL> def _get_channel_id ( self ) : <EOL> res = http . get ( self . url ) <EOL> match = _channel_id_re . search ( res . text ) <EOL> if match : <EOL> return int ( match . group ( <NUM_LIT:1> ) ) <EOL> def _get_hls_streams ( self , channel_id , wait_for_transcode = False ) : <EOL> attempts = wait_for_transcode and <NUM_LIT:10> or <NUM_LIT:1> <EOL> playlist_url = HLS_PLAYLIST_URL . format ( channel_id ) <EOL> streams = { } <EOL> while attempts and not streams : <EOL> try : <EOL> streams = HLSStream . parse_variant_playlist ( self . session , <EOL> playlist_url , <EOL> nameprefix = "<STR_LIT>" ) <EOL> except IOError : <EOL> break <EOL> attempts -= <NUM_LIT:1> <EOL> sleep ( <NUM_LIT:3> ) <EOL> return streams <EOL> def _create_rtmp_stream ( self , cdn , stream_name ) : <EOL> parsed = urlparse ( cdn ) <EOL> params = { <EOL> "<STR_LIT>" : cdn , <EOL> "<STR_LIT>" : parsed . path [ <NUM_LIT:1> : ] , <EOL> "<STR_LIT>" : stream_name , <EOL> "<STR_LIT>" : self . url , <EOL> "<STR_LIT>" : SWF_URL , <EOL> "<STR_LIT>" : True <EOL> } <EOL> return RTMPStream ( self . session , params ) <EOL> def _get_module_info ( self , app , media_id , password = "<STR_LIT>" , schema = None ) : <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> conn = create_ums_connection ( app , media_id , self . url , password ) <EOL> attempts = <NUM_LIT:3> <EOL> while conn . connected and attempts : <EOL> try : <EOL> result = conn . process_packets ( invoked_method = "<STR_LIT>" , <EOL> timeout = <NUM_LIT:10> ) <EOL> except ( IOError , librtmp . RTMPError ) as err : <EOL> raise PluginError ( "<STR_LIT>" . format ( err ) ) <EOL> try : <EOL> result = _module_info_schema . validate ( result ) <EOL> break <EOL> except PluginError : <EOL> attempts -= <NUM_LIT:1> <EOL> conn . close ( ) <EOL> if schema : <EOL> result = schema . validate ( result ) <EOL> return result <EOL> def _get_desktop_streams ( self , channel_id ) : <EOL> password = self . options . get ( "<STR_LIT:password>" ) <EOL> channel = self . _get_module_info ( "<STR_LIT>" , channel_id , password , <EOL> schema = _channel_schema ) <EOL> if not isinstance ( channel . get ( "<STR_LIT>" ) , list ) : <EOL> raise NoStreamsError ( self . url ) <EOL> streams = { } <EOL> for provider in channel [ "<STR_LIT>" ] : <EOL> if provider [ "<STR_LIT:name>" ] == u"<STR_LIT>" : <EOL> continue <EOL> provider_url = provider [ "<STR_LIT:url>" ] <EOL> provider_name = provider [ "<STR_LIT:name>" ] <EOL> for stream_index , stream_info in enumerate ( provider [ "<STR_LIT>" ] ) : <EOL> stream = None <EOL> stream_height = int ( stream_info . get ( "<STR_LIT>" , <NUM_LIT:0> ) ) <EOL> stream_name = stream_info . get ( "<STR_LIT:description>" ) <EOL> if not stream_name : <EOL> if stream_height > <NUM_LIT:0> : <EOL> if not stream_info . get ( "<STR_LIT>" ) : <EOL> stream_name = "<STR_LIT>" . format ( stream_height ) <EOL> else : <EOL> stream_name = "<STR_LIT>" . format ( stream_height ) <EOL> else : <EOL> stream_name = "<STR_LIT>" <EOL> if stream_name in streams : <EOL> provider_name_clean = provider_name . replace ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> stream_name += "<STR_LIT>" . format ( provider_name_clean ) <EOL> if provider_name . startswith ( "<STR_LIT>" ) : <EOL> stream = UHSStream ( self . session , channel_id , <EOL> self . url , provider_name , <EOL> stream_index , password ) <EOL> elif provider_url . startswith ( "<STR_LIT>" ) : <EOL> playpath = stream_info [ "<STR_LIT>" ] <EOL> stream = self . _create_rtmp_stream ( provider_url , <EOL> playpath ) <EOL> if stream : <EOL> streams [ stream_name ] = stream <EOL> return streams <EOL> def _get_live_streams ( self , channel_id ) : <EOL> has_desktop_streams = False <EOL> if HAS_LIBRTMP : <EOL> try : <EOL> streams = self . _get_desktop_streams ( channel_id ) <EOL> for stream in streams . items ( ) : <EOL> has_desktop_streams = True <EOL> yield stream <EOL> except PluginError as err : <EOL> self . logger . error ( "<STR_LIT>" , err ) <EOL> except NoStreamsError : <EOL> pass <EOL> else : <EOL> self . logger . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> try : <EOL> streams = self . _get_hls_streams ( channel_id , <EOL> wait_for_transcode = not has_desktop_streams ) <EOL> for stream in streams . items ( ) : <EOL> yield stream <EOL> except PluginError as err : <EOL> self . logger . error ( "<STR_LIT>" , err ) <EOL> except NoStreamsError : <EOL> pass <EOL> def _get_recorded_streams ( self , video_id ) : <EOL> if HAS_LIBRTMP : <EOL> recording = self . _get_module_info ( "<STR_LIT>" , video_id , <EOL> schema = _recorded_schema ) <EOL> if not isinstance ( recording . get ( "<STR_LIT>" ) , list ) : <EOL> return <EOL> for provider in recording [ "<STR_LIT>" ] : <EOL> base_url = provider . get ( "<STR_LIT:url>" ) <EOL> for stream_info in provider [ "<STR_LIT>" ] : <EOL> bitrate = int ( stream_info . get ( "<STR_LIT>" , <NUM_LIT:0> ) ) <EOL> stream_name = ( bitrate > <NUM_LIT:0> and "<STR_LIT>" . format ( bitrate ) or <EOL> "<STR_LIT>" ) <EOL> url = stream_info [ "<STR_LIT>" ] <EOL> if base_url : <EOL> url = base_url + url <EOL> if url . startswith ( "<STR_LIT:http>" ) : <EOL> yield stream_name , HTTPStream ( self . session , url ) <EOL> elif url . startswith ( "<STR_LIT>" ) : <EOL> params = dict ( rtmp = url , pageUrl = self . url ) <EOL> yield stream_name , RTMPStream ( self . session , params ) <EOL> else : <EOL> self . logger . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> url = RECORDED_URL . format ( video_id ) <EOL> random_hash = "<STR_LIT>" . format ( randint ( <NUM_LIT:0> , <NUM_LIT:255> ) , <EOL> randint ( <NUM_LIT:0> , <NUM_LIT:255> ) ) <EOL> params = dict ( hash = random_hash ) <EOL> stream = HTTPStream ( self . session , url , params = params ) <EOL> yield "<STR_LIT>" , stream <EOL> def _get_streams ( self ) : <EOL> match = _url_re . match ( self . url ) <EOL> video_id = match . group ( "<STR_LIT>" ) <EOL> if video_id : <EOL> return self . _get_recorded_streams ( video_id ) <EOL> channel_id = match . group ( "<STR_LIT>" ) or self . _get_channel_id ( ) <EOL> if channel_id : <EOL> return self . _get_live_streams ( channel_id ) <EOL> __plugin__ = UStreamTV </s>
