Spaces:
Running
Running
File size: 2,221 Bytes
1ebbb74 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 |
import datetime
import logging
from cachetools import TTLCache
from .http_utils import download_file_with_retry
from .mpd_utils import parse_mpd, parse_mpd_dict
logger = logging.getLogger(__name__)
# cache dictionary
mpd_cache = TTLCache(maxsize=100, ttl=300) # 5 minutes default TTL
init_segment_cache = TTLCache(maxsize=100, ttl=3600) # 1 hour default TTL
async def get_cached_mpd(
mpd_url: str, headers: dict, parse_drm: bool, parse_segment_profile_id: str | None = None
) -> dict:
"""
Retrieves and caches the MPD manifest, parsing it if not already cached.
Args:
mpd_url (str): The URL of the MPD manifest.
headers (dict): The headers to include in the request.
parse_drm (bool): Whether to parse DRM information.
parse_segment_profile_id (str, optional): The profile ID to parse segments for. Defaults to None.
Returns:
dict: The parsed MPD manifest data.
"""
current_time = datetime.datetime.now(datetime.UTC)
if mpd_url in mpd_cache and mpd_cache[mpd_url]["expires"] > current_time:
logger.info(f"Using cached MPD for {mpd_url}")
return parse_mpd_dict(mpd_cache[mpd_url]["mpd"], mpd_url, parse_drm, parse_segment_profile_id)
mpd_dict = parse_mpd(await download_file_with_retry(mpd_url, headers))
parsed_mpd_dict = parse_mpd_dict(mpd_dict, mpd_url, parse_drm, parse_segment_profile_id)
current_time = datetime.datetime.now(datetime.UTC)
expiration_time = current_time + datetime.timedelta(seconds=parsed_mpd_dict.get("minimumUpdatePeriod", 300))
mpd_cache[mpd_url] = {"mpd": mpd_dict, "expires": expiration_time}
return parsed_mpd_dict
async def get_cached_init_segment(init_url: str, headers: dict) -> bytes:
"""
Retrieves and caches the initialization segment.
Args:
init_url (str): The URL of the initialization segment.
headers (dict): The headers to include in the request.
Returns:
bytes: The initialization segment content.
"""
if init_url not in init_segment_cache:
init_content = await download_file_with_retry(init_url, headers)
init_segment_cache[init_url] = init_content
return init_segment_cache[init_url]
|