Barbagus
db0a954497
- Rename variables and function to reflect model names. - Convert infrastructure data (JSON, M3U8) to model types. - Change algorithms to produce/consume `Source` model, in particular using generator functions to build a list of `Source`s rather than the opaque `rendition => variant => urls` mapping (this will make #7 very straight forward). - Download all master playlists after API call before selecting rendition/variants. Motivation for the last point: We use to offer rendition choosing right after the API call, before we download the appropriate master playlist to figure out the available variants. The problem with that is that ArteTV's codes for the renditions (given by the API) do not necessarily include complete languages information (if it is not French or German), for instance a original audio track in Portuguese would show as `VOEU-` (as in "EUropean"). The actual mention of the Portuguese would only show up in the master playlist. So, the new implementation actually downloads all master playlists straight after the API call. This is a bit wasteful, but I figured it was necessary to provide quality interaction with the user. Bonus? Now when we first prompt the user for rendition choice, we actually already know the available variants available, maybe we make use of that fact in the future...
75 lines
2.0 KiB
Python
75 lines
2.0 KiB
Python
# License: GNU AGPL v3: http://www.gnu.org/licenses/
|
|
# This file is part of `delarte` (https://git.afpy.org/fcode/delarte.git)
|
|
|
|
"""Provide ArteTV JSON API utilities."""
|
|
|
|
from . import common, model
|
|
|
|
MIME_TYPE = "application/vnd.api+json; charset=utf-8"
|
|
|
|
|
|
class UnexpectedResponse(common.UnexpectedError):
|
|
"""Unexpected response from ArteTV."""
|
|
|
|
|
|
class NotFound(common.Error):
|
|
"""Program not found on ArteTV."""
|
|
|
|
|
|
class UnsupportedProtocol(common.Error):
|
|
"""Program type not supported."""
|
|
|
|
|
|
def _fetch_api_data(http_session, path, object_type):
|
|
# Fetch an API object.
|
|
url = "https://api.arte.tv/api/player/v2/" + path
|
|
|
|
r = http_session.get(url)
|
|
if r.status_code == 404:
|
|
raise NotFound(url)
|
|
|
|
r.raise_for_status()
|
|
|
|
if (_ := r.headers["content-type"]) != MIME_TYPE:
|
|
raise UnexpectedResponse("MIME_TYPE", path, MIME_TYPE, _)
|
|
|
|
obj = r.json()["data"]
|
|
|
|
if (_ := obj["type"]) != object_type:
|
|
raise UnexpectedResponse("OBJECT_TYPE", path, object_type, _)
|
|
|
|
return obj["attributes"]
|
|
|
|
|
|
def fetch_program_info(http_session, target_id, www_lang):
|
|
"""Fetch the given target's associated program information."""
|
|
obj = _fetch_api_data(
|
|
http_session, f"config/{www_lang}/{target_id}", "ConfigPlayer"
|
|
)
|
|
|
|
metadata = model.Metadata(
|
|
obj["metadata"]["providerId"],
|
|
obj["metadata"]["title"],
|
|
obj["metadata"]["subtitle"],
|
|
obj["metadata"]["description"],
|
|
obj["metadata"]["duration"]["seconds"],
|
|
)
|
|
|
|
cache = set()
|
|
|
|
for s in obj["streams"]:
|
|
if (_ := s["protocol"]) != "HLS_NG":
|
|
raise UnsupportedProtocol(target_id, www_lang, _)
|
|
|
|
if (master_playlist_url := s["url"]) in cache:
|
|
raise UnexpectedResponse(
|
|
"DUPLICATE_MASTER_PLAYLIST_URL",
|
|
target_id,
|
|
www_lang,
|
|
master_playlist_url,
|
|
)
|
|
|
|
cache.add(master_playlist_url)
|
|
|
|
yield (metadata, master_playlist_url)
|