<s> """<STR_LIT>""" <EOL> import re <EOL> import sys <EOL> import os <EOL> import subprocess <EOL> import time <EOL> import requests <EOL> def _regex_match_get ( pattern , text , position = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> match = re . search ( pattern , text ) <EOL> if match is not None : <EOL> return match . group ( position ) <EOL> else : <EOL> return None <EOL> def _convert_display_size ( size ) : <EOL> """<STR_LIT>""" <EOL> unit_gigabyte = <NUM_LIT> * <NUM_LIT> * <NUM_LIT> <EOL> unit_megabyte = <NUM_LIT> * <NUM_LIT> <EOL> unit_kilobyte = <NUM_LIT> <EOL> if size > unit_gigabyte : <EOL> return "<STR_LIT>" . format ( <EOL> size / unit_gigabyte ) <EOL> elif size > unit_megabyte : <EOL> return "<STR_LIT>" . format ( <EOL> size / unit_megabyte ) <EOL> elif size > unit_kilobyte : <EOL> return "<STR_LIT>" . format ( <EOL> size / unit_kilobyte ) <EOL> return "<STR_LIT>" . format ( size ) <EOL> class ExtractionError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class DownloadError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class VideoDownloader ( object ) : <EOL> """<STR_LIT>""" <EOL> _CLEAR_STDOUT = '<STR_LIT>' <EOL> def __init__ ( self , opts ) : <EOL> self . info_extract = VideoInfoExtractor ( ) <EOL> self . opts = opts <EOL> def _to_stderr ( self , message ) : <EOL> """<STR_LIT>""" <EOL> sys . stderr . write ( message ) <EOL> def _to_stdout ( self , message , skip_eol = False ) : <EOL> """<STR_LIT>""" <EOL> sys . stdout . write ( ( "<STR_LIT>" . format ( message , [ "<STR_LIT:\n>" , "<STR_LIT>" ] [ skip_eol ] ) ) ) <EOL> sys . stdout . flush ( ) <EOL> def show_video_detail ( self , filename , metadata ) : <EOL> """<STR_LIT>""" <EOL> print ( "<STR_LIT>" . format ( filename ) ) <EOL> print ( "<STR_LIT>" . format ( metadata [ '<STR_LIT>' ] ) ) <EOL> print ( "<STR_LIT>" . format ( metadata [ '<STR_LIT>' ] ) ) <EOL> print ( "<STR_LIT>" . format ( metadata [ '<STR_LIT>' ] ) ) <EOL> print ( "<STR_LIT>" . format ( metadata [ '<STR_LIT>' ] ) ) <EOL> def error ( self , err , message = None ) : <EOL> """<STR_LIT>""" <EOL> if message is not None : <EOL> self . _to_stderr ( message ) <EOL> raise err ( message ) <EOL> def run ( self , filename , url ) : <EOL> """<STR_LIT>""" <EOL> extractor = self . info_extract <EOL> video_info = extractor . get_info ( url ) <EOL> if not video_info : <EOL> self . error ( ExtractionError , <EOL> '<STR_LIT>' ) <EOL> meta_url = video_info [ '<STR_LIT>' ] <EOL> meta_data = extractor . get_streaming_meta ( meta_url ) <EOL> if filename is None : <EOL> if self . opts . title is False : <EOL> name = video_info [ '<STR_LIT>' ] <EOL> else : <EOL> name = meta_data [ '<STR_LIT>' ] <EOL> filename = "<STR_LIT>" . format ( <EOL> name , <EOL> meta_data [ '<STR_LIT>' ] ) <EOL> self . show_video_detail ( filename , meta_data ) <EOL> if self . dump_video ( filename , meta_data ) : <EOL> print ( "<STR_LIT>" ) <EOL> def dump_video ( self , filename , meta_data ) : <EOL> """<STR_LIT>""" <EOL> if os . path . exists ( filename ) : <EOL> if self . opts . overwrite is True and self . opts . resume is False : <EOL> os . remove ( filename ) <EOL> elif self . opts . overwrite is False and self . opts . resume is False : <EOL> self . error ( DownloadError , <EOL> ( "<STR_LIT>" + <EOL> "<STR_LIT>" ) ) <EOL> try : <EOL> stdout = open ( os . path . devnull , '<STR_LIT:w>' ) <EOL> subprocess . call ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> stdout = stdout , <EOL> stderr = subprocess . STDOUT ) <EOL> except ( OSError , IOError ) : <EOL> print ( '<STR_LIT>' ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> finally : <EOL> stdout . close ( ) <EOL> basic_args = [ <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , meta_data [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' , meta_data [ '<STR_LIT:source>' ] , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , filename , <EOL> ] <EOL> if self . opts . resume is True : <EOL> basic_args . append ( '<STR_LIT>' ) <EOL> retval = subprocess . Popen ( basic_args ) <EOL> while True : <EOL> if retval . poll ( ) is not None : <EOL> break <EOL> if not os . path . exists ( filename ) : <EOL> continue <EOL> previous_size = os . path . getsize ( filename ) <EOL> display_size = _convert_display_size ( previous_size ) <EOL> self . _to_stdout ( self . _CLEAR_STDOUT , skip_eol = True ) <EOL> self . _to_stdout ( '<STR_LIT>' . format ( <EOL> display_size ) , <EOL> skip_eol = True ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> if retval . wait ( ) == <NUM_LIT:0> : <EOL> current_size = os . path . getsize ( filename ) <EOL> display_size = _convert_display_size ( current_size ) <EOL> self . _to_stdout ( self . _CLEAR_STDOUT , skip_eol = True ) <EOL> self . _to_stdout ( '<STR_LIT>' . format ( <EOL> display_size ) ) <EOL> return True <EOL> else : <EOL> self . error ( DownloadError , <EOL> ( '<STR_LIT>' + <EOL> '<STR_LIT>' ) ) <EOL> return False <EOL> class VideoInfoExtractor ( object ) : <EOL> """<STR_LIT>""" <EOL> _VALIDATE_URL = r'<STR_LIT>' <EOL> _GOOGLE_VIDEO = r'<STR_LIT>' <EOL> _META_TITLE_NAME = r'<STR_LIT>' <EOL> _ORIG_FILENAME = r'<STR_LIT>' <EOL> _XHR_REQUEST_PATH = r'<STR_LIT>' <EOL> def __init__ ( self ) : <EOL> self . body = '<STR_LIT>' <EOL> def _to_stderr ( self , message ) : <EOL> """<STR_LIT>""" <EOL> sys . stderr . write ( message ) <EOL> def error ( self , err , message = None ) : <EOL> """<STR_LIT>""" <EOL> if message is not None : <EOL> self . _to_stderr ( message ) <EOL> raise err ( message ) <EOL> def get_info ( self , url ) : <EOL> """<STR_LIT>""" <EOL> if not self . valid_url ( url ) : <EOL> return False <EOL> view_resp = requests . get ( url ) <EOL> body = view_resp . text . encode ( '<STR_LIT:utf8>' , '<STR_LIT:ignore>' ) . decode ( '<STR_LIT:utf8>' ) <EOL> info = self . extract_info ( body ) <EOL> info . update ( self . extract_streaming_path ( body ) ) <EOL> return info <EOL> def get_streaming_meta ( self , meta_url ) : <EOL> """<STR_LIT>""" <EOL> meta_resp = requests . get ( meta_url ) <EOL> meta_body = meta_resp . text . encode ( '<STR_LIT:utf8>' , '<STR_LIT:ignore>' ) . decode ( '<STR_LIT:utf8>' ) <EOL> meta = self . extract_streaming_source ( meta_body ) <EOL> meta . update ( self . extract_metadata ( meta_body ) ) <EOL> return meta <EOL> def extract_streaming_path ( self , body ) : <EOL> """<STR_LIT>""" <EOL> xhr_path_match = re . search ( self . _XHR_REQUEST_PATH , body ) <EOL> if xhr_path_match is not None : <EOL> xhr_path = xhr_path_match . group ( <NUM_LIT:1> ) <EOL> else : <EOL> xhr_path = None <EOL> xhr_url = "<STR_LIT>" . format ( xhr_path ) <EOL> return { '<STR_LIT>' : xhr_url } <EOL> def extract_metadata ( self , meta_body ) : <EOL> """<STR_LIT>""" <EOL> base_pattern = r'<STR_LIT>' <EOL> meta_title = _regex_match_get ( base_pattern . format ( '<STR_LIT:title>' ) , meta_body ) <EOL> meta_part = _regex_match_get ( base_pattern . format ( '<STR_LIT>' ) , meta_body ) <EOL> meta_date = _regex_match_get ( base_pattern . format ( '<STR_LIT:date>' ) , meta_body ) <EOL> meta_type = _regex_match_get ( base_pattern . format ( '<STR_LIT:type>' ) , meta_body ) <EOL> return { <EOL> '<STR_LIT>' : meta_title , <EOL> '<STR_LIT>' : meta_part , <EOL> '<STR_LIT>' : meta_date , <EOL> '<STR_LIT>' : meta_type , <EOL> } <EOL> def extract_streaming_source ( self , meta_body ) : <EOL> """<STR_LIT>""" <EOL> rtmp_source_match = re . search ( ( <EOL> r'<STR_LIT>' + <EOL> r'<STR_LIT>' + <EOL> r'<STR_LIT>' ) , <EOL> meta_body ) <EOL> if rtmp_source_match is not None : <EOL> ext = rtmp_source_match . group ( <NUM_LIT:1> ) <EOL> streamer = rtmp_source_match . group ( <NUM_LIT:2> ) <EOL> rtmp_source = rtmp_source_match . group ( <NUM_LIT:3> ) <EOL> else : <EOL> ext = None <EOL> streamer = None <EOL> rtmp_source = None <EOL> return { <EOL> '<STR_LIT>' : ext , <EOL> '<STR_LIT>' : streamer , <EOL> '<STR_LIT:source>' : rtmp_source , <EOL> } <EOL> def extract_info ( self , body ) : <EOL> """<STR_LIT>""" <EOL> xhr_url_match = re . search ( self . _XHR_REQUEST_PATH , body ) <EOL> if xhr_url_match is not None : <EOL> xhr_url = xhr_url_match . group ( <NUM_LIT:1> ) <EOL> else : <EOL> xhr_url = None <EOL> if xhr_url is not None and xhr_url . endswith ( '<STR_LIT>' ) : <EOL> default_filename = xhr_url . split ( '<STR_LIT:/>' ) [ <NUM_LIT:1> ] <EOL> else : <EOL> self . error ( ExtractionError , <EOL> "<STR_LIT>" ) <EOL> return { <EOL> '<STR_LIT>' : default_filename , <EOL> '<STR_LIT>' : xhr_url , <EOL> } <EOL> def valid_url ( self , url ) : <EOL> """<STR_LIT>""" <EOL> return ( re . match ( VideoInfoExtractor . _VALIDATE_URL , url ) is not None ) </s>
