本文整理汇总了Python中streamlink.plugin.api.http.json函数的典型用法代码示例。如果您正苦于以下问题:Python json函数的具体用法?Python json怎么用?Python json使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了json函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: _get_streams
def _get_streams(self):
match = _url_re.match(self.url)
if not match:
return
channel, media_id = match.group("channel", "media_id")
self.logger.debug("Matched URL: channel={0}, media_id={1}".format(channel, media_id))
if not media_id:
res = http.get(LIVE_API.format(channel))
livestream = http.json(res, schema=_live_schema)
if livestream.get("media_hosted_media"):
hosted = _live_schema.validate(livestream["media_hosted_media"])
self.logger.info("{0} is hosting {1}", livestream["media_user_name"], hosted["media_user_name"])
livestream = hosted
if not livestream["media_is_live"]:
return
media_id = livestream["media_id"]
media_type = "live"
else:
media_type = "video"
res = http.get(PLAYER_API.format(media_type, media_id))
player = http.json(res, schema=_player_schema)
if media_type == "live":
return self._get_live_streams(player)
else:
return self._get_video_streams(player)
开发者ID:amadu80,项目名称:repository.xvbmc,代码行数:30,代码来源:hitbox.py
示例2: _get_streams
def _get_streams(self):
match = self._url_re.match(self.url)
channel = match.group('channel')
res = http.get(self.FORMATS_URL.format(channel))
streams = http.json(res, schema=self._formats_schema)['streams']
if streams == []:
self.logger.error('Channel may be geo-restricted, not directly provided by PlayTV or not freely available')
return
for language in streams:
for protocol, bitrates in list(streams[language].items()):
# - Ignore non-supported protocols (RTSP, DASH)
# - Ignore deprecated Flash (RTMPE/HDS) streams (PlayTV doesn't provide anymore a Flash player)
if protocol in ['rtsp', 'flash', 'dash', 'hds']:
continue
for bitrate in bitrates['bitrates']:
if bitrate['value'] == 0:
continue
api_url = self.API_URL.format(channel, protocol, language, bitrate['value'])
res = http.get(api_url)
video_url = http.json(res, schema=self._api_schema)['url']
bs = '{0}k'.format(bitrate['value'])
if protocol == 'hls':
for _, stream in HLSStream.parse_variant_playlist(self.session, video_url).items():
yield bs, stream
elif protocol == 'hds':
for _, stream in HDSStream.parse_manifest(self.session, video_url).items():
yield bs, stream
开发者ID:amadu80,项目名称:repository.xvbmc,代码行数:31,代码来源:playtv.py
示例3: _get_streams
def _get_streams(self):
res = http.get(self.url)
match = _embed_re.search(res.text)
if match:
res = http.get(match.group(1))
match = _aptoma_id_re.search(res.text)
if not match:
return
aptoma_id = match.group(1)
if not _live_re.search(res.text):
res = http.get(METADATA_URL.format(aptoma_id))
metadata = http.json(res)
video_id = metadata["videoId"]
else:
video_id = aptoma_id
res = http.get(VIDEO_INFO_URL, params=dict(id=video_id))
video = http.json(res, schema=_video_schema)
streams = {}
for fmt, providers in video["formats"].items():
for name, provider in providers.items():
for ext, playlists in provider.items():
for playlist in playlists:
url = PLAYLIST_URL_FORMAT.format(**playlist)
parser = STREAM_TYPES[fmt]
try:
streams.update(parser(self.session, url))
except IOError as err:
self.logger.error("Failed to extract {0} streams: {1}",
fmt.upper(), err)
return streams
开发者ID:amadu80,项目名称:repository.xvbmc,代码行数:35,代码来源:aftonbladet.py
示例4: _get_qq_streams
def _get_qq_streams(self, vid):
res = http.get(QQ_STREAM_INFO_URL % (vid, 1))
info = http.json(res, schema=_qq_schema)
yield "live", HTTPStream(self.session, info)
res = http.get(QQ_STREAM_INFO_URL % (vid, 2))
info = http.json(res, schema=_qq_schema)
yield "live_http", HLSStream(self.session, info)
开发者ID:coder-alpha,项目名称:CcloudTv.bundle,代码行数:8,代码来源:tga.py
示例5: _get_live_streams
def _get_live_streams(self, subdomain):
"""
Get the live stream in a particular language
:param subdomain:
:return:
"""
res = http.get(self._live_api_url.format(subdomain))
live_res = http.json(res, schema=self._live_schema)
api_res = http.get(live_res[u"url"])
stream_data = http.json(api_res, schema=self._stream_api_schema)
return HLSStream.parse_variant_playlist(self.session, stream_data[u'primary'])
开发者ID:amadu80,项目名称:repository.xvbmc,代码行数:11,代码来源:euronews.py
示例6: _get_live_streams
def _get_live_streams(self, lang, path):
"""
Get the live stream in a particular language
:param lang:
:param path:
:return:
"""
res = http.get(self._live_api_url.format(lang, path))
live_res = http.json(res)['default']['uid']
post_data = '{"channel_url":"/api/channels/%s/"}' % live_res
try:
stream_data = http.json(http.post(self._stream_get_url, data=post_data))['stream_url']
except BaseException:
stream_data = http.json(http.post(self._stream_get_url, data=post_data))['channel_url']
return HLSStream.parse_variant_playlist(self.session, stream_data)
开发者ID:justastranger,项目名称:Twitchy,代码行数:15,代码来源:olympicchannel.py
示例7: _get_live_streams
def _get_live_streams(self, slug):
res = http.get(LIVE_CHANNELS_API_URL)
res = http.json(res, schema=_channels_schema)
for channel in filter(lambda c: c["Slug"] == slug, res):
servers = channel["StreamingServers"]
return self._parse_streaming_servers(servers)
开发者ID:coder-alpha,项目名称:CcloudTv.bundle,代码行数:7,代码来源:drdk.py
示例8: _get_streams
def _get_streams(self):
http.headers.update({"User-Agent": useragents.CHROME,
"Referer": self.referer})
fragment = dict(parse_qsl(urlparse(self.url).fragment))
link = fragment.get("link")
if not link:
link = self._get_tv_link()
if not link:
self.logger.error("Missing link fragment: stream unavailable")
return
player_url = self._api_url.format(link)
self.logger.debug("Requesting player API: {0} (referer={1})", player_url, self.referer)
res = http.get(player_url,
params={"_": int(time.time() * 1000)},
headers={"X-Requested-With": "XMLHttpRequest"})
try:
data = http.json(res, schema=self.api_schema)
except PluginError as e:
print(e)
self.logger.error("Cannot play this stream type")
else:
if data["status"]:
if data["file"].startswith("<"):
self.logger.error("Cannot play embedded streams")
else:
return HLSStream.parse_variant_playlist(self.session, data["file"])
else:
self.logger.error(data["text"])
开发者ID:justastranger,项目名称:Twitchy,代码行数:31,代码来源:seetv.py
示例9: _get_streams
def _get_streams(self):
flashvars = http.get(self.url, schema=_flashvars_schema)
if not flashvars:
return
params = {
"rt": "json",
"lc": "en_US",
"pt": "view",
"bpw": "",
"bid": flashvars["id"],
"adok": "",
"bno": ""
}
if re.search(_url_re_tw, self.url):
res = http.get(VIEW_LIVE_API_URL_TW, params=params)
elif re.search(_url_re_jp, self.url):
res = http.get(VIEW_LIVE_API_URL_JP, params=params)
else:
res = http.get(VIEW_LIVE_API_URL, params=params)
streams = http.json(res, schema=_view_live_schema)
for stream in streams:
stream_name = "{0}p".format(stream["bps"])
stream_params = {
"rtmp": stream["purl"],
"live": True
}
yield stream_name, RTMPStream(self.session, stream_params)
开发者ID:coder-alpha,项目名称:CcloudTv.bundle,代码行数:31,代码来源:afreecatv.py
示例10: _get_streams
def _get_streams(self):
channel = self.url_re.match(self.url).group(1)
res = http.get(self.api_url.format(channel))
data = http.json(res, schema=self.api_schema)
return HLSStream.parse_variant_playlist(self.session, data["channel_stream_url"])
开发者ID:amadu80,项目名称:repository.xvbmc,代码行数:7,代码来源:powerapp.py
示例11: _get_vod_stream
def _get_vod_stream(self, vod_id):
res = self._get_api_res("recordings", vod_id)
for sdata in http.json(res, schema=self._vod_schema):
if sdata["format"] == "hls":
hls_url = urljoin(sdata["url"], "manifest.m3u8")
yield "{0}p".format(sdata["height"]), HLSStream(self.session, hls_url)
开发者ID:amadu80,项目名称:repository.xvbmc,代码行数:7,代码来源:mixer.py
示例12: _get_channel_id
def _get_channel_id(self, domain):
channel_info = http.get(CHANNEL_INFO_URL % str(domain))
info = http.json(channel_info, schema=_channel_schema)
if info is None:
return 0, 0
return info['channel']['vid'], info['channel']['id']
开发者ID:coder-alpha,项目名称:CcloudTv.bundle,代码行数:7,代码来源:tga.py
示例13: _get_streams
def _get_streams(self):
match = _url_re.match(self.url)
username = match.group("username")
CSRFToken = str(uuid.uuid4().hex.upper()[0:32])
headers = {
"Content-Type": "application/x-www-form-urlencoded",
"X-CSRFToken": CSRFToken,
"X-Requested-With": "XMLHttpRequest",
"Referer": self.url,
}
cookies = {
"csrftoken": CSRFToken,
}
post_data = "room_slug={0}&bandwidth=high".format(username)
res = http.post(API_HLS, headers=headers, cookies=cookies, data=post_data)
data = http.json(res, schema=_post_schema)
if data["success"] is True and data["room_status"] == "public":
for s in HLSStream.parse_variant_playlist(self.session, data["url"]).items():
yield s
开发者ID:amadu80,项目名称:repository.xvbmc,代码行数:25,代码来源:chaturbate.py
示例14: _get_streams
def _get_streams(self):
match = _url_re.match(self.url)
res = http.get(STREAM_INFO_URL,
params=match.groupdict(),
acceptable_status=STATUS_UNAVAILABLE)
if res.status_code in STATUS_UNAVAILABLE:
return
data = http.json(res, schema=_stream_schema)
if data.get("hls_url"):
hls_url = data["hls_url"]
hls_name = "live"
elif data.get("replay_url"):
self.logger.info("Live Stream ended, using replay instead")
hls_url = data["replay_url"]
hls_name = "replay"
else:
raise NoStreamsError(self.url)
streams = HLSStream.parse_variant_playlist(self.session, hls_url)
if not streams:
return {hls_name: HLSStream(self.session, hls_url)}
else:
return streams
开发者ID:amadu80,项目名称:repository.xvbmc,代码行数:25,代码来源:periscope.py
示例15: _get_streams
def _get_streams(self):
self.url = http.resolve_url(self.url)
match = _url_re.match(self.url)
parsed = urlparse(self.url)
if parsed.fragment:
channel_id = parsed.fragment
elif parsed.path[:3] == '/v/':
channel_id = parsed.path.split('/')[-1]
else:
channel_id = match.group("channel")
if not channel_id:
return
channel_id = channel_id.lower().replace("/", "_")
res = http.get(API_URL.format(channel_id))
info = http.json(res, schema=_schema)
if not info["success"]:
return
if info.get("isLive"):
name = "live"
else:
name = "vod"
stream = HTTPStream(self.session, info["payload"])
# Wrap the stream in a FLVPlaylist to verify the FLV tags
stream = FLVPlaylist(self.session, [stream])
return {name: stream}
开发者ID:coder-alpha,项目名称:CcloudTv.bundle,代码行数:31,代码来源:veetle.py
示例16: _get_vod_stream
def _get_vod_stream(self):
vod_url = self.url
if vod_url.endswith('/'):
vod_url = vod_url[:-1]
json_url = '{0}.securevideo.json'.format(vod_url)
res = http.get(json_url)
match = _json_re.search(res.text)
if not match:
return
data = parse_json(match.group(1))
res = http.get(API_VOD.format(data['clientid'], data['mzid']))
data = http.json(res, schema=_stream_schema)
for d in data['targetUrls']:
if d['type'] == 'HDS':
hds_url = d['url']
for s in HDSStream.parse_manifest(self.session, hds_url).items():
yield s
if d['type'] == 'HLS':
hls_url = d['url']
for s in HLSStream.parse_variant_playlist(self.session, hls_url).items():
yield s
开发者ID:amadu80,项目名称:repository.xvbmc,代码行数:26,代码来源:vrtbe.py
示例17: _post_api
def _post_api(self, api, payload, schema):
res = http.post(api, json=payload)
data = http.json(res, schema=schema)
if data["result"] == "success":
post_data = data["reply"]
return post_data
开发者ID:justastranger,项目名称:Twitchy,代码行数:7,代码来源:garena.py
示例18: _get_streams_from_media
def _get_streams_from_media(self, media_id):
res = http.get(STREAM_INFO_URL.format(media_id), cookies=COOKIES)
media = http.json(res, schema=_media_schema)
params = extra_params = swf_url = None
for __ in media:
for __ in __["layerList"]:
for __ in __.get("sequenceList", []):
for layer in __["layerList"]:
name = layer["name"]
if name == "video":
params = layer.get("param")
elif name == "reporting":
extra_params = layer.get("param", {})
extra_params = extra_params.get("extraParams", {})
if not params:
return
if extra_params:
swf_url = extra_params.get("videoSwfURL")
mode = params.get("mode")
if mode == "live":
return self._get_live_streams(params, swf_url)
elif mode == "vod":
return self._get_vod_streams(params)
开发者ID:coder-alpha,项目名称:CcloudTv.bundle,代码行数:27,代码来源:dailymotion.py
示例19: _get_vod_streams
def _get_vod_streams(self, params):
manifest_url = params.get("autoURL")
if not manifest_url:
return
res = http.get(manifest_url)
if res.headers.get("Content-Type") == "application/f4m+xml":
streams = HDSStream.parse_manifest(self.session, res.url)
# TODO: Replace with "yield from" when dropping Python 2.
for __ in streams.items():
yield __
elif res.headers.get("Content-Type") == "application/vnd.apple.mpegurl":
streams = HLSStream.parse_variant_playlist(self.session, res.url)
# TODO: Replace with "yield from" when dropping Python 2.
for __ in streams.items():
yield __
else:
manifest = http.json(res, schema=_vod_manifest_schema)
for params in manifest["alternates"]:
name = "{0}p".format(params["height"])
stream = self._create_flv_playlist(params["template"])
yield name, stream
failovers = params.get("failover", [])
for failover in failovers:
stream = self._create_flv_playlist(failover)
yield name, stream
开发者ID:coder-alpha,项目名称:CcloudTv.bundle,代码行数:29,代码来源:dailymotion.py
示例20: _get_streams
def _get_streams(self):
# Get video ID and channel from URL
match = self._url_re.match(self.url)
video_id = match.group('video_id')
if video_id is None:
# Retrieve URL page and search for video ID
res = http.get(self.url)
match = self._video_id_re.search(res.text)
if match is None:
return
video_id = match.group('video_id')
res = http.get(self.API_URL.format(video_id))
videos = http.json(res, schema=self._api_schema)
parsed = []
headers = {'User-Agent': self._user_agent}
# Some videos may be also available on Dailymotion (especially on CNews)
if videos['ID_DM'] != '':
for stream in self.session.streams('https://www.dailymotion.com/video/' + videos['ID_DM']).items():
yield stream
for quality, video_url in list(videos['MEDIA']['VIDEOS'].items()):
# Ignore empty URLs
if video_url == '':
continue
# Ignore duplicate video URLs
if video_url in parsed:
continue
parsed.append(video_url)
try:
# HDS streams don't seem to work for live videos
if '.f4m' in video_url and 'LIVE' not in videos['TYPE']:
for stream in HDSStream.parse_manifest(self.session,
video_url,
params={'hdcore': self.HDCORE_VERSION},
headers=headers).items():
yield stream
elif '.m3u8' in video_url:
for stream in HLSStream.parse_variant_playlist(self.session,
video_url,
headers=headers).items():
yield stream
elif '.mp4' in video_url:
# Get bitrate from video filename
match = self._mp4_bitrate_re.match(video_url)
if match is not None:
bitrate = match.group('bitrate')
else:
bitrate = quality
yield bitrate, HTTPStream(self.session,
video_url,
params={'secret': self.SECRET},
headers=headers)
except IOError as err:
if '403 Client Error' in str(err):
self.logger.error('Failed to access stream, may be due to geo-restriction')
开发者ID:amadu80,项目名称:repository.xvbmc,代码行数:59,代码来源:canalplus.py
注:本文中的streamlink.plugin.api.http.json函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论