<commit_before>from . import Stream, StreamError
from ..utils import urlget, RingBuffer, absolute_url
from ..compat import urljoin, queue

from time import time, sleep
from threading import Lock, Thread, Timer

import io
import os.path
import re

try:
    from Crypto.Cipher import AES
    import struct

    def num_to_iv(n):
        return struct.pack(">8xq", n)

    CAN_DECRYPT = True

except ImportError:
    CAN_DECRYPT = False


def parse_m3u_attributes(data):
    attr = re.findall("([A-Z\-]+)=(\d+\.\d+|0x[0-9A-z]+|\d+x\d+|\d+|\"(.+?)\"|[0-9A-z\-]+)", data)
    rval = {}

    for key, val, strval in attr:
        if len(strval) > 0:
            rval[key] = strval
        else:
            rval[key] = val

    if len(rval) > 0:
        return rval

    return data

def parse_m3u_tag(data):
    key = data[1:]
    value = None
    valpos = data.find(":")

    if valpos > 0:
        key = data[1:valpos]
        value = data[valpos+1:]

    return (key, value)

def parse_m3u(data):
    lines = [line for line in data.splitlines() if len(line) > 0]
    tags = {}
    entries = []

    lasttag = None

    for i, line in enumerate(lines):
        if line.startswith("#EXT"):
            (key, value) = parse_m3u_tag(line)

            if value is not None:
                if key == "EXTINF":
                    duration, title = value.split(",")
                    value = (float(duration), title)
                else:
                    value = parse_m3u_attributes(value)

            if key in tags:
                tags[key].append(value)
            else:
                tags[key] = [value]

            lasttag = (key, value)
        else:
            entry = { "url": line, "tag": lasttag }
            entries.append(entry)

    return (tags, entries)


class HLSStreamFiller(Thread):
    def __init__(self, stream):
        Thread.__init__(self)

        self.daemon = True
        self.queue = queue.Queue()
        self.stream = stream

    def download_sequence(self, entry):
        try:
            res = urlget(entry["url"], stream=True,
                         exception=IOError)
        except IOError as err:
            self.stream.logger.error("Failed to open sequence {0}: {1}",
                                     entry["sequence"], str(err))
            return

        if self.stream.decryptor_key:
            iv = num_to_iv(entry["sequence"])
            decryptor = AES.new(self.stream.decryptor_key, AES.MODE_CBC, iv)
        else:
            decryptor = None

        while self.running:
            try:
                chunk = res.raw.read(8192)
            except IOError as err:
                self.stream.logger.error("Failed to read sequence {0}: {1}",
                                         entry["sequence"], str(err))
                break

            if len(chunk) == 0:
                self.stream.logger.debug("Download of sequence {0} complete", entry["sequence"])
                break

            if decryptor:
                chunk = decryptor.decrypt(chunk)

            self.stream.buffer.write(chunk)

    def run(self):
        self.stream.logger.debug("Starting buffer filler thread")
        self.running = True

        while self.running:
            entry = self.queue.get()
            self.download_sequence(entry)

            if entry["sequence"] == self.stream.playlist_end:
                break

        if self.stream.playlist_timer:
            self.stream.playlist_timer.cancel()

        self.stream.buffer.close()
        self.stream.logger.debug("Buffer filler thread completed")

class HLSStreamIO(io.IOBase):
    def __init__(self, session, url, timeout=60):
        self.session = session
        self.url = url
        self.timeout = timeout

        self.logger = session.logger.new_module("stream.hls")
        self.buffer = None

    def open(self):
        self.playlist_end = None
        self.playlist_lock = Lock()<commit_msg>stream.hls: Avoid potential dead lock.<commit_after>from . import Stream, StreamError
from ..utils import urlget, RingBuffer, absolute_url
from ..compat import urljoin, queue

from time import time, sleep
from threading import Lock, Thread, Timer

import io
import os.path
import re

try:
    from Crypto.Cipher import AES
    import struct

    def num_to_iv(n):
        return struct.pack(">8xq", n)

    CAN_DECRYPT = True

except ImportError:
    CAN_DECRYPT = False


def parse_m3u_attributes(data):
    attr = re.findall("([A-Z\-]+)=(\d+\.\d+|0x[0-9A-z]+|\d+x\d+|\d+|\"(.+?)\"|[0-9A-z\-]+)", data)
    rval = {}

    for key, val, strval in attr:
        if len(strval) > 0:
            rval[key] = strval
        else:
            rval[key] = val

    if len(rval) > 0:
        return rval

    return data

def parse_m3u_tag(data):
    key = data[1:]
    value = None
    valpos = data.find(":")

    if valpos > 0:
        key = data[1:valpos]
        value = data[valpos+1:]

    return (key, value)

def parse_m3u(data):
    lines = [line for line in data.splitlines() if len(line) > 0]
    tags = {}
    entries = []

    lasttag = None

    for i, line in enumerate(lines):
        if line.startswith("#EXT"):
            (key, value) = parse_m3u_tag(line)

            if value is not None:
                if key == "EXTINF":
                    duration, title = value.split(",")
                    value = (float(duration), title)
                else:
                    value = parse_m3u_attributes(value)

            if key in tags:
                tags[key].append(value)
            else:
                tags[key] = [value]

            lasttag = (key, value)
        else:
            entry = { "url": line, "tag": lasttag }
            entries.append(entry)

    return (tags, entries)


class HLSStreamFiller(Thread):
    def __init__(self, stream):
        Thread.__init__(self)

        self.daemon = True
        self.queue = queue.Queue()
        self.stream = stream

    def download_sequence(self, entry):
        try:
            res = urlget(entry["url"], stream=True,
                         exception=IOError)
        except IOError as err:
            self.stream.logger.error("Failed to open sequence {0}: {1}",
                                     entry["sequence"], str(err))
            return

        if self.stream.decryptor_key:
            iv = num_to_iv(entry["sequence"])
            decryptor = AES.new(self.stream.decryptor_key, AES.MODE_CBC, iv)
        else:
            decryptor = None

        while self.running:
            try:
                chunk = res.raw.read(8192)
            except IOError as err:
                self.stream.logger.error("Failed to read sequence {0}: {1}",
                                         entry["sequence"], str(err))
                break

            if len(chunk) == 0:
                self.stream.logger.debug("Download of sequence {0} complete", entry["sequence"])
                break

            if decryptor:
                chunk = decryptor.decrypt(chunk)

            self.stream.buffer.write(chunk)

    def run(self):
        self.stream.logger.debug("Starting buffer filler thread")
        self.running = True

        while self.running:
            try:
                entry = self.queue.get(True, 5)
            except queue.Empty:
                continue

            self.download_sequence(entry)

            if entry["sequence"] == self.stream.playlist_end:
                break

        if self.stream.playlist_timer:
            self.stream.playlist_timer.cancel()

        self.stream.buffer.close()
        self.stream.logger.debug("Buffer filler thread completed")

class HLSStreamIO(io.IOBase):
    def __init__(self, session, url, timeout=60):
        self.session = session
        self.url = url
        self.timeout = timeout

        self.logger = session.logger.new_module("stream.hls")
        self.buffer = None

    def open(self):
        self.playlist_end = None
        self.playlist_lock = Lock()
{'fraction_non_alphanumeric': 0.11532187225498808, 'fraction_numerical': 0.007654661814531309, 'mean_word_length': 1.718281036834925, 'pattern_counts': {'":': 8, '<': 3, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3805750', 'n_tokens_mistral': 2449, 'n_tokens_neox': 2306, 'n_words': 574}