Module hdrezka.api

HDRezka API helpers (search and AJAX).

Sub-modules

hdrezka.api.ajax

Sending AJAX requests.

hdrezka.api.http

HTTP request typing helpers and browser-like session factories.

hdrezka.api.search

Implementation of search on HDRezka pages (see post.page.Page).

hdrezka.api.types

API response types.

Classes

class AJAX (client: HDRezkaClient)

Expand source code

class AJAX:
    """AJAX endpoints bound to a single ``HDRezkaClient`` session."""

    __slots__ = ('_client',)

    def __init__(self, client: 'HDRezkaClient'):
        """
        :param client: Session used for all AJAX requests.
        """
        self._client = client

    @staticmethod
    def _check(resp: httpx.Response):
        answer = resp.json()
        if not answer.get('success', True):
            raise AJAXFail(answer.get('message', answer))
        return answer

    async def _send_data(self, action: str, **kwargs):
        return self._check(
            await self._client.get_response('POST', self._client.host_join(f'ajax/{action}/'), **kwargs)
        )

    async def get_cdn_series(self, data: dict[str, SupportsInt | str]):
        """Request series payload for one translation."""
        return await self._send_data('get_cdn_series', data=data)

    async def get_episodes(self, id_: SupportsInt | str, translator_id: SupportsInt | str):
        """Request available episodes."""
        return await self.get_cdn_series(
            {'action': 'get_episodes',
             'id': id_,
             'translator_id': translator_id}
        )

    async def get_stream(self, id_: SupportsInt | str, translator_id: SupportsInt | str,
                         season: SupportsInt | str, episode: SupportsInt | str) -> APIResponse:
        """Request stream URLs for an episode."""
        return await self.get_cdn_series(
            {'action': 'get_stream',
             'id': id_,
             'translator_id': translator_id,
             'season': season,
             'episode': episode}
        )

    async def get_movie(self, id_: SupportsInt | str, translator_id: SupportsInt | str):
        """Request stream URLs for a movie."""
        return await self.get_cdn_series(
            {'action': 'get_movie',
             'id': id_, 'translator_id': translator_id}
        )

    async def get_trailer_video(self, id_: SupportsInt | str) -> TrailerResponse:
        """Request trailer video iframe HTML."""
        return self._check(await self._client.get_response(
            'POST',
            self._client.host_join('engine/ajax/gettrailervideo.php'),
            data={'id': id_},
        ))

    async def _favorites(self, **data) -> dict:
        """POST form data to ``/ajax/favorites/``."""
        return self._check(await self._client.get_response(
            'POST',
            self._client.host_join('ajax/favorites/'),
            data=data,
        ))

    async def add_favorites_cat(self, name: str) -> FavoritesCatResponse:
        """
        Create a favorites collection in the user profile.

        Equivalent to ``action=add_cat``.
        """
        return await self._favorites(name=name, action='add_cat')

    async def rename_favorites_cat(
            self,
            cat_id: SupportsInt | str,
            name: str,
    ) -> FavoritesOkResponse:
        """
        Rename a favorites collection.

        Equivalent to ``action=change_cat_name``.
        """
        return await self._favorites(name=name, cat_id=cat_id, action='change_cat_name')

    async def add_favorites_post(
            self,
            post_id: SupportsInt | str,
            cat_id: SupportsInt | str,
    ) -> FavoritesOkResponse:
        """
        Add a title (post) to a favorites collection.

        Equivalent to ``action=add_post``.
        """
        return await self._favorites(post_id=post_id, cat_id=cat_id, action='add_post')

    async def remove_favorites_cat(self, cat_id: SupportsInt | str) -> FavoritesOkResponse:
        """
        Remove a favorites collection from the profile.

        Equivalent to ``action=remove_cat``.
        """
        return await self._favorites(cat_id=cat_id, action='remove_cat')

AJAX endpoints bound to a single HDRezkaClient session.

:param client: Session used for all AJAX requests.

Methods

async def add_favorites_cat(self, name: str) ‑> FavoritesCatResponse

Expand source code

async def add_favorites_cat(self, name: str) -> FavoritesCatResponse:
    """
    Create a favorites collection in the user profile.

    Equivalent to ``action=add_cat``.
    """
    return await self._favorites(name=name, action='add_cat')

Create a favorites collection in the user profile.

Equivalent to action=add_cat.

async def add_favorites_post(self, post_id: typing.SupportsInt | str, cat_id: typing.SupportsInt | str) ‑> FavoritesOkResponse

Expand source code

async def add_favorites_post(
        self,
        post_id: SupportsInt | str,
        cat_id: SupportsInt | str,
) -> FavoritesOkResponse:
    """
    Add a title (post) to a favorites collection.

    Equivalent to ``action=add_post``.
    """
    return await self._favorites(post_id=post_id, cat_id=cat_id, action='add_post')

Add a title (post) to a favorites collection.

Equivalent to action=add_post.

async def get_cdn_series(self, data: dict[str, typing.SupportsInt | str])

Expand source code

async def get_cdn_series(self, data: dict[str, SupportsInt | str]):
    """Request series payload for one translation."""
    return await self._send_data('get_cdn_series', data=data)

Request series payload for one translation.

async def get_episodes(self, id_: typing.SupportsInt | str, translator_id: typing.SupportsInt | str)

Expand source code

async def get_episodes(self, id_: SupportsInt | str, translator_id: SupportsInt | str):
    """Request available episodes."""
    return await self.get_cdn_series(
        {'action': 'get_episodes',
         'id': id_,
         'translator_id': translator_id}
    )

Request available episodes.

async def get_movie(self, id_: typing.SupportsInt | str, translator_id: typing.SupportsInt | str)

Expand source code

async def get_movie(self, id_: SupportsInt | str, translator_id: SupportsInt | str):
    """Request stream URLs for a movie."""
    return await self.get_cdn_series(
        {'action': 'get_movie',
         'id': id_, 'translator_id': translator_id}
    )

Request stream URLs for a movie.

async def get_stream(self,
id_: typing.SupportsInt | str,
translator_id: typing.SupportsInt | str,
season: typing.SupportsInt | str,
episode: typing.SupportsInt | str) ‑> APIResponse

Expand source code

async def get_stream(self, id_: SupportsInt | str, translator_id: SupportsInt | str,
                     season: SupportsInt | str, episode: SupportsInt | str) -> APIResponse:
    """Request stream URLs for an episode."""
    return await self.get_cdn_series(
        {'action': 'get_stream',
         'id': id_,
         'translator_id': translator_id,
         'season': season,
         'episode': episode}
    )

Request stream URLs for an episode.

async def get_trailer_video(self, id_: typing.SupportsInt | str) ‑> TrailerResponse

Expand source code

async def get_trailer_video(self, id_: SupportsInt | str) -> TrailerResponse:
    """Request trailer video iframe HTML."""
    return self._check(await self._client.get_response(
        'POST',
        self._client.host_join('engine/ajax/gettrailervideo.php'),
        data={'id': id_},
    ))

Request trailer video iframe HTML.

async def remove_favorites_cat(self, cat_id: typing.SupportsInt | str) ‑> FavoritesOkResponse

Expand source code

async def remove_favorites_cat(self, cat_id: SupportsInt | str) -> FavoritesOkResponse:
    """
    Remove a favorites collection from the profile.

    Equivalent to ``action=remove_cat``.
    """
    return await self._favorites(cat_id=cat_id, action='remove_cat')

Remove a favorites collection from the profile.

Equivalent to action=remove_cat.

async def rename_favorites_cat(self, cat_id: typing.SupportsInt | str, name: str) ‑> FavoritesOkResponse

Expand source code

async def rename_favorites_cat(
        self,
        cat_id: SupportsInt | str,
        name: str,
) -> FavoritesOkResponse:
    """
    Rename a favorites collection.

    Equivalent to ``action=change_cat_name``.
    """
    return await self._favorites(name=name, cat_id=cat_id, action='change_cat_name')

Rename a favorites collection.

Equivalent to action=change_cat_name.

class Search (query: str = '', *, client: HDRezkaClient)

Expand source code

class Search(Page):
    """HDRezka search listing bound to a client session."""
    __slots__ = ('_query',)

    def __init__(self, query: str = '', *, client: 'HDRezkaClient'):
        """
        :param query: Search query string.
        :param client: Session used for HTTP requests and host resolution.
        """
        self._query = (query if isinstance(query, str) else str(query)).strip()
        super().__init__(self._search_url(client, self._query), client=client)

    @staticmethod
    def _search_url(client: 'HDRezkaClient', query: str) -> str:
        return client.host_join(f'search/?do=search&subaction=search&q={query}')

    def search_url(self, query: str) -> str:
        """Return the search URL for ``query`` on this client's host."""
        return self._search_url(self.client, query)

    @property
    def query(self) -> str:
        """Current search query."""
        return self._query

    @query.setter
    def query(self, value):
        """Cast value to str and refresh the page URL."""
        self._query = value if isinstance(value, str) else str(value)
        self.page = self.search_url(self.query)

    @staticmethod
    @override
    def _concat_paginator(url: str) -> str:
        return f'{url}&page={{0}}'

    def __repr__(self):
        return f"{self.__class__.__qualname__}({repr(self.query) if self.query else ''})"

HDRezka search listing bound to a client session.

:param query: Search query string. :param client: Session used for HTTP requests and host resolution.

Ancestors

• Page

Instance variables

prop query : str

Expand source code

@property
def query(self) -> str:
    """Current search query."""
    return self._query

Current search query.

Methods

def search_url(self, query: str) ‑> str

Expand source code

def search_url(self, query: str) -> str:
    """Return the search URL for ``query`` on this client's host."""
    return self._search_url(self.client, query)

Return the search URL for query on this client's host.

Inherited members

• Page:
  • get_favorites_cats
  • get_filters
  • get_navbar
  • get_page
  • get_page_content
  • get_series_updates
  • page