from __future__ import annotations
import base64
import json
import re
from typing import TYPE_CHECKING
import httpx
from ..._shared import TTLCache, _copy_docstring
from ._shared import PrivateTIDALResourceAPI
from .pages import PrivatePagesAPI
from .users import PrivateUsersAPI
if TYPE_CHECKING:
from typing import Any
from ...._types import Collection
[docs]
class PrivateVideosAPI(PrivateTIDALResourceAPI):
"""
Videos API endpoints for the private TIDAL API.
.. important::
This class is managed by
:class:`~minim.api.tidal.PrivateTIDALAPIClient` and should not be
instantiated directly.
"""
_M3U_RE = re.compile(
r"#EXT-X-STREAM-INF:(?=[^\n]*BANDWIDTH=(\d+))"
r'(?=[^\n]*CODECS="([^"]+)")[^\n]+\n(\S+)'
)
_VIDEO_QUALITIES = {"AUDIO_ONLY", "LOW", "MEDIUM", "HIGH"}
__slots__ = ()
def _download_video_stream(
self, manifest: bytes | str, /
) -> tuple[str, bytes]:
"""
Download the video stream data for a music video.
Parameters
----------
manifest : bytes or str; positional-only
Metadata for the video's source files.
Returns
-------
codec : str
Video codec.
stream : bytes
Video stream data.
"""
self._validate_type("manifest", manifest, bytes | str)
if isinstance(manifest, str):
manifest = base64.b64decode(manifest)
if manifest[0] == 123: # JSON
_, codec, m3u = max(
self._M3U_RE.findall(
httpx.get(json.loads(manifest)["urls"][0]).text
),
key=lambda m3u: int(m3u[0]),
)
return codec, b"".join(
httpx.get(ts).content
for ts in re.compile("(?<=\n).*(http.*)").findall(
httpx.get(m3u).text
)
)
elif manifest[0] == 60: # XML (audio-only)
return self._client.tracks._download_track_stream(manifest)
else:
raise ValueError(
"`manifest`, when decoded, is not in the JSON format "
"or XML format."
)
[docs]
@TTLCache.cached_method(ttl="popularity")
def get_video(
self, video_id: int | str, /, country_code: str | None = None
) -> dict[str, Any]:
"""
Get TIDAL catalog information for a video.
Parameters
----------
video_id : int or str; positional-only
TIDAL ID of the video.
**Examples**: :code:`29597422`, :code:`"59727844"`.
country_code : str; optional
ISO 3166-1 alpha-2 country code. If not provided, the
country associated with the current user account or IP
address is used.
**Example**: :code:`"US"`.
Returns
-------
video : dict[str, Any]
TIDAL metadata for the video.
.. admonition:: Sample response
:class: response dropdown
.. code-block::
{
"adSupportedStreamReady": <bool>,
"adsPrePaywallOnly": <bool>,
"adsUrl": <str>,
"album": {},
"allowStreaming": <bool>,
"artist": {
"handle": <str>,
"id": <int>,
"name": <str>,
"picture": <str>,
"type": "MAIN"
},
"artists": [
{
"handle": <str>,
"id": <int>,
"name": <str>,
"picture": <str>,
"type": <str>
}
],
"djReady": <bool>,
"duration": <int>,
"explicit": <bool>,
"id": <int>,
"imageId": <str>,
"imagePath": <str>,
"popularity": <int>,
"quality": <str>,
"releaseDate": <str>,
"stemReady": <bool>,
"streamReady": <bool>,
"streamStartDate": <str>,
"title": <str>,
"videoNumber": <int>,
"type": <str>,
"vibrantColor": <str>,
"volumeNumber": <int>
}
"""
return self._get_resource(
"videos", video_id, country_code=country_code
)
[docs]
@TTLCache.cached_method(ttl="static")
def get_video_contributors(
self,
video_id: int | str,
/,
country_code: str | None = None,
*,
limit: int | None = None,
offset: int | None = None,
) -> dict[str, Any]:
"""
Get TIDAL catalog information for contributors to a video.
Parameters
----------
video_id : int or str; positional-only
TIDAL ID of the video.
**Examples**: :code:`29597422`, :code:`"59727844"`.
country_code : str; optional
ISO 3166-1 alpha-2 country code. If not provided, the
country associated with the current user account or IP
address is used.
**Example**: :code:`"US"`.
limit : int; keyword-only; optional
Maximum number of videos to return.
**Valid range**: :code:`1` to :code:`100`.
**API default**: :code:`10`.
offset : int; keyword-only; optional
Index of the first video to return. Use with `limit` to get
the next batch of videos.
**Minimum value**: :code:`0`.
**API default**: :code:`0`.
Returns
-------
contributors : dict[str, Any]
Page of TIDAL metadata for the video's contributors.
.. admonition:: Sample response
:class: response dropdown
.. code-block::
{
"items": [
{
"name": <str>,
"role": <str>
}
],
"limit": <int>,
"offset": <int>,
"totalNumberOfItems": <int>
}
"""
return self._get_resource_relationship(
"videos",
video_id,
"contributors",
country_code=country_code,
limit=limit,
offset=offset,
)
[docs]
@_copy_docstring(PrivatePagesAPI.get_video_page)
def get_video_page(
self,
video_id: int | str,
/,
country_code: str | None = None,
*,
device_type: str = "BROWSER",
locale: str | None = None,
) -> dict[str, Any]:
return self._client.pages.get_video_page(
video_id,
country_code=country_code,
device_type=device_type,
locale=locale,
)
[docs]
@_copy_docstring(PrivateUsersAPI.get_user_saved_videos)
def get_user_saved_videos(
self,
user_id: int | str | None = None,
/,
country_code: str | None = None,
*,
limit: int | None = None,
offset: int | None = None,
sort_by: str | None = None,
descending: bool | None = None,
) -> dict[str, Any]:
return self._client.users.get_user_saved_videos(
user_id,
country_code=country_code,
limit=limit,
offset=offset,
sort_by=sort_by,
descending=descending,
)
[docs]
@_copy_docstring(PrivateUsersAPI.save_videos)
def save_videos(
self,
video_ids: int | str | Collection[int | str],
/,
user_id: int | str | None = None,
country_code: str | None = None,
*,
on_missing: str | None = None,
) -> None:
self._client.users.save_videos(
video_ids,
user_id=user_id,
country_code=country_code,
on_missing=on_missing,
)
[docs]
@_copy_docstring(PrivateUsersAPI.remove_saved_videos)
def remove_saved_videos(
self,
video_ids: int | str | Collection[int | str],
/,
user_id: int | str | None = None,
) -> None:
self._client.users.remove_saved_videos(video_ids, user_id=user_id)
[docs]
@_copy_docstring(PrivateUsersAPI.get_user_blocked_videos)
def get_user_blocked_videos(
self,
user_id: int | str | None = None,
/,
*,
limit: int | None = None,
offset: int | None = None,
) -> dict[str, Any]:
return self._client.users.get_user_blocked_videos(
user_id, limit=limit, offset=offset
)
[docs]
@_copy_docstring(PrivateUsersAPI.block_video)
def block_video(
self, video_id: int | str, /, user_id: int | str | None = None
) -> None:
self._client.users.block_video(video_id, user_id=user_id)
[docs]
@_copy_docstring(PrivateUsersAPI.unblock_video)
def unblock_video(
self, video_id: int | str, /, user_id: int | str | None = None
) -> None:
self._client.users.unblock_video(video_id, user_id=user_id)
