Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
addr, port = args.socks_proxy.split(':')
socks.set_default_proxy(socks.SOCKS5, addr, int(port))
socket.socket = socks.socksocket
except:
print('Failed to set socks5 proxy. Please install PySocks.', file=sys.stderr)
opener = build_opener(*handlers)
install_opener(opener)
m, u = url_to_module(args.video_url)
info = m.parser(u)
# Is a playlist?
if m.list_only():
video_list = m.prepare_list()
result = [ {'title': match1(get_content(url), r'<title>(.+?)</title>'), 'url': url} for url in video_list ]
else:
result = info.jsonlize()
print(json.dumps(result, indent=4, sort_keys=True, ensure_ascii=False))
def prepare(self):
info = VideoInfo(self.name, True)
html = get_content(self.url)
self.vid = match1(html, '"qipuId":(\d+),')
title = match1(html, '"roomTitle":"([^"]+)",')
artist = match1(html, '"anchorNickname":"([^"]+)",')
info.title = u'{} - {}'.format(title, artist)
info.artist = artist
data = getlive(self.vid)
self.logger.debug('data:\n' + str(data))
assert data['code'] == 'A00000', data.get('msg', 'can\'t play this live video!!')
data = data['data']
for stream in data['streams']:
stream_type = stream['steamType'] # typo 'streamType' to 'steamType'
stream_id = self.type_2_id[stream_type]
if stream['formatType'] == 'HLFLV':
stream_params = stream['url'].split('?')[-1]
stream_params_dict = dict((k, v[0]) for k, v in parse_qs(stream_params).items())
if stream_params_dict['hl_sttp'] != 'flv':
quality = max([q['width'], q['height']])
stream_type = self.quality2_2_id[quality]
stream_profile = self.id_2_profile[stream_type]
info.stream_types.append(stream_type)
info.streams[stream_type] = {
'container': 'm3u8',
'video_profile': stream_profile,
'src': [q[url]],
'size': 0
}
else:
add_header('Referer', 'https://www.acfun.cn/')
if m3u8Info is None:
data = json.loads(get_content('https://www.acfun.cn/rest/pc-direct/play/playInfo/m3u8Auto?videoId={}'.format(sourceVid)))
m3u8Info = data['playInfo']['streams'][0]
# some videos are broken of CDN, random choose one
m3u8api = random.choice(m3u8Info['playUrls'])
self.logger.warning('Request m3u8 info via CDN: %s\nIf video has broken on this CDN, please retry.', m3u8api)
lines = get_content(m3u8api)
self.logger.debug('m3u8 api: %s', lines)
lines = lines.split('\n')
resolutions = None
for line in lines:
if resolutions is None:
resolutions = match1(line, 'RESOLUTION=(\d+x\d+)')
if resolutions:
resolutions = [int(q) for q in resolutions.split('x')]
elif match1(line, '(\.m3u8)'):
def prepare(self):
add_header("Cookie", '__ysuid=%d' % time.time())
info = VideoInfo(self.name)
if not self.vid:
self.vid = match1(self.url.split('//', 1)[1],
'^v[^\.]?\.[^/]+/v_show/id_([a-zA-Z0-9=]+)',
'^player[^/]+/(?:player\.php/sid|embed)/([a-zA-Z0-9=]+)',
'^static.+loader\.swf\?VideoIDS=([a-zA-Z0-9=]+)',
'^(?:new-play|video)\.tudou\.com/v/([a-zA-Z0-9=]+)')
if not self.vid:
html = get_content(self.url)
self.vid = match1(html, r'videoIds?[\"\']?\s*[:=]\s*[\"\']?([a-zA-Z0-9=]+)')
if self.vid.isdigit():
import base64
vid = base64.b64encode(b'%d' % (int(self.vid) * 4))
if not isinstance(vid, str):
def prepare(self):
add_default_handler(HTTPCookieProcessor)
install_default_handlers()
info = VideoInfo(self.name, True)
html = get_content(self.url)
self.vid = match1(html, '"userNum":(\d+)')
live_id = match1(html, '"liveId":\s*(\d+)')
assert live_id, u"主播正在休息"
info.stream_types.append('current')
info.streams['current'] = {'container': 'mp4', 'src': ['http://extapi.live.netease.com/redirect/video/{}'.format(self.vid)], 'size' : float('inf')}
info.artist = match1(html, '"nick":"([^"]+)')
info.title = match1(html, '<title>([^<]+)').split('-')[0]
return info
</title>
def prepare(self):
add_header("Cookie", '__ysuid=%d' % time.time())
info = VideoInfo(self.name)
if not self.vid:
self.vid = match1(self.url.split('//', 1)[1],
'^v[^\.]?\.[^/]+/v_show/id_([a-zA-Z0-9=]+)',
'^player[^/]+/(?:player\.php/sid|embed)/([a-zA-Z0-9=]+)',
'^static.+loader\.swf\?VideoIDS=([a-zA-Z0-9=]+)',
'^(?:new-play|video)\.tudou\.com/v/([a-zA-Z0-9=]+)')
if not self.vid:
html = get_content(self.url)
self.vid = match1(html, r'videoIds?[\"\']?\s*[:=]\s*[\"\']?([a-zA-Z0-9=]+)')
if self.vid.isdigit():
import base64
})
handlers.append(proxy_handler)
elif args.socks_proxy:
try:
import socks
addr, port = args.socks_proxy.split(':')
socks.set_default_proxy(socks.SOCKS5, addr, int(port))
socket.socket = socks.socksocket
except:
print('Failed to set socks5 proxy. Please install PySocks.', file=sys.stderr)
opener = build_opener(*handlers)
install_opener(opener)
m, u = url_to_module(args.video_url)
info = m.parser(u)
# Is a playlist?
if m.list_only():
video_list = m.prepare_list()
result = [ {'title': match1(get_content(url), r'<title>(.+?)</title>'), 'url': url} for url in video_list ]
else:
result = info.jsonlize()
print(json.dumps(result, indent=4, sort_keys=True, ensure_ascii=False))
def main():
args = arg_parser()
handlers = []
if args.timeout:
socket.setdefaulttimeout(args.timeout)
if args.user_agent:
fake_headers['User-Agent'] = args.user_agent
if args.http_proxy:
proxy_handler = ProxyHandler({
'http': args.http_proxy,
'https': args.http_proxy
})
handlers.append(proxy_handler)
elif args.socks_proxy:
try:
import socks
addr, port = args.socks_proxy.split(':')
socks.set_default_proxy(socks.SOCKS5, addr, int(port))
socket.socket = socks.socksocket
except:
print('Failed to set socks5 proxy. Please install PySocks.', file=sys.stderr)
def l_assert(self):
if self.url.startswith('http://'):
self.url = self.url.replace('http://', 'https://', 1)
self.url = get_location(self.url)
if 'passport.weibo.com' in self.url:
query = urlsplit(self.url).query
self.url = parse_qs(query)['url'][0]
return self.l_assert()
# Mobile ver.
if 'm.weibo.cn' in self.url:
self.title_patterns = '"content2": "(.+?)",', '"status_title": "(.+?)",'
self.url_patterns = '"stream_url_hd": "([^"]+)', '"stream_url": "([^"]+)'
return
if '/tv/v/' in self.url or 'fid=' not in self.url:
self.title_patterns = 'class="info_txt \w+">([^<]+)]+>\s*(?:)?\s*([^<]+)'
self.url_patterns = 'video-sources\s*=\s*".+?(?:&\d+=http.+?)*&\d+=(http.+?[^=])(?:&\d+=)*&qType=\w+"',
return
proxy_handler = ProxyHandler({
'http': args.http_proxy,
'https': args.http_proxy
})
handlers.append(proxy_handler)
elif args.socks_proxy:
try:
import socks
addr, port = args.socks_proxy.split(':')
socks.set_default_proxy(socks.SOCKS5, addr, int(port))
socket.socket = socks.socksocket
except:
print('Failed to set socks5 proxy. Please install PySocks.', file=sys.stderr)
opener = build_opener(*handlers)
install_opener(opener)
m, u = url_to_module(args.video_url)
info = m.parser(u)
# Is a playlist?
if m.list_only():
video_list = m.prepare_list()
result = [ {'title': match1(get_content(url), r'<title>(.+?)</title>'), 'url': url} for url in video_list ]
else:
result = info.jsonlize()
print(json.dumps(result, indent=4, sort_keys=True, ensure_ascii=False))