<commit_before>import json
import logging
import re
from collections import namedtuple
from random import random
from urllib.parse import urlparse

import requests

from streamlink.exceptions import NoStreamsError, PluginError
from streamlink.plugin import Plugin, PluginArgument, PluginArguments
from streamlink.plugin.api import validate
from streamlink.plugin.api.utils import parse_json, parse_query
from streamlink.stream import HLSStream, HTTPStream
from streamlink.stream.hls import HLSStreamWorker
from streamlink.stream.hls_filtered import FilteredHLSStreamReader, FilteredHLSStreamWriter
from streamlink.stream.hls_playlist import M3U8, M3U8Parser, load as load_hls_playlist
from streamlink.utils.times import hours_minutes_seconds

log = logging.getLogger(__name__)

Segment = namedtuple("Segment", "uri duration title key discontinuity ad byterange date map prefetch")

LOW_LATENCY_MAX_LIVE_EDGE = 2


class TwitchM3U8(M3U8):
    def __init__(self):
        super(TwitchM3U8, self).__init__()
        self.dateranges_ads = []


class TwitchM3U8Parser(M3U8Parser):
    def parse_tag_ext_x_twitch_prefetch(self, value):
        segments = self.m3u8.segments
        if segments:
            segments.append(segments[-1]._replace(uri=self.uri(value), prefetch=True))

    def parse_tag_ext_x_daterange(self, value):
        super(TwitchM3U8Parser, self).parse_tag_ext_x_daterange(value)
        daterange = self.m3u8.dateranges[-1]
        is_ad = (
            daterange.classname == "twitch-stitched-ad"
            or str(daterange.id or "").startswith("stitched-ad-")
            or any(attr_key.startswith("X-TV-TWITCH-AD-") for attr_key in daterange.x.keys())
        )
        if is_ad:
            self.m3u8.dateranges_ads.append(daterange)

    def get_segment(self, uri):
        byterange = self.state.pop("byterange", None)
        extinf = self.state.pop("extinf", (0, None))
        date = self.state.pop("date", None)
        map_ = self.state.get("map")
        key = self.state.get("key")
        discontinuity = self.state.pop("discontinuity", False)
        ad = any(self.m3u8.is_date_in_daterange(date, daterange) for daterange in self.m3u8.dateranges_ads)

        return Segment(
            uri,
            extinf[0],
            extinf[1],
            key,
            discontinuity,
            ad,
            byterange,
            date,
            map_,
            prefetch=False
        )


class TwitchHLSStreamWorker(HLSStreamWorker):
    def __init__(self, reader, *args, **kwargs):
        self.had_content = False
        super(TwitchHLSStreamWorker, self).__init__(reader, *args, **kwargs)

    def _reload_playlist(self, *args):
        return load_hls_playlist(*args, parser=TwitchM3U8Parser, m3u8=TwitchM3U8)

    def _playlist_reload_time(self, playlist, sequences):
        if self.stream.low_latency and sequences:
            return sequences[-1].segment.duration

        return super(TwitchHLSStreamWorker, self)._playlist_reload_time(playlist, sequences)

    def process_sequences(self, playlist, sequences):
        # ignore prefetch segments if not LL streaming
        if not self.stream.low_latency:
            sequences = [seq for seq in sequences if not seq.segment.prefetch]

        # check for sequences with real content
        if not self.had_content:
            self.had_content = next((True for seq in sequences if not seq.segment.ad), False)

            # When filtering ads, to check whether it's a LL stream, we need to wait for the real content to show up,
            # since playlists with only ad segments don't contain prefetch segments
            if (
                self.stream.low_latency
                and self.had_content
                and not next((True for seq in sequences if seq.segment.prefetch), False)
            ):
                log.info("This is not a low latency stream")

        # show pre-roll ads message only on the first playlist containing ads
        if self.stream.disable_ads and self.playlist_sequence == -1 and not self.had_content:
            log.info("Waiting for pre-roll ads to finish, be patient")

        return super(TwitchHLSStreamWorker, self).process_sequences(playlist, sequences)


class TwitchHLSStreamWriter(FilteredHLSStreamWriter):
    def should_filter_sequence(self, sequence):
        return self.stream.disable_ads and sequence.segment.ad


class TwitchHLSStreamReader(FilteredHLSStreamReader):
    __worker__ = TwitchHLSStreamWorker
    __writer__ = TwitchHLSStreamWriter


class TwitchHLSStream(HLSStream):
    def __init__(self, *args, **kwargs):
        super(TwitchHLSStream, self).__init__(*args, **kwargs)
        self.disable_ads = self.session.get_plugin_option("twitch", "disable-ads")
        self.low_latency = self.session.get_plugin_option("twitch", "low-latency")

    def open(self):
        if self.disable_ads:
            log.info("Will skip ad segments")
        if self.low_latency:
            live_edge = max(1, min(LOW_LATENCY_MAX_LIVE_EDGE, self.session.options.get("hls-live-edge")))
            self.session.options.set("hls-live-edge", live_edge)
            self.session.options.set("hls-segment-stream-data", True)
            log.info("Low latency streaming (HLS live edge: {0})".format(live_edge))

        reader = TwitchHLSStreamReader(self)
        reader.open()

        return reader

    @classmethod
    def _get_variant_playlist(cls, res):
        return load_hls_playlist(res.text, base_uri=res.url)


class UsherService(object):
    def __init__(self, session):
        self.session = session

    def _create_url(self, endpoint, **extra_params):
        url = "https://usher.ttvnw.net{0}".format(endpoint)
        params = {
            "player": "twitchweb",
            "p": int(random() * 999999),
            "type": "any",
            "allow_source": "true",
            "allow_audio_only": "true",
            "allow_spectre": "false",
        }
        params.update(extra_params)

        req = requests.Request("GET", url, params=params)
        req = self.session.http.prepare_request(req)

        return req.url

    def channel(self, channel, **extra_params):
        return self._create_url("/api/channel/hls/{0}.m3u8".format(channel),
                                **extra_params)

    def video(self, video_id, **extra_params):
        return self._create_url("/vod/{0}".format(video_id), **extra_params)


class TwitchAPI(object):
    # Streamlink's client-id used for public API calls (don't steal this and register your own application on Twitch)
    TWITCH_CLIENT_ID = "pwkzresl8kj2rdj6g7bvxl9ys1wly3j"
    # Twitch's client-id used for private API calls (see issue #2680 for why we are doing this)
    TWITCH_CLIENT_ID_PRIVATE = "kimne78kx3ncx6brgo4mv6wki5h1ko"

    def __init__(self, session):
        self.session = session

    def _call(self, method="GET", subdomain="api", path="/", headers=None, private=False, data=None, **params):
        url = "https://{0}.twitch.tv{1}".format(subdomain, path)
        headers = headers or dict()
        headers.update({
            "Client-ID": self.TWITCH_CLIENT_ID if not private else self.TWITCH_CLIENT_ID_PRIVATE
        })

        return self.session.http.request(method, url, data=data, params=params, headers=headers)

    def call(self, path, schema=None, **params):
        headers = {"Accept": "application/vnd.twitchtv.v5+json"}
        res = self._call(path=path, headers=headers, **params)

        return self.session.http.json(res, schema=schema)

    def call_gql(self, data, schema=None, **params):
        res = self._call(method="POST", subdomain="gql", path="/gql", data=json.dumps(data), private=True, **params)

        return self.session.http.json(res, schema=schema)

    # Public API calls

    def channel_from_video_id(self, video_id):
        return self.call("/kraken/videos/{0}".format(video_id), schema=validate.Schema(
            {
                "channel": {
                    "_id": validate.any(int, validate.text),
                    "name": validate.text
                }
            },
            validate.get("channel"),
            validate.union((
                validate.all(validate.get("_id"), validate.transform(int)),
                validate.all(validate.get("name"), validate.transform(lambda n: n.lower()))
            ))
        ))

    def channel_from_login(self, channel):
        return self.call("/kraken/users", login=channel, schema=validate.Schema(
            {
                "users": [{
                    "_id": validate.any(int, validate.text)
                }]
            },
            validate.get("users"),
            validate.get(0),
            validate.get("_id"),
            validate.transform(int)
        ))

    def stream_rerun(self, channel_id):
        return self.call("/kraken/streams/{0}".format(channel_id), schema=validate.Schema(
            {
                "stream": validate.any(None, validate.all({
                    "stream_type": validate.text,
                    "broadcast_platform": validate.text,
                    "channel": validate.all({
                        "broadcaster_software": validate.text
                    }, validate.get("broadcaster_software"))
                }, validate.union((
                    validate.get("stream_type"),
                    validate.get("broadcast_platform"),
                    validate.get("channel")
                ))))
            },
            validate.get("stream")
        ))

    def metadata_video(self, video_id):
        return self.call("/kraken/videos/{0}".format(video_id), schema=validate.Schema(validate.any(
            validate.all(
                {
                    "title": validate.text,
                    "game": validate.text,
                    "channel": validate.all(
                        {"display_name": validate.text},
                        validate.get("display_name")
                    )
                },
                validate.transform(lambda data: (data["channel"], data["title"], data["game"]))
            ),
            validate.all({}, validate.transform(lambda _: (None,) * 3))
        )))

    def metadata_channel(self, channel_id):
        return self.call("/kraken/streams/{0}".format(channel_id), schema=validate.Schema(
            {
                "stream": validate.any(
                    validate.all(
                        {"channel": {
                            "display_name": validate.text,
                            "game": validate.text,
                            "status": validate.text
                        }},
                        validate.get("channel"),
                        validate.transform(lambda ch: (ch["display_name"], ch["status"], ch["game"]))
                    ),
                    validate.all(None, validate.transform(lambda _: (None,) * 3))
                )
            },
            validate.get("stream")
        ))

    # Private API calls

    def access_token(self, endpoint, asset):
        return self.call("/api/{0}/{1}/access_token".format(endpoint, asset), private=True, schema=validate.Schema(
            {
                "token": validate.text,
                "sig": validate.text
            },
            validate.union((
                validate.get("sig"),
                validate.get("token")
            ))
        ))
<commit_msg>plugins.twitch: player_type access token parameter<commit_after>import json
import logging
import re
from collections import namedtuple
from random import random
from urllib.parse import urlparse

import requests

from streamlink.exceptions import NoStreamsError, PluginError
from streamlink.plugin import Plugin, PluginArgument, PluginArguments
from streamlink.plugin.api import validate
from streamlink.plugin.api.utils import parse_json, parse_query
from streamlink.stream import HLSStream, HTTPStream
from streamlink.stream.hls import HLSStreamWorker
from streamlink.stream.hls_filtered import FilteredHLSStreamReader, FilteredHLSStreamWriter
from streamlink.stream.hls_playlist import M3U8, M3U8Parser, load as load_hls_playlist
from streamlink.utils.times import hours_minutes_seconds

log = logging.getLogger(__name__)

Segment = namedtuple("Segment", "uri duration title key discontinuity ad byterange date map prefetch")

LOW_LATENCY_MAX_LIVE_EDGE = 2


class TwitchM3U8(M3U8):
    def __init__(self):
        super(TwitchM3U8, self).__init__()
        self.dateranges_ads = []


class TwitchM3U8Parser(M3U8Parser):
    def parse_tag_ext_x_twitch_prefetch(self, value):
        segments = self.m3u8.segments
        if segments:
            segments.append(segments[-1]._replace(uri=self.uri(value), prefetch=True))

    def parse_tag_ext_x_daterange(self, value):
        super(TwitchM3U8Parser, self).parse_tag_ext_x_daterange(value)
        daterange = self.m3u8.dateranges[-1]
        is_ad = (
            daterange.classname == "twitch-stitched-ad"
            or str(daterange.id or "").startswith("stitched-ad-")
            or any(attr_key.startswith("X-TV-TWITCH-AD-") for attr_key in daterange.x.keys())
        )
        if is_ad:
            self.m3u8.dateranges_ads.append(daterange)

    def get_segment(self, uri):
        byterange = self.state.pop("byterange", None)
        extinf = self.state.pop("extinf", (0, None))
        date = self.state.pop("date", None)
        map_ = self.state.get("map")
        key = self.state.get("key")
        discontinuity = self.state.pop("discontinuity", False)
        ad = any(self.m3u8.is_date_in_daterange(date, daterange) for daterange in self.m3u8.dateranges_ads)

        return Segment(
            uri,
            extinf[0],
            extinf[1],
            key,
            discontinuity,
            ad,
            byterange,
            date,
            map_,
            prefetch=False
        )


class TwitchHLSStreamWorker(HLSStreamWorker):
    def __init__(self, reader, *args, **kwargs):
        self.had_content = False
        super(TwitchHLSStreamWorker, self).__init__(reader, *args, **kwargs)

    def _reload_playlist(self, *args):
        return load_hls_playlist(*args, parser=TwitchM3U8Parser, m3u8=TwitchM3U8)

    def _playlist_reload_time(self, playlist, sequences):
        if self.stream.low_latency and sequences:
            return sequences[-1].segment.duration

        return super(TwitchHLSStreamWorker, self)._playlist_reload_time(playlist, sequences)

    def process_sequences(self, playlist, sequences):
        # ignore prefetch segments if not LL streaming
        if not self.stream.low_latency:
            sequences = [seq for seq in sequences if not seq.segment.prefetch]

        # check for sequences with real content
        if not self.had_content:
            self.had_content = next((True for seq in sequences if not seq.segment.ad), False)

            # When filtering ads, to check whether it's a LL stream, we need to wait for the real content to show up,
            # since playlists with only ad segments don't contain prefetch segments
            if (
                self.stream.low_latency
                and self.had_content
                and not next((True for seq in sequences if seq.segment.prefetch), False)
            ):
                log.info("This is not a low latency stream")

        # show pre-roll ads message only on the first playlist containing ads
        if self.stream.disable_ads and self.playlist_sequence == -1 and not self.had_content:
            log.info("Waiting for pre-roll ads to finish, be patient")

        return super(TwitchHLSStreamWorker, self).process_sequences(playlist, sequences)


class TwitchHLSStreamWriter(FilteredHLSStreamWriter):
    def should_filter_sequence(self, sequence):
        return self.stream.disable_ads and sequence.segment.ad


class TwitchHLSStreamReader(FilteredHLSStreamReader):
    __worker__ = TwitchHLSStreamWorker
    __writer__ = TwitchHLSStreamWriter


class TwitchHLSStream(HLSStream):
    def __init__(self, *args, **kwargs):
        super(TwitchHLSStream, self).__init__(*args, **kwargs)
        self.disable_ads = self.session.get_plugin_option("twitch", "disable-ads")
        self.low_latency = self.session.get_plugin_option("twitch", "low-latency")

    def open(self):
        if self.disable_ads:
            log.info("Will skip ad segments")
        if self.low_latency:
            live_edge = max(1, min(LOW_LATENCY_MAX_LIVE_EDGE, self.session.options.get("hls-live-edge")))
            self.session.options.set("hls-live-edge", live_edge)
            self.session.options.set("hls-segment-stream-data", True)
            log.info("Low latency streaming (HLS live edge: {0})".format(live_edge))

        reader = TwitchHLSStreamReader(self)
        reader.open()

        return reader

    @classmethod
    def _get_variant_playlist(cls, res):
        return load_hls_playlist(res.text, base_uri=res.url)


class UsherService(object):
    def __init__(self, session):
        self.session = session

    def _create_url(self, endpoint, **extra_params):
        url = "https://usher.ttvnw.net{0}".format(endpoint)
        params = {
            "player": "twitchweb",
            "p": int(random() * 999999),
            "type": "any",
            "allow_source": "true",
            "allow_audio_only": "true",
            "allow_spectre": "false",
        }
        params.update(extra_params)

        req = requests.Request("GET", url, params=params)
        req = self.session.http.prepare_request(req)

        return req.url

    def channel(self, channel, **extra_params):
        return self._create_url("/api/channel/hls/{0}.m3u8".format(channel),
                                **extra_params)

    def video(self, video_id, **extra_params):
        return self._create_url("/vod/{0}".format(video_id), **extra_params)


class TwitchAPI(object):
    # Streamlink's client-id used for public API calls (don't steal this and register your own application on Twitch)
    TWITCH_CLIENT_ID = "pwkzresl8kj2rdj6g7bvxl9ys1wly3j"
    # Twitch's client-id used for private API calls (see issue #2680 for why we are doing this)
    TWITCH_CLIENT_ID_PRIVATE = "kimne78kx3ncx6brgo4mv6wki5h1ko"

    def __init__(self, session):
        self.session = session

    def _call(self, method="GET", subdomain="api", path="/", headers=None, private=False, data=None, **params):
        url = "https://{0}.twitch.tv{1}".format(subdomain, path)
        headers = headers or dict()
        headers.update({
            "Client-ID": self.TWITCH_CLIENT_ID if not private else self.TWITCH_CLIENT_ID_PRIVATE
        })

        return self.session.http.request(method, url, data=data, params=params, headers=headers)

    def call(self, path, schema=None, **params):
        headers = {"Accept": "application/vnd.twitchtv.v5+json"}
        res = self._call(path=path, headers=headers, **params)

        return self.session.http.json(res, schema=schema)

    def call_gql(self, data, schema=None, **params):
        res = self._call(method="POST", subdomain="gql", path="/gql", data=json.dumps(data), private=True, **params)

        return self.session.http.json(res, schema=schema)

    # Public API calls

    def channel_from_video_id(self, video_id):
        return self.call("/kraken/videos/{0}".format(video_id), schema=validate.Schema(
            {
                "channel": {
                    "_id": validate.any(int, validate.text),
                    "name": validate.text
                }
            },
            validate.get("channel"),
            validate.union((
                validate.all(validate.get("_id"), validate.transform(int)),
                validate.all(validate.get("name"), validate.transform(lambda n: n.lower()))
            ))
        ))

    def channel_from_login(self, channel):
        return self.call("/kraken/users", login=channel, schema=validate.Schema(
            {
                "users": [{
                    "_id": validate.any(int, validate.text)
                }]
            },
            validate.get("users"),
            validate.get(0),
            validate.get("_id"),
            validate.transform(int)
        ))

    def stream_rerun(self, channel_id):
        return self.call("/kraken/streams/{0}".format(channel_id), schema=validate.Schema(
            {
                "stream": validate.any(None, validate.all({
                    "stream_type": validate.text,
                    "broadcast_platform": validate.text,
                    "channel": validate.all({
                        "broadcaster_software": validate.text
                    }, validate.get("broadcaster_software"))
                }, validate.union((
                    validate.get("stream_type"),
                    validate.get("broadcast_platform"),
                    validate.get("channel")
                ))))
            },
            validate.get("stream")
        ))

    def metadata_video(self, video_id):
        return self.call("/kraken/videos/{0}".format(video_id), schema=validate.Schema(validate.any(
            validate.all(
                {
                    "title": validate.text,
                    "game": validate.text,
                    "channel": validate.all(
                        {"display_name": validate.text},
                        validate.get("display_name")
                    )
                },
                validate.transform(lambda data: (data["channel"], data["title"], data["game"]))
            ),
            validate.all({}, validate.transform(lambda _: (None,) * 3))
        )))

    def metadata_channel(self, channel_id):
        return self.call("/kraken/streams/{0}".format(channel_id), schema=validate.Schema(
            {
                "stream": validate.any(
                    validate.all(
                        {"channel": {
                            "display_name": validate.text,
                            "game": validate.text,
                            "status": validate.text
                        }},
                        validate.get("channel"),
                        validate.transform(lambda ch: (ch["display_name"], ch["status"], ch["game"]))
                    ),
                    validate.all(None, validate.transform(lambda _: (None,) * 3))
                )
            },
            validate.get("stream")
        ))

    # Private API calls

    def access_token(self, endpoint, asset):
        return self.call("/api/{0}/{1}/access_token".format(endpoint, asset), private=True, schema=validate.Schema(
            {
                "token": validate.text,
                "sig": validate.text
            },
            validate.union((
                validate.get("sig"),
                validate.get("token")
            ))
        ), player_type="frontpage")

{'fraction_non_alphanumeric': 0.11308689558514366, 'fraction_numerical': 0.007358093903293623, 'mean_word_length': 2.2539546814878153, 'pattern_counts': {'":': 58, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7637955', 'n_tokens_mistral': 6714, 'n_tokens_neox': 6471, 'n_words': 1353}