Module hdrezka.stream.player

General user-friendly interface to HDRezka streams.

Functions

async def Player(url_or_path: Any, *, client: HDRezkaClient) ‑> PlayerMovie | PlayerSeries

Expand source code

async def player(url_or_path: Any, *, client: 'HDRezkaClient') -> PlayerMovie | PlayerSeries:
    """
    Return ``PlayerSeries`` for series or ``PlayerMovie`` for a movie.

    Raises ``UnknownContentType`` for unsupported post types.
    """
    cast = PlayerBase(url_or_path, client=client)
    cached = client.player_cache.get(cast.post.url)
    if cached is not None:
        return cached
    await cast
    value: PlayerMovie | PlayerSeries
    match cast.post.type:
        case 'tv_series':
            value = PlayerSeries(cast, client=client)
        case 'movie':
            value = PlayerMovie(cast, client=client)
        case _ as e:
            raise UnknownContentType(e)
    client.player_cache.set(cast.post.url, value)
    return value

Return PlayerSeries for series or PlayerMovie for a movie.

Raises UnknownContentType for unsupported post types.

Classes

class PlayerBase (url_or_cast: Any, *, client: HDRezkaClient)

Expand source code

class PlayerBase:
    """Base type of Player."""
    __slots__ = ('post', 'client')

    def __init__(self, url_or_cast: Any, *, client: 'HDRezkaClient'):
        """
        Need await.

        :param url_or_cast: Post URL/path, or another ``PlayerBase`` to copy.
        :param client: Session used for HTTP and AJAX requests.
        """
        self.client = client
        if isinstance(url_or_cast, PlayerBase):
            self.post: Post = url_or_cast.post
            return
        elif not isinstance(url_or_cast, str):
            url_or_cast = str(url_or_cast)
        self.post = Post(url_or_cast, client=client)

    def __await__(self):
        """
        Initialize ``self.post``.
        Do not call twice!
        """
        yield from self.post.__await__()
        return self

    async def get_trailer_iframe(self) -> str:
        """Get trailer ``<iframe>`` HTML."""
        return (await self.client.ajax.get_trailer_video(self.post.id)).get('code', '')

    def _translator(self, translator_id: Optional[SupportsInt] = None) -> int:
        if translator_id is None:
            return self.post.translator_id
        translator_id = int(translator_id)
        return self.post.translators.ids[abs(translator_id)] if translator_id <= 0 else translator_id

    def __repr__(self):
        return f'{self.__class__.__qualname__}({self.post.url!r})'

Base type of Player.

Need await.

:param url_or_cast: Post URL/path, or another PlayerBase to copy. :param client: Session used for HTTP and AJAX requests.

Subclasses

• PlayerMovie
• PlayerSeries

Instance variables

var client

Expand source code

class PlayerBase:
    """Base type of Player."""
    __slots__ = ('post', 'client')

    def __init__(self, url_or_cast: Any, *, client: 'HDRezkaClient'):
        """
        Need await.

        :param url_or_cast: Post URL/path, or another ``PlayerBase`` to copy.
        :param client: Session used for HTTP and AJAX requests.
        """
        self.client = client
        if isinstance(url_or_cast, PlayerBase):
            self.post: Post = url_or_cast.post
            return
        elif not isinstance(url_or_cast, str):
            url_or_cast = str(url_or_cast)
        self.post = Post(url_or_cast, client=client)

    def __await__(self):
        """
        Initialize ``self.post``.
        Do not call twice!
        """
        yield from self.post.__await__()
        return self

    async def get_trailer_iframe(self) -> str:
        """Get trailer ``<iframe>`` HTML."""
        return (await self.client.ajax.get_trailer_video(self.post.id)).get('code', '')

    def _translator(self, translator_id: Optional[SupportsInt] = None) -> int:
        if translator_id is None:
            return self.post.translator_id
        translator_id = int(translator_id)
        return self.post.translators.ids[abs(translator_id)] if translator_id <= 0 else translator_id

    def __repr__(self):
        return f'{self.__class__.__qualname__}({self.post.url!r})'

var post

Expand source code

class PlayerBase:
    """Base type of Player."""
    __slots__ = ('post', 'client')

    def __init__(self, url_or_cast: Any, *, client: 'HDRezkaClient'):
        """
        Need await.

        :param url_or_cast: Post URL/path, or another ``PlayerBase`` to copy.
        :param client: Session used for HTTP and AJAX requests.
        """
        self.client = client
        if isinstance(url_or_cast, PlayerBase):
            self.post: Post = url_or_cast.post
            return
        elif not isinstance(url_or_cast, str):
            url_or_cast = str(url_or_cast)
        self.post = Post(url_or_cast, client=client)

    def __await__(self):
        """
        Initialize ``self.post``.
        Do not call twice!
        """
        yield from self.post.__await__()
        return self

    async def get_trailer_iframe(self) -> str:
        """Get trailer ``<iframe>`` HTML."""
        return (await self.client.ajax.get_trailer_video(self.post.id)).get('code', '')

    def _translator(self, translator_id: Optional[SupportsInt] = None) -> int:
        if translator_id is None:
            return self.post.translator_id
        translator_id = int(translator_id)
        return self.post.translators.ids[abs(translator_id)] if translator_id <= 0 else translator_id

    def __repr__(self):
        return f'{self.__class__.__qualname__}({self.post.url!r})'

Methods

async def get_trailer_iframe(self) ‑> str

Expand source code

async def get_trailer_iframe(self) -> str:
    """Get trailer ``<iframe>`` HTML."""
    return (await self.client.ajax.get_trailer_video(self.post.id)).get('code', '')

Get trailer <iframe> HTML.

class PlayerMovie (url_or_cast: Any, *, client: HDRezkaClient)

Expand source code

class PlayerMovie(PlayerBase):
    """Movies player type."""
    __slots__ = ()

    async def get_stream(self, translator_id: Optional[SupportsInt] = None) -> URLs:
        """Return movie stream ``URLs``."""
        return urls_from_ajax_response(
            await self.client.ajax.get_movie(self.post.id, self._translator(translator_id)),
            client=self.client,
        )

Movies player type.

Need await.

:param url_or_cast: Post URL/path, or another PlayerBase to copy. :param client: Session used for HTTP and AJAX requests.

Ancestors

• PlayerBase

Methods

async def get_stream(self, translator_id:  | None = None) ‑> URLs

Expand source code

async def get_stream(self, translator_id: Optional[SupportsInt] = None) -> URLs:
    """Return movie stream ``URLs``."""
    return urls_from_ajax_response(
        await self.client.ajax.get_movie(self.post.id, self._translator(translator_id)),
        client=self.client,
    )

Return movie stream URLs.

Inherited members

• PlayerBase:
  • get_trailer_iframe

class PlayerSeries (url_or_cast: Any, *, client: HDRezkaClient)

Expand source code

class PlayerSeries(PlayerBase):
    """TV series player type."""
    __slots__ = ()

    async def get_episodes(self, translator_id: Optional[SupportsInt] = None) -> defaultdict[int, tuple[int, ...]]:
        """Return available episodes grouped by season."""
        episodes = BeautifulSoup(
            (await self.client.ajax.get_episodes(self.post.id, self._translator(translator_id)))['episodes'],
            builder=BUILDER,
        )
        result: defaultdict[int, tuple[int, ...]] = defaultdict(tuple)
        for i in episodes.find_all(class_='b-simple_episode__item', attrs=('data-season_id', 'data-episode_id')):
            result[int(i.attrs.get('data-season_id'))] += int(i.attrs.get('data-episode_id', '0')),
        return result

    async def get_stream(self, season: int, episode: int, translator_id: Optional[SupportsInt] = None) -> URLs:
        """Return episode stream ``URLs``."""
        return urls_from_ajax_response(
            await self.client.ajax.get_stream(self.post.id, self._translator(translator_id), season, episode),
            client=self.client,
        )

TV series player type.

Need await.

:param url_or_cast: Post URL/path, or another PlayerBase to copy. :param client: Session used for HTTP and AJAX requests.

Ancestors

• PlayerBase

Methods

async def get_episodes(self, translator_id:  | None = None) ‑> collections.defaultdict[int, tuple[int, ...]]

Expand source code

async def get_episodes(self, translator_id: Optional[SupportsInt] = None) -> defaultdict[int, tuple[int, ...]]:
    """Return available episodes grouped by season."""
    episodes = BeautifulSoup(
        (await self.client.ajax.get_episodes(self.post.id, self._translator(translator_id)))['episodes'],
        builder=BUILDER,
    )
    result: defaultdict[int, tuple[int, ...]] = defaultdict(tuple)
    for i in episodes.find_all(class_='b-simple_episode__item', attrs=('data-season_id', 'data-episode_id')):
        result[int(i.attrs.get('data-season_id'))] += int(i.attrs.get('data-episode_id', '0')),
    return result

Return available episodes grouped by season.

async def get_stream(self,
season: int,
episode: int,
translator_id:  | None = None) ‑> URLs

Expand source code

async def get_stream(self, season: int, episode: int, translator_id: Optional[SupportsInt] = None) -> URLs:
    """Return episode stream ``URLs``."""
    return urls_from_ajax_response(
        await self.client.ajax.get_stream(self.post.id, self._translator(translator_id), season, episode),
        client=self.client,
    )

Return episode stream URLs.

Inherited members

• PlayerBase:
  • get_trailer_iframe