Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from streamlink.plugin.api import useragents
from streamlink.plugin.api import validate
from streamlink.stream import HLSStream
from streamlink.utils import parse_json
log = logging.getLogger(__name__)
class Mitele(Plugin):
_url_re = re.compile(r"https?://(?:www\.)?mitele\.es/directo/(?P[\w-]+)")
pdata_url = "https://indalo.mediaset.es/mmc-player/api/mmc/v1/{channel}/live/html5.json"
gate_url = "https://gatekeeper.mediaset.es"
error_schema = validate.Schema({
"code": validate.any(validate.text, int),
"message": validate.text,
})
pdata_schema = validate.Schema(validate.transform(parse_json), validate.any(
validate.all(
{
"locations": [{
"gcp": validate.text,
"ogn": validate.any(None, validate.text),
}],
},
validate.get("locations"),
validate.get(0),
),
error_schema,
))
gate_schema = validate.Schema(
import re
from functools import partial
from streamlink.plugin import Plugin
from streamlink.plugin.api import validate
from streamlink.stream import HLSStream
from streamlink.utils import parse_json
class TV360(Plugin):
url_re = re.compile(r"https?://(?:www.)?tv360.com.tr/CanliYayin")
data_re = re.compile(r'''div.*?data-tp=(?P<q>["'])(?P<data>.*?)(?P=q)''', re.DOTALL)
_js_to_json = partial(re.compile(r"""(\w+):(["']|\d+,|true|false)""").sub, r'"\1":\2')
data_schema = validate.Schema(
validate.transform(data_re.search),
validate.any(
None,
validate.all(
validate.get("data"),
validate.transform(_js_to_json),
validate.transform(lambda x: x.replace("'", '"')),
validate.transform(parse_json),
{
"tp_type": "hls4",
"tp_file": validate.url(),
}
)
)
)
@classmethod
def can_handle_url(cls, url):</data></q>
/broadcast/(?P\d+)
""", re.VERBOSE)
_api_url = "https://live-api.line-apps.com/app/v3.2/channel/{0}/broadcast/{1}/player_status"
_player_status_schema = validate.Schema(
{
"liveStatus": validate.text,
"liveHLSURLs": validate.any(None, {
"720": validate.any(None, validate.url(scheme="http", path=validate.endswith(".m3u8"))),
"480": validate.any(None, validate.url(scheme="http", path=validate.endswith(".m3u8"))),
"360": validate.any(None, validate.url(scheme="http", path=validate.endswith(".m3u8"))),
"240": validate.any(None, validate.url(scheme="http", path=validate.endswith(".m3u8"))),
"144": validate.any(None, validate.url(scheme="http", path=validate.endswith(".m3u8"))),
}),
"archivedHLSURLs": validate.any(None, {
"720": validate.any(None, validate.url(scheme="http", path=validate.endswith(".m3u8"))),
"480": validate.any(None, validate.url(scheme="http", path=validate.endswith(".m3u8"))),
"360": validate.any(None, validate.url(scheme="http", path=validate.endswith(".m3u8"))),
"240": validate.any(None, validate.url(scheme="http", path=validate.endswith(".m3u8"))),
"144": validate.any(None, validate.url(scheme="http", path=validate.endswith(".m3u8"))),
}),
})
@classmethod
def can_handle_url(cls, url):
return cls._url_re.match(url) is not None
def _get_live_streams(self, json):
for stream in json["liveHLSURLs"]:
url = json["liveHLSURLs"][stream]
if url is not None:
"user": {
"unique_name": validate.text,
"name": validate.all(validate.text,
validate.transform(maybe_decode))
},
validate.optional("hls_movie"): {
"url": validate.text
}
}
)
_user_schema = validate.Schema(
{
"owner": _user_dict_schema,
"performers": [
validate.any(_user_dict_schema, None)
]
}
)
_data_lives_schema = validate.Schema(
{
"data": {
"lives": [_user_schema]
}
},
validate.get("data"),
validate.get("lives")
)
api_lives = "https://sketch.pixiv.net/api/lives.json"
login_url_get = "https://accounts.pixiv.net/login"
from functools import partial
from streamlink.plugin import Plugin
from streamlink.plugin.api import validate, StreamMapper
from streamlink.stream import HLSStream, DASHStream
from streamlink.utils import parse_json, update_scheme, search_dict
log = logging.getLogger(__name__)
class AtresPlayer(Plugin):
url_re = re.compile(r"https?://(?:www.)?atresplayer.com/")
state_re = re.compile(r"""window.__PRELOADED_STATE__\s*=\s*({.*?});""", re.DOTALL)
channel_id_schema = validate.Schema(
validate.transform(state_re.search),
validate.any(
None,
validate.all(
validate.get(1),
validate.transform(parse_json),
validate.transform(partial(search_dict, key="urlVideo"))
)
)
)
stream_schema = validate.Schema(
validate.transform(parse_json),
{"sources": [
validate.all({
"src": validate.url(),
validate.optional("type"): validate.text
})
]}, validate.get("sources"))
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36"
HUAJIAO_URL = "http://www.huajiao.com/l/{}"
LAPI_URL = "http://g2.live.360.cn/liveplay?stype=flv&channel={}&bid=huajiao&sn={}&sid={}&_rate=xd&ts={}&r={}&_ostype=flash&_delay=0&_sign=null&_ver=13"
_url_re = re.compile(r"""
http(s)?://(www\.)?huajiao.com
/l/(?P[^/]+)
""", re.VERBOSE)
_feed_json_re = re.compile(r'^\s*var\s*feed\s*=\s*(?P{.*})\s*;', re.MULTILINE)
_feed_json_schema = validate.Schema(
validate.all(
validate.transform(_feed_json_re.search),
validate.any(
None,
validate.all(
validate.get('feed'),
validate.transform(json.loads)
)
)
)
)
class Huajiao(Plugin):
@classmethod
def can_handle_url(self, url):
return _url_re.match(url)
def _get_streams(self):
"CHANNEL": {
"RESULT": validate.transform(int),
validate.optional("BPWD"): validate.text,
validate.optional("BNO"): validate.text,
validate.optional("RMD"): validate.text,
validate.optional("AID"): validate.text,
validate.optional("CDN"): validate.text
}
},
validate.get("CHANNEL")
)
_stream_schema = validate.Schema(
{
validate.optional("view_url"): validate.url(
scheme=validate.any("rtmp", "http")
),
"stream_status": validate.text
}
)
class AfreecaTV(Plugin):
login_url = "https://member.afreecatv.com:8111/login/LoginAction.php"
arguments = PluginArguments(
PluginArgument(
"username",
requires=["password"],
metavar="USERNAME",
help="The username used to register with afreecatv.com."
),
_stream_size_re = re.compile(r'https?://.+-(?P\d+p?)\..+?$')
_video_player_re = re.compile(r'.+?)".*?', re.DOTALL)
_video_stream_data_re = re.compile(r'
"rtmp_live": validate.text,
"hls_url": validate.text,
"rtmp_multi_bitrate": validate.all(
validate.any([], {
validate.text: validate.text
}),
validate.transform(dict)
)
})
},
validate.get("data")
)
_vapi_schema = validate.Schema(
{
"data": validate.any(None, {
"video_url": validate.text
})
},
validate.get("data")
)
class Douyutv(Plugin):
@classmethod
def can_handle_url(cls, url):
return _url_re.match(url)
@classmethod
def stream_weight(cls, stream):
if stream in STREAM_WEIGHTS:
return STREAM_WEIGHTS[stream], "douyutv"