|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
import logging |
|
|
import os |
|
|
import sys |
|
|
|
|
|
from .httpclient import DefaultHTTPClient, _parsed_url |
|
|
from .model import (M3U8, Segment, SegmentList, PartialSegment, |
|
|
PartialSegmentList, Key, Playlist, IFramePlaylist, |
|
|
Media, MediaList, PlaylistList, Start, |
|
|
RenditionReport, RenditionReportList, ServerControl, |
|
|
Skip, PartInformation, PreloadHint, DateRange, |
|
|
DateRangeList) |
|
|
from .parser import parse, is_url, ParseError |
|
|
|
|
|
|
|
|
__all__ = ('M3U8', 'Segment', 'SegmentList', 'PartialSegment', |
|
|
'PartialSegmentList', 'Key', 'Playlist', 'IFramePlaylist', |
|
|
'Media', 'MediaList', 'PlaylistList', 'Start', 'RenditionReport', |
|
|
'RenditionReportList', 'ServerControl', 'Skip', 'PartInformation', |
|
|
'PreloadHint' 'DateRange', 'DateRangeList', 'loads', 'load', |
|
|
'parse', 'ParseError') |
|
|
|
|
|
LOGGER = None |
|
|
|
|
|
def loads(content, uri=None, custom_tags_parser=None): |
|
|
''' |
|
|
Given a string with a m3u8 content, returns a M3U8 object. |
|
|
Optionally parses a uri to set a correct base_uri on the M3U8 object. |
|
|
Raises ValueError if invalid content |
|
|
''' |
|
|
global LOGGER |
|
|
if LOGGER is None: |
|
|
LOGGER = logging.getLogger(__name__) |
|
|
if not content.startswith('#EXTM3U'): |
|
|
LOGGER.warning('INVALID m3u format: #EXTM3U missing {}'.format(uri)) |
|
|
return None |
|
|
if uri is None: |
|
|
return M3U8(content, custom_tags_parser=custom_tags_parser) |
|
|
else: |
|
|
base_uri = _parsed_url(uri) |
|
|
return M3U8(content, base_uri=base_uri, custom_tags_parser=custom_tags_parser) |
|
|
|
|
|
|
|
|
def load(uri, timeout=9, headers={}, custom_tags_parser=None, http_client=DefaultHTTPClient(), verify_ssl=True, http_session=None): |
|
|
''' |
|
|
Retrieves the content from a given URI and returns a M3U8 object. |
|
|
Raises ValueError if invalid content or IOError if request fails. |
|
|
''' |
|
|
global LOGGER |
|
|
if LOGGER is None: |
|
|
LOGGER = logging.getLogger(__name__) |
|
|
if is_url(uri): |
|
|
content, base_uri = http_client.download(uri, timeout, headers, verify_ssl, http_session) |
|
|
if content is None: |
|
|
LOGGER.warning('Unable to obtain m3u file {}'.format(uri)) |
|
|
return None |
|
|
if not content.startswith('#EXTM3U'): |
|
|
LOGGER.warning('INVALID m3u format: #EXTM3U missing {}'.format(uri)) |
|
|
return None |
|
|
return M3U8(content, base_uri=base_uri, custom_tags_parser=custom_tags_parser) |
|
|
else: |
|
|
return _load_from_file(uri, custom_tags_parser) |
|
|
|
|
|
|
|
|
def _load_from_file(uri, custom_tags_parser=None): |
|
|
global LOGGER |
|
|
if LOGGER is None: |
|
|
LOGGER = logging.getLogger(__name__) |
|
|
with open(uri, encoding='utf8') as fileobj: |
|
|
raw_content = fileobj.read().strip() |
|
|
base_uri = os.path.dirname(uri) |
|
|
if not raw_content.startswith('#EXTM3U'): |
|
|
LOGGER.warning('INVALID m3u format: #EXTM3U missing {}'.format(uri)) |
|
|
return None |
|
|
return M3U8(raw_content, base_uri=base_uri, custom_tags_parser=custom_tags_parser) |
|
|
|
