Package hdrezka

HDRezka asynchronous Python client library.

Use HDRezkaClient as the entry point for per-user sessions (suitable for building multi-user APIs and mobile backends). See the project README for usage and migration notes from 4.x.

Sub-modules

hdrezka.api

HDRezka API helpers (search and AJAX).

hdrezka.client

Per-user HDRezka session client.

hdrezka.errors

Error definitions

hdrezka.post

Post, catalog pages, site chrome, and stream URL types.

hdrezka.stream

HDRezka stream module

hdrezka.translators

Translators definition module

hdrezka.url

Default HDRezka host and redirect URL constants.

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.

def parse_favorites_cats(html: str | bs4.BeautifulSoup | bs4.element.Tag) ‑> tuple[FavoritesCat, ...]
Expand source code
def parse_favorites_cats(html: str | BeautifulSoup | Tag) -> tuple[FavoritesCat, ...]:
    """
    Parse ``a.b-favorites_content__cats_list_link`` entries.

    Returns an empty tuple when the list is missing.
    """
    soup = as_soup(html)
    cats = []
    for a in soup.select('a.b-favorites_content__cats_list_link'):
        href = attr(a, 'href', default='') or ''
        cats.append(FavoritesCat(
            name=text(a.select_one('.name')),
            url=href,
            count=get_any_int(text(a.select_one('.num-holder'))),
            id=_cat_id_from_url(href),
        ))
    return tuple(cats)

Parse a.b-favorites_content__cats_list_link entries.

Returns an empty tuple when the list is missing.

def parse_navbar(html: str | bs4.BeautifulSoup | bs4.element.Tag) ‑> Navbar
Expand source code
def parse_navbar(html: str | BeautifulSoup | Tag) -> Navbar:
    """
    Parse ``ul#topnav-menu`` into a ``Navbar``.

    Returns an empty ``Navbar`` when the menu is missing.
    """
    soup = as_soup(html)
    menu = soup.select_one('ul#topnav-menu.b-topnav__inner') or soup.select_one('ul#topnav-menu')
    if menu is None:
        return Navbar()
    items = []
    for li in menu.find_all('li', class_='b-topnav__item', recursive=False):
        classes = li.get('class') or []
        is_single = 'single' in classes
        link = li.select_one('a.b-topnav__item-link')
        if link is None:
            continue
        items.append(NavItem(
            name=text(link),
            url=attr(link, 'href', default='') or '',
            single=is_single,
            submenu=None if is_single else _parse_submenu(li),
        ))
    return Navbar(items=tuple(items))

Parse ul#topnav-menu into a Navbar.

Returns an empty Navbar when the menu is missing.

def parse_page_filters(html: str | bs4.BeautifulSoup | bs4.element.Tag) ‑> PageFilters
Expand source code
def parse_page_filters(html: str | BeautifulSoup | Tag) -> PageFilters:
    """
    Parse ``ul.b-content__main_filters`` into ``PageFilters``.

    Returns empty tuples when the block is absent.
    """
    soup = as_soup(html)
    root = soup.select_one('ul.b-content__main_filters')
    if root is None:
        return PageFilters()
    sorts = tuple(
        _link(a, param='filter')
        for a in root.select('li.b-content__main_filters_item > a.b-content__main_filters_link')
    )
    types = tuple(
        _link(a, param='genre')
        for a in root.select('li.b-content__filters_types a.filter-link')
    )
    return PageFilters(sorts=sorts, types=types)

Parse ul.b-content__main_filters into PageFilters.

Returns empty tuples when the block is absent.

def parse_schedule(html: str | bs4.BeautifulSoup | bs4.element.Tag) ‑> tuple[ScheduleBlock, ...]
Expand source code
def parse_schedule(html: str | BeautifulSoup | Tag) -> tuple[ScheduleBlock, ...]:
    """
    Parse ``.b-post__schedule_block`` tables from a title page.

    Returns an empty tuple when the schedule markup is missing.
    """
    soup = as_soup(html)
    root = soup.select_one('.b-post__schedule') or soup
    blocks = []
    for block in root.select('.b-post__schedule_block'):
        title = text(block.select_one('.b-post__schedule_block_title .title'))
        items = []
        for tr in block.select('table.b-post__schedule_table tr'):
            item = _parse_row(tr)
            if item is not None:
                items.append(item)
        blocks.append(ScheduleBlock(title=title, items=tuple(items)))
    return tuple(blocks)

Parse .b-post__schedule_block tables from a title page.

Returns an empty tuple when the schedule markup is missing.

def parse_series_updates(html: str | bs4.BeautifulSoup | bs4.element.Tag) ‑> tuple[SeriesUpdateBlock, ...]
Expand source code
def parse_series_updates(html: str | BeautifulSoup | Tag) -> tuple[SeriesUpdateBlock, ...]:
    """
    Parse ``.b-seriesupdate__block`` entries from the inline sidebar.

    Returns an empty tuple when the sidebar feed is missing.
    """
    soup = as_soup(html)
    sidebar = soup.select_one('.b-content__inline_sidebar') or soup
    blocks = []
    for block in sidebar.select('.b-seriesupdate__block'):
        date = text(block.select_one('.b-seriesupdate__block_date'))
        items = []
        for li in block.select('ul.b-seriesupdate__block_list > li.b-seriesupdate__block_list_item'):
            item = _parse_item(li)
            if item is not None:
                items.append(item)
        blocks.append(SeriesUpdateBlock(date=date, items=tuple(items)))
    return tuple(blocks)

Parse .b-seriesupdate__block entries from the inline sidebar.

Returns an empty tuple when the sidebar feed is missing.

def urls_from_ajax_response(response: APIResponse,
*,
client: HDRezkaClient | None = None) ‑> URLs
Expand source code
def urls_from_ajax_response(response: APIResponse, *, client: 'HDRezkaClient | None' = None) -> URLs:
    """Parse an AJAX stream response into ``URLs``."""
    return URLs(VideoURLs(response.get('url', ''), client=client),
                SubtitleURLs(response.get('subtitle', ''),
                             response.get('subtitle_lns', {'off': ''}),
                             response.get('subtitle_def', '')))

Parse an AJAX stream response into URLs.

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 Favorites (cat_id: typing.SupportsInt | str | None = None,
*,
client: HDRezkaClient)
Expand source code
class Favorites(Page):
    """
    Paginated favorites collection page.

    Titles are loaded via the usual ``get_page`` / ``get_page_content`` methods.
    Collection links on the same HTML are available through ``get_cats``.
    """

    __slots__ = ('_cat_id',)

    def __init__(self, cat_id: SupportsInt | str | None = None, *, client: 'HDRezkaClient'):
        """
        :param cat_id: Collection id for ``/favorites/{id}/``, or ``None`` for ``/favorites/``.
        :param client: Authenticated session (favorites require login).
        """
        self._cat_id = None if cat_id is None else int(cat_id)
        if self._cat_id is None:
            url = client.host_join('favorites/')
        else:
            url = client.host_join(f'favorites/{self._cat_id}/')
        super().__init__(url, client=client)

    @property
    def cat_id(self) -> int | None:
        """Collection id for this page, if any."""
        return self._cat_id

    async def get_cats(self) -> tuple[FavoritesCat, ...]:
        """
        Fetch this favorites page and parse the collection list.

        Returns an empty tuple when the list markup is missing.
        """
        html = (await self.client.get_response('GET', self._page)).text
        return parse_favorites_cats(html)

Paginated favorites collection page.

Titles are loaded via the usual get_page / get_page_content methods. Collection links on the same HTML are available through get_cats.

:param cat_id: Collection id for /favorites/{id}/, or None for /favorites/. :param client: Authenticated session (favorites require login).

Ancestors

Instance variables

prop cat_id : int | None
Expand source code
@property
def cat_id(self) -> int | None:
    """Collection id for this page, if any."""
    return self._cat_id

Collection id for this page, if any.

Methods

async def get_cats(self) ‑> tuple[FavoritesCat, ...]
Expand source code
async def get_cats(self) -> tuple[FavoritesCat, ...]:
    """
    Fetch this favorites page and parse the collection list.

    Returns an empty tuple when the list markup is missing.
    """
    html = (await self.client.get_response('GET', self._page)).text
    return parse_favorites_cats(html)

Fetch this favorites page and parse the collection list.

Returns an empty tuple when the list markup is missing.

Inherited members

class FavoritesCat (name: str, url: str, count: int = 0, id: int | None = None)
Expand source code
class FavoritesCat(NamedTuple):
    """User favorites collection link from the favorites page sidebar/list."""
    name: str
    url: str
    count: int = 0
    id: int | None = None

User favorites collection link from the favorites page sidebar/list.

Ancestors

  • builtins.tuple

Instance variables

var count : int
Expand source code
class FavoritesCat(NamedTuple):
    """User favorites collection link from the favorites page sidebar/list."""
    name: str
    url: str
    count: int = 0
    id: int | None = None

Alias for field number 2

var id : int | None
Expand source code
class FavoritesCat(NamedTuple):
    """User favorites collection link from the favorites page sidebar/list."""
    name: str
    url: str
    count: int = 0
    id: int | None = None

Alias for field number 3

var name : str
Expand source code
class FavoritesCat(NamedTuple):
    """User favorites collection link from the favorites page sidebar/list."""
    name: str
    url: str
    count: int = 0
    id: int | None = None

Alias for field number 0

var url : str
Expand source code
class FavoritesCat(NamedTuple):
    """User favorites collection link from the favorites page sidebar/list."""
    name: str
    url: str
    count: int = 0
    id: int | None = None

Alias for field number 1

Expand source code
class FilterLink(NamedTuple):
    """One filter or content-type control on a catalog page."""
    name: str
    url: str
    param: str
    value: str
    active: bool = False

One filter or content-type control on a catalog page.

Ancestors

  • builtins.tuple

Instance variables

var active : bool
Expand source code
class FilterLink(NamedTuple):
    """One filter or content-type control on a catalog page."""
    name: str
    url: str
    param: str
    value: str
    active: bool = False

Alias for field number 4

var name : str
Expand source code
class FilterLink(NamedTuple):
    """One filter or content-type control on a catalog page."""
    name: str
    url: str
    param: str
    value: str
    active: bool = False

Alias for field number 0

var param : str
Expand source code
class FilterLink(NamedTuple):
    """One filter or content-type control on a catalog page."""
    name: str
    url: str
    param: str
    value: str
    active: bool = False

Alias for field number 2

var url : str
Expand source code
class FilterLink(NamedTuple):
    """One filter or content-type control on a catalog page."""
    name: str
    url: str
    param: str
    value: str
    active: bool = False

Alias for field number 1

var value : str
Expand source code
class FilterLink(NamedTuple):
    """One filter or content-type control on a catalog page."""
    name: str
    url: str
    param: str
    value: str
    active: bool = False

Alias for field number 3

class FindBestBlock (categories: tuple[FindBestOption, ...] = (),
years: tuple[FindBestOption, ...] = ())
Expand source code
class FindBestBlock(NamedTuple):
    """Category and year selectors from ``.b-topnav__findbest_block``."""
    categories: tuple[FindBestOption, ...] = ()
    years: tuple[FindBestOption, ...] = ()

Category and year selectors from .b-topnav__findbest_block.

Ancestors

  • builtins.tuple

Instance variables

var categories : tuple[FindBestOption, ...]
Expand source code
class FindBestBlock(NamedTuple):
    """Category and year selectors from ``.b-topnav__findbest_block``."""
    categories: tuple[FindBestOption, ...] = ()
    years: tuple[FindBestOption, ...] = ()

Alias for field number 0

var years : tuple[FindBestOption, ...]
Expand source code
class FindBestBlock(NamedTuple):
    """Category and year selectors from ``.b-topnav__findbest_block``."""
    categories: tuple[FindBestOption, ...] = ()
    years: tuple[FindBestOption, ...] = ()

Alias for field number 1

class FindBestOption (name: str, value: str, selected: bool = False)
Expand source code
class FindBestOption(NamedTuple):
    """Option from a navbar “best of” ``<select>``."""
    name: str
    value: str
    selected: bool = False

Option from a navbar “best of” <select>.

Ancestors

  • builtins.tuple

Instance variables

var name : str
Expand source code
class FindBestOption(NamedTuple):
    """Option from a navbar “best of” ``<select>``."""
    name: str
    value: str
    selected: bool = False

Alias for field number 0

var selected : bool
Expand source code
class FindBestOption(NamedTuple):
    """Option from a navbar “best of” ``<select>``."""
    name: str
    value: str
    selected: bool = False

Alias for field number 2

var value : str
Expand source code
class FindBestOption(NamedTuple):
    """Option from a navbar “best of” ``<select>``."""
    name: str
    value: str
    selected: bool = False

Alias for field number 1

class HDRezkaClient (*,
host: str | None = None,
proxy: str | None = None,
cookies: httpx.Cookies | http.cookiejar.CookieJar | dict[str, str] | list[tuple[str, str]] = None,
http_client: httpx.AsyncClient | None = None,
request_kwargs: RequestKwargs | dict | None = None,
headers: dict | None = None,
redirect_url: str | None = None,
impersonate: bool | str = True)
Expand source code
class HDRezkaClient:
    """
    Isolated HDRezka HTTP session for one user or one API request scope.

    Each instance owns its own ``httpx.AsyncClient``, host, cookies, and player cache.
    Create one client per authenticated user when building a multi-user API.

    By default the session uses browser TLS/HTTP2 impersonation (``curl_cffi`` via
    ``httpx-curl-cffi``) so mirrors behind Cloudflare are more likely to respond.
    Pass ``impersonate=False`` to use plain httpx, or a string target such as
    ``"chrome"`` / ``"safari"`` to pick a specific fingerprint.

    Example::

        async with HDRezkaClient() as client:
            await client.login(email, password)
            items = await client.search('Breaking Bad').get_page(1)
            player = await client.player(items[0].url)
    """

    __slots__ = (
        '_host',
        '_redirect_url',
        '_http',
        '_owns_http',
        '_impersonate',
        '_request_kwargs',
        '_ajax',
        '_player_cache',
        '_auth_expires_at'
    )

    def __init__(
            self,
            *,
            host: str | None = None,
            proxy: str | None = None,
            cookies: httpx.Cookies | CookieJar | dict[str, str] | list[tuple[str, str]] = None,
            http_client: httpx.AsyncClient | None = None,
            request_kwargs: RequestKwargs | dict | None = None,
            headers: dict | None = None,
            redirect_url: str | None = None,
            impersonate: ImpersonateTarget = True,
    ):
        """
        Create a new session.

        :param host: Base mirror URL. Defaults to ``DEFAULT_HOST``.
        :param proxy: Optional proxy URL (e.g. ``socks5://localhost:9050``).
            Ignored when ``http_client`` is provided.
        :param cookies: Optional cookies dict/string to initialize state-less requests.
        :param http_client: Existing ``httpx.AsyncClient`` to reuse. The client
            will not be closed by ``aclose`` / context exit when passed this way.
            When set, ``impersonate`` / ``proxy`` / ``headers`` are ignored.
        :param request_kwargs: Default kwargs merged into every ``get_response`` call
            (per-call kwargs win).
        :param headers: Extra headers for a newly created HTTP client.
        :param redirect_url: Standby URL used by ``login`` to discover an active mirror.
        :param impersonate: Browser TLS impersonation. ``True`` (default) uses Chrome;
            ``False`` disables it; a string selects a ``curl_cffi`` target.
        """
        self._host = (host or DEFAULT_HOST).rstrip('/') + '/'
        self._redirect_url = redirect_url or DEFAULT_REDIRECT_URL
        self._request_kwargs: dict = dict(request_kwargs or {})
        self._player_cache = CacheStorage()
        self._owns_http = http_client is None
        self._impersonate: ImpersonateTarget = impersonate if http_client is None else False
        self._auth_expires_at = None
        if http_client is not None:
            self._http = http_client
        else:
            self._http = build_async_client(
                proxy=proxy,
                headers=headers,
                impersonate=impersonate,
            )
        if cookies:
            self._http.cookies.update(cookies)
        self._ajax = AJAX(self)

    @property
    def host(self) -> str:
        """Active mirror base URL for this session."""
        return self._host

    @host.setter
    def host(self, value: str) -> None:
        """Set mirror base URL (trailing slash normalized)."""
        self._host = value.rstrip('/') + '/'

    @property
    def redirect_url(self) -> str:
        """Standby URL used to discover an active mirror during login."""
        return self._redirect_url

    @property
    def http(self) -> httpx.AsyncClient:
        """Underlying ``httpx.AsyncClient`` for this session."""
        return self._http

    @property
    def impersonate(self) -> ImpersonateTarget:
        """Impersonation setting used when this client created its HTTP session."""
        return self._impersonate

    @property
    def cookies(self) -> httpx.Cookies:
        """Cookie jar of the underlying HTTP client."""
        return self._http.cookies

    @property
    def ajax(self) -> AJAX:
        """AJAX API bound to this client."""
        return self._ajax

    @property
    def request_kwargs(self) -> dict:
        """Default request kwargs applied by ``get_response``."""
        return self._request_kwargs

    @property
    def player_cache(self) -> CacheStorage:
        """Per-client cache of resolved ``Player`` instances."""
        return self._player_cache

    def host_join(self, url: str | None, allow_fragments: bool = True) -> str:
        """Join ``url`` with this client's ``host``."""
        return urljoin(self._host, url, allow_fragments=allow_fragments)

    async def get_response(self, method: str, url: str | httpx.URL, **kwargs) -> httpx.Response:
        """
        Send an HTTP request through this client's session.

        Explicit ``kwargs`` override ``request_kwargs`` for the same keys.
        """
        for key, value in self._request_kwargs.items():
            if key not in kwargs:
                kwargs[key] = value
        return await self._http.request(method, url, **kwargs)

    async def login(self, email: str, password: str) -> Self:
        """Authenticate against HDRezka and switch ``host`` to the active mirror.

        Discovers a working mirror via ``redirect_url``, posts credentials to
        ``/ajax/login/``, and stores session cookies on this client.
        """
        follow_redirects = self._http.follow_redirects
        self._http.follow_redirects = False
        try:
            resp = await self.get_response('GET', self._redirect_url)
        finally:
            self._http.follow_redirects = follow_redirects
        url = httpx.URL(resp.headers.get('Location', str(resp.url)))
        ajax_login_url = url.join('/ajax/login/')
        await self.get_response(
            'POST',
            ajax_login_url,
            data={
                'login_name': email,
                'login_password': password,
                'login_not_save': '0',
                'login': 'submit',
            },
        )
        self.host = f'https://{url.host}'
        self._extract_and_store_cookies()
        return self

    def _extract_and_store_cookies(self) -> None:
        """Retrieves cookies and their lifetime from the internal CookieJar."""
        jar = self._http.cookies.jar
        if not isinstance(jar, CookieJar):
            return
        auth_expires = []
        _cookies_details = {}
        for cookie in jar:
            expires_dt = None
            if cookie.expires:
                expires_dt = datetime.fromtimestamp(
                    cookie.expires, tz=timezone.utc
                )
            _cookies_details[cookie.name] = expires_dt
            if cookie.name in ('dle_user_id', 'dle_password') and expires_dt:
                auth_expires.append(expires_dt)
        if auth_expires:
            self._auth_expires_at = min(auth_expires)

    @property
    def is_authenticated(self) -> bool:
        """Checking if authorization is still active."""
        return not self._auth_expires_at or datetime.now(timezone.utc) < self._auth_expires_at

    def search(self, query: str = ''):
        """Return a ``Search`` bound to this client."""
        from .api.search import Search
        return Search(query, client=self)

    def page(self, url: str | None = None):
        """Return a catalog ``Page`` bound to this client (defaults to ``host``)."""
        from .post.page import Page
        return Page(url or self.host, client=self)

    def favorites(self, cat_id: int | str | None = None):
        """
        Return a ``Favorites`` page for ``/favorites/`` or ``/favorites/{cat_id}/``.

        Requires an authenticated session (``login``).
        """
        from .post.favorites import Favorites
        return Favorites(cat_id, client=self)

    async def add_favorites_cat(self, name: str):
        """Create a favorites collection. See ``AJAX.add_favorites_cat``."""
        return await self.ajax.add_favorites_cat(name)

    async def rename_favorites_cat(self, cat_id: int | str, name: str):
        """Rename a favorites collection. See ``AJAX.rename_favorites_cat``."""
        return await self.ajax.rename_favorites_cat(cat_id, name)

    async def add_favorites_post(self, post_id: int | str, cat_id: int | str):
        """Add a title to a favorites collection. See ``AJAX.add_favorites_post``."""
        return await self.ajax.add_favorites_post(post_id, cat_id)

    async def remove_favorites_cat(self, cat_id: int | str):
        """Remove a favorites collection. See ``AJAX.remove_favorites_cat``."""
        return await self.ajax.remove_favorites_cat(cat_id)

    async def navbar(self):
        """
        Fetch the host home page and parse the top navigation map.

        Returns an empty ``Navbar`` when the menu markup is missing.
        """
        from .post.site import parse_navbar
        html = (await self.get_response('GET', self.host)).text
        return parse_navbar(html)

    async def series_updates(self):
        """
        Fetch the host home page and parse sidebar series updates.

        Returns an empty tuple when the block is missing.
        """
        from .post.site import parse_series_updates
        html = (await self.get_response('GET', self.host)).text
        return parse_series_updates(html)

    async def player(self, url_or_path: Any):
        """Resolve and return a ``PlayerMovie`` or ``PlayerSeries`` for the URL."""
        from .stream.player import player
        return await player(url_or_path, client=self)

    async def post(self, url: str):
        """Fetch and return an initialized ``Post`` for the URL."""
        from .post.post import Post
        return await Post(url, client=self)

    async def aclose(self) -> None:
        """Close the underlying HTTP client if this instance created it."""
        if self._owns_http:
            await self._http.aclose()

    async def __aenter__(self) -> Self:
        return self

    async def __aexit__(self, exc_type, exc, tb) -> None:
        await self.aclose()

    def __repr__(self) -> str:
        return f'{self.__class__.__qualname__}(host={self._host!r})'

Isolated HDRezka HTTP session for one user or one API request scope.

Each instance owns its own httpx.AsyncClient, host, cookies, and player cache. Create one client per authenticated user when building a multi-user API.

By default the session uses browser TLS/HTTP2 impersonation (curl_cffi via httpx-curl-cffi) so mirrors behind Cloudflare are more likely to respond. Pass impersonate=False to use plain httpx, or a string target such as "chrome" / "safari" to pick a specific fingerprint.

Example::

async with HDRezkaClient() as client:
    await client.login(email, password)
    items = await client.search('Breaking Bad').get_page(1)
    player = await client.player(items[0].url)

Create a new session.

:param host: Base mirror URL. Defaults to DEFAULT_HOST. :param proxy: Optional proxy URL (e.g. socks5://localhost:9050). Ignored when http_client is provided. :param cookies: Optional cookies dict/string to initialize state-less requests. :param http_client: Existing httpx.AsyncClient to reuse. The client will not be closed by aclose / context exit when passed this way. When set, impersonate / proxy / headers are ignored. :param request_kwargs: Default kwargs merged into every get_response call (per-call kwargs win). :param headers: Extra headers for a newly created HTTP client. :param redirect_url: Standby URL used by login to discover an active mirror. :param impersonate: Browser TLS impersonation. True (default) uses Chrome; False disables it; a string selects a curl_cffi target.

Instance variables

prop ajaxAJAX
Expand source code
@property
def ajax(self) -> AJAX:
    """AJAX API bound to this client."""
    return self._ajax

AJAX API bound to this client.

prop cookies : httpx.Cookies
Expand source code
@property
def cookies(self) -> httpx.Cookies:
    """Cookie jar of the underlying HTTP client."""
    return self._http.cookies

Cookie jar of the underlying HTTP client.

prop host : str
Expand source code
@property
def host(self) -> str:
    """Active mirror base URL for this session."""
    return self._host

Active mirror base URL for this session.

prop http : httpx.AsyncClient
Expand source code
@property
def http(self) -> httpx.AsyncClient:
    """Underlying ``httpx.AsyncClient`` for this session."""
    return self._http

Underlying httpx.AsyncClient for this session.

prop impersonate : bool | str
Expand source code
@property
def impersonate(self) -> ImpersonateTarget:
    """Impersonation setting used when this client created its HTTP session."""
    return self._impersonate

Impersonation setting used when this client created its HTTP session.

prop is_authenticated : bool
Expand source code
@property
def is_authenticated(self) -> bool:
    """Checking if authorization is still active."""
    return not self._auth_expires_at or datetime.now(timezone.utc) < self._auth_expires_at

Checking if authorization is still active.

prop player_cache : hdrezka.stream._cache.CacheStorage
Expand source code
@property
def player_cache(self) -> CacheStorage:
    """Per-client cache of resolved ``Player`` instances."""
    return self._player_cache

Per-client cache of resolved player() instances.

prop redirect_url : str
Expand source code
@property
def redirect_url(self) -> str:
    """Standby URL used to discover an active mirror during login."""
    return self._redirect_url

Standby URL used to discover an active mirror during login.

prop request_kwargs : dict
Expand source code
@property
def request_kwargs(self) -> dict:
    """Default request kwargs applied by ``get_response``."""
    return self._request_kwargs

Default request kwargs applied by get_response.

Methods

async def aclose(self) ‑> None
Expand source code
async def aclose(self) -> None:
    """Close the underlying HTTP client if this instance created it."""
    if self._owns_http:
        await self._http.aclose()

Close the underlying HTTP client if this instance created it.

async def add_favorites_cat(self, name: str)
Expand source code
async def add_favorites_cat(self, name: str):
    """Create a favorites collection. See ``AJAX.add_favorites_cat``."""
    return await self.ajax.add_favorites_cat(name)

Create a favorites collection. See AJAX.add_favorites_cat().

async def add_favorites_post(self, post_id: int | str, cat_id: int | str)
Expand source code
async def add_favorites_post(self, post_id: int | str, cat_id: int | str):
    """Add a title to a favorites collection. See ``AJAX.add_favorites_post``."""
    return await self.ajax.add_favorites_post(post_id, cat_id)

Add a title to a favorites collection. See AJAX.add_favorites_post().

def favorites(self, cat_id: int | str | None = None)
Expand source code
def favorites(self, cat_id: int | str | None = None):
    """
    Return a ``Favorites`` page for ``/favorites/`` or ``/favorites/{cat_id}/``.

    Requires an authenticated session (``login``).
    """
    from .post.favorites import Favorites
    return Favorites(cat_id, client=self)

Return a Favorites page for /favorites/ or /favorites/{cat_id}/.

Requires an authenticated session (login).

async def get_response(self, method: str, url: str | httpx.URL, **kwargs) ‑> httpx.Response
Expand source code
async def get_response(self, method: str, url: str | httpx.URL, **kwargs) -> httpx.Response:
    """
    Send an HTTP request through this client's session.

    Explicit ``kwargs`` override ``request_kwargs`` for the same keys.
    """
    for key, value in self._request_kwargs.items():
        if key not in kwargs:
            kwargs[key] = value
    return await self._http.request(method, url, **kwargs)

Send an HTTP request through this client's session.

Explicit kwargs override request_kwargs for the same keys.

def host_join(self, url: str | None, allow_fragments: bool = True) ‑> str
Expand source code
def host_join(self, url: str | None, allow_fragments: bool = True) -> str:
    """Join ``url`` with this client's ``host``."""
    return urljoin(self._host, url, allow_fragments=allow_fragments)

Join hdrezka.url with this client's host.

async def login(self, email: str, password: str) ‑> Self
Expand source code
async def login(self, email: str, password: str) -> Self:
    """Authenticate against HDRezka and switch ``host`` to the active mirror.

    Discovers a working mirror via ``redirect_url``, posts credentials to
    ``/ajax/login/``, and stores session cookies on this client.
    """
    follow_redirects = self._http.follow_redirects
    self._http.follow_redirects = False
    try:
        resp = await self.get_response('GET', self._redirect_url)
    finally:
        self._http.follow_redirects = follow_redirects
    url = httpx.URL(resp.headers.get('Location', str(resp.url)))
    ajax_login_url = url.join('/ajax/login/')
    await self.get_response(
        'POST',
        ajax_login_url,
        data={
            'login_name': email,
            'login_password': password,
            'login_not_save': '0',
            'login': 'submit',
        },
    )
    self.host = f'https://{url.host}'
    self._extract_and_store_cookies()
    return self

Authenticate against HDRezka and switch host to the active mirror.

Discovers a working mirror via redirect_url, posts credentials to /ajax/login/, and stores session cookies on this client.

async def navbar(self)
Expand source code
async def navbar(self):
    """
    Fetch the host home page and parse the top navigation map.

    Returns an empty ``Navbar`` when the menu markup is missing.
    """
    from .post.site import parse_navbar
    html = (await self.get_response('GET', self.host)).text
    return parse_navbar(html)

Fetch the host home page and parse the top navigation map.

Returns an empty Navbar when the menu markup is missing.

def page(self, url: str | None = None)
Expand source code
def page(self, url: str | None = None):
    """Return a catalog ``Page`` bound to this client (defaults to ``host``)."""
    from .post.page import Page
    return Page(url or self.host, client=self)

Return a catalog Page bound to this client (defaults to host).

async def player(self, url_or_path: Any)
Expand source code
async def player(self, url_or_path: Any):
    """Resolve and return a ``PlayerMovie`` or ``PlayerSeries`` for the URL."""
    from .stream.player import player
    return await player(url_or_path, client=self)

Resolve and return a PlayerMovie or PlayerSeries for the URL.

async def post(self, url: str)
Expand source code
async def post(self, url: str):
    """Fetch and return an initialized ``Post`` for the URL."""
    from .post.post import Post
    return await Post(url, client=self)

Fetch and return an initialized Post for the URL.

async def remove_favorites_cat(self, cat_id: int | str)
Expand source code
async def remove_favorites_cat(self, cat_id: int | str):
    """Remove a favorites collection. See ``AJAX.remove_favorites_cat``."""
    return await self.ajax.remove_favorites_cat(cat_id)

Remove a favorites collection. See AJAX.remove_favorites_cat().

async def rename_favorites_cat(self, cat_id: int | str, name: str)
Expand source code
async def rename_favorites_cat(self, cat_id: int | str, name: str):
    """Rename a favorites collection. See ``AJAX.rename_favorites_cat``."""
    return await self.ajax.rename_favorites_cat(cat_id, name)

Rename a favorites collection. See AJAX.rename_favorites_cat().

def search(self, query: str = '')
Expand source code
def search(self, query: str = ''):
    """Return a ``Search`` bound to this client."""
    from .api.search import Search
    return Search(query, client=self)

Return a Search bound to this client.

async def series_updates(self)
Expand source code
async def series_updates(self):
    """
    Fetch the host home page and parse sidebar series updates.

    Returns an empty tuple when the block is missing.
    """
    from .post.site import parse_series_updates
    html = (await self.get_response('GET', self.host)).text
    return parse_series_updates(html)

Fetch the host home page and parse sidebar series updates.

Returns an empty tuple when the block is missing.

class InlineInfo (year: int, year_final: int | ellipsis | None, country: str, genre: str)
Expand source code
class InlineInfo(NamedTuple):
    """Info about an inline catalog item (bottom line)."""
    year: int
    year_final: int | EllipsisType | None
    'If the film is equal to None, if ongoing, equal ``...``'
    country: str
    genre: str

Info about an inline catalog item (bottom line).

Ancestors

  • builtins.tuple

Instance variables

var country : str
Expand source code
class InlineInfo(NamedTuple):
    """Info about an inline catalog item (bottom line)."""
    year: int
    year_final: int | EllipsisType | None
    'If the film is equal to None, if ongoing, equal ``...``'
    country: str
    genre: str

Alias for field number 2

var genre : str
Expand source code
class InlineInfo(NamedTuple):
    """Info about an inline catalog item (bottom line)."""
    year: int
    year_final: int | EllipsisType | None
    'If the film is equal to None, if ongoing, equal ``...``'
    country: str
    genre: str

Alias for field number 3

var year : int
Expand source code
class InlineInfo(NamedTuple):
    """Info about an inline catalog item (bottom line)."""
    year: int
    year_final: int | EllipsisType | None
    'If the film is equal to None, if ongoing, equal ``...``'
    country: str
    genre: str

Alias for field number 0

var year_final : int | ellipsis | None
Expand source code
class InlineInfo(NamedTuple):
    """Info about an inline catalog item (bottom line)."""
    year: int
    year_final: int | EllipsisType | None
    'If the film is equal to None, if ongoing, equal ``...``'
    country: str
    genre: str

If the film is equal to None, if ongoing, equal

class InlineItem (url: str,
name: str,
info: InlineInfo,
poster: str,
client: ForwardRef('HDRezkaClient'))
Expand source code
class InlineItem(NamedTuple):
    """Content inline item from a catalog or search page."""
    url: str
    name: str
    info: InlineInfo
    poster: str
    'Image URL'
    client: 'HDRezkaClient'
    'Session used to resolve ``player``'

    @property
    async def player(self):
        """Return a ``Player`` instance for this item."""
        return await self.client.player(self.url)

Content inline item from a catalog or search page.

Ancestors

  • builtins.tuple

Instance variables

var client
Expand source code
class InlineItem(NamedTuple):
    """Content inline item from a catalog or search page."""
    url: str
    name: str
    info: InlineInfo
    poster: str
    'Image URL'
    client: 'HDRezkaClient'
    'Session used to resolve ``player``'

    @property
    async def player(self):
        """Return a ``Player`` instance for this item."""
        return await self.client.player(self.url)

Session used to resolve player

var info
Expand source code
class InlineItem(NamedTuple):
    """Content inline item from a catalog or search page."""
    url: str
    name: str
    info: InlineInfo
    poster: str
    'Image URL'
    client: 'HDRezkaClient'
    'Session used to resolve ``player``'

    @property
    async def player(self):
        """Return a ``Player`` instance for this item."""
        return await self.client.player(self.url)

Alias for field number 2

var name
Expand source code
class InlineItem(NamedTuple):
    """Content inline item from a catalog or search page."""
    url: str
    name: str
    info: InlineInfo
    poster: str
    'Image URL'
    client: 'HDRezkaClient'
    'Session used to resolve ``player``'

    @property
    async def player(self):
        """Return a ``Player`` instance for this item."""
        return await self.client.player(self.url)

Alias for field number 1

prop player
Expand source code
@property
async def player(self):
    """Return a ``Player`` instance for this item."""
    return await self.client.player(self.url)

Return a player() instance for this item.

var poster
Expand source code
class InlineItem(NamedTuple):
    """Content inline item from a catalog or search page."""
    url: str
    name: str
    info: InlineInfo
    poster: str
    'Image URL'
    client: 'HDRezkaClient'
    'Session used to resolve ``player``'

    @property
    async def player(self):
        """Return a ``Player`` instance for this item."""
        return await self.client.player(self.url)

Image URL

var url
Expand source code
class InlineItem(NamedTuple):
    """Content inline item from a catalog or search page."""
    url: str
    name: str
    info: InlineInfo
    poster: str
    'Image URL'
    client: 'HDRezkaClient'
    'Session used to resolve ``player``'

    @property
    async def player(self):
        """Return a ``Player`` instance for this item."""
        return await self.client.player(self.url)

Alias for field number 0

class NavCollection (name: str, url: str, classes: tuple[str, ...] = ())
Expand source code
class NavCollection(NamedTuple):
    """Collection / platform link from a category submenu (right column)."""
    name: str
    url: str
    classes: tuple[str, ...] = ()

Collection / platform link from a category submenu (right column).

Ancestors

  • builtins.tuple

Instance variables

var classes : tuple[str, ...]
Expand source code
class NavCollection(NamedTuple):
    """Collection / platform link from a category submenu (right column)."""
    name: str
    url: str
    classes: tuple[str, ...] = ()

Alias for field number 2

var name : str
Expand source code
class NavCollection(NamedTuple):
    """Collection / platform link from a category submenu (right column)."""
    name: str
    url: str
    classes: tuple[str, ...] = ()

Alias for field number 0

var url : str
Expand source code
class NavCollection(NamedTuple):
    """Collection / platform link from a category submenu (right column)."""
    name: str
    url: str
    classes: tuple[str, ...] = ()

Alias for field number 1

class NavItem (name: str,
url: str,
single: bool = False,
submenu: NavSubmenu | None = None)
Expand source code
class NavItem(NamedTuple):
    """Single top-nav entry (with or without submenu)."""
    name: str
    url: str
    single: bool = False
    submenu: NavSubmenu | None = None

Single top-nav entry (with or without submenu).

Ancestors

  • builtins.tuple

Instance variables

var name : str
Expand source code
class NavItem(NamedTuple):
    """Single top-nav entry (with or without submenu)."""
    name: str
    url: str
    single: bool = False
    submenu: NavSubmenu | None = None

Alias for field number 0

var single : bool
Expand source code
class NavItem(NamedTuple):
    """Single top-nav entry (with or without submenu)."""
    name: str
    url: str
    single: bool = False
    submenu: NavSubmenu | None = None

Alias for field number 2

var submenuNavSubmenu | None
Expand source code
class NavItem(NamedTuple):
    """Single top-nav entry (with or without submenu)."""
    name: str
    url: str
    single: bool = False
    submenu: NavSubmenu | None = None

Alias for field number 3

var url : str
Expand source code
class NavItem(NamedTuple):
    """Single top-nav entry (with or without submenu)."""
    name: str
    url: str
    single: bool = False
    submenu: NavSubmenu | None = None

Alias for field number 1

class NavSubmenu (genres: tuple[Hyperlink, ...] = (),
collections: tuple[NavCollection, ...] = (),
find_best: FindBestBlock | None = None)
Expand source code
class NavSubmenu(NamedTuple):
    """Dropdown content under a top-nav category."""
    genres: tuple[Hyperlink, ...] = ()
    collections: tuple[NavCollection, ...] = ()
    find_best: FindBestBlock | None = None

Dropdown content under a top-nav category.

Ancestors

  • builtins.tuple

Instance variables

var collections : tuple[NavCollection, ...]
Expand source code
class NavSubmenu(NamedTuple):
    """Dropdown content under a top-nav category."""
    genres: tuple[Hyperlink, ...] = ()
    collections: tuple[NavCollection, ...] = ()
    find_best: FindBestBlock | None = None

Alias for field number 1

var find_bestFindBestBlock | None
Expand source code
class NavSubmenu(NamedTuple):
    """Dropdown content under a top-nav category."""
    genres: tuple[Hyperlink, ...] = ()
    collections: tuple[NavCollection, ...] = ()
    find_best: FindBestBlock | None = None

Alias for field number 2

var genres : tuple[Hyperlink, ...]
Expand source code
class NavSubmenu(NamedTuple):
    """Dropdown content under a top-nav category."""
    genres: tuple[Hyperlink, ...] = ()
    collections: tuple[NavCollection, ...] = ()
    find_best: FindBestBlock | None = None

Alias for field number 0

class Navbar (items: tuple[NavItem, ...] = ())
Expand source code
class Navbar(NamedTuple):
    """Full site top navigation map."""
    items: tuple[NavItem, ...] = ()

Full site top navigation map.

Ancestors

  • builtins.tuple

Instance variables

var items : tuple[NavItem, ...]
Expand source code
class Navbar(NamedTuple):
    """Full site top navigation map."""
    items: tuple[NavItem, ...] = ()

Alias for field number 0

class Page (url: str,
*,
client: HDRezkaClient)
Expand source code
class Page:
    """Paginated HDRezka catalog or listing page."""
    __slots__ = ('_page', '_page_format', '__yields', '__yields_page', 'client')

    def __init__(self, url: str, *, client: 'HDRezkaClient'):
        """
        :param url: Absolute page URL (or host root).
        :param client: Session used for HTTP requests.
        """
        self.client = client
        self.__yields: list[InlineItem] = []
        self.__yields_page = 0
        self.page = url

    @property
    def page(self) -> str:
        """Current page URL template base."""
        return self._page

    @page.setter
    def page(self, value):
        """Cast value to str and set the paginator format."""
        if not isinstance(value, str):
            value = str(value)
        # noinspection HttpUrlsUsage
        if not (value.startswith('https://') or value.startswith('http://')):
            value = f'https://{value}'
        self._page = value
        self._page_format = self._concat_paginator(value.removesuffix('/'))

    @staticmethod
    def _concat_paginator(url: str) -> Callable[[...], str]:
        def __concat_paginator(page, *args, **kwargs):
            if page is not None:
                return f'{url}/page/{{0}}/'.format(page, *args, **kwargs)
            return url.format(*args, **kwargs)

        return __concat_paginator

    @staticmethod
    def _inline_info(years: str, country: str, genre: str):
        year, *finals = years.split('-')
        if finals:
            final, = finals
            year_final = ... if final.strip() == '...' else int(final)
        else:
            year_final = None
        return InlineInfo(int(year), year_final, country.strip(), genre.strip())

    def _request_url(self, page: int | slice | Iterable[int] | Any, **query) -> str:
        """Build a paginated request URL with optional query parameters."""
        return merge_query(self._page_format(page), **query)

    def _parse_items(self, soup: BeautifulSoup) -> list[InlineItem]:
        items = soup.find_all(class_='b-content__inline_item')
        return [InlineItem(
            (a := (link := i.find(class_='b-content__inline_item-link')).find('a'))['href'],
            a.text,
            self._inline_info(*link.find('div').text.split(',', 3)),
            i.find(class_='b-content__inline_item-cover').find('img').get('src', ''),
            self.client,
        )
            for i in items]

    async def get_page_content(
            self,
            page: int | slice | Iterable[int] | None = None,
            *,
            filter: str | None = None,
            genre: str | int | None = None,
            **query,
    ) -> PageContent:
        """
        Fetch a catalog page and parse items, filters, series updates, and favorites cats.

        Extra keyword arguments are merged into the request query string
        (for example ``filter`` / ``genre`` from ``PageFilters``).
        Raises ``EmptyPage`` when no inline items are present.
        """
        if filter is not None:
            query['filter'] = filter
        if genre is not None:
            query['genre'] = genre
        html = (await self.client.get_response('GET', self._request_url(page, **query))).text
        soup = BeautifulSoup(html, builder=BUILDER)
        items = self._parse_items(soup)
        if not items:
            raise EmptyPage(page)
        return PageContent(
            items=items,
            filters=parse_page_filters(soup),
            series_updates=parse_series_updates(soup),
            favorites_cats=parse_favorites_cats(soup),
        )

    async def get_page(
            self,
            page: int | slice | Iterable[int] | None = None,
            *,
            filter: str | None = None,
            genre: str | int | None = None,
            **query,
    ) -> list[InlineItem]:
        """
        Fetch inline items for the given page number.

        Optional ``filter`` / ``genre`` (and other query kwargs) are appended
        to the request URL. Raises ``EmptyPage`` when the page has no items.
        """
        return (await self.get_page_content(page, filter=filter, genre=genre, **query)).items

    async def get_filters(
            self,
            page: int | slice | Iterable[int] = 1,
            *,
            filter: str | None = None,
            genre: str | int | None = None,
            **query,
    ) -> PageFilters:
        """
        Fetch the page and return available sort/type filters (empty if absent).

        Does not require inline items to be present.
        """
        if filter is not None:
            query['filter'] = filter
        if genre is not None:
            query['genre'] = genre
        html = (await self.client.get_response('GET', self._request_url(page, **query))).text
        return parse_page_filters(html)

    async def get_series_updates(self) -> tuple[SeriesUpdateBlock, ...]:
        """
        Fetch this page URL and parse sidebar series updates.

        Typically present on the site home page. Returns an empty tuple if missing.
        """
        html = (await self.client.get_response('GET', self._page)).text
        return parse_series_updates(html)

    async def get_navbar(self) -> Navbar:
        """Fetch this page URL and parse the top navigation menu."""
        html = (await self.client.get_response('GET', self._page)).text
        return parse_navbar(html)

    async def get_favorites_cats(self) -> tuple[FavoritesCat, ...]:
        """
        Fetch this page URL and parse favorites collection links.

        Present on ``/favorites/`` pages. Returns an empty tuple if missing.
        """
        html = (await self.client.get_response('GET', self._page)).text
        return parse_favorites_cats(html)

    def __aiter__(self):
        """Async iterator over items across successive pages."""
        self.__yields_page = 0
        self.__yields.clear()
        return self

    async def __anext__(self) -> InlineItem:
        """Return the next ``InlineItem`` across pages."""
        if not self.__yields:
            try:
                self.__yields_page += 1
                self.__yields += await self.get_page(self.__yields_page)
            except EmptyPage:
                raise StopAsyncIteration
        return self.__yields.pop(0)

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

Paginated HDRezka catalog or listing page.

:param url: Absolute page URL (or host root). :param client: Session used for HTTP requests.

Subclasses

Instance variables

var client
Expand source code
class Page:
    """Paginated HDRezka catalog or listing page."""
    __slots__ = ('_page', '_page_format', '__yields', '__yields_page', 'client')

    def __init__(self, url: str, *, client: 'HDRezkaClient'):
        """
        :param url: Absolute page URL (or host root).
        :param client: Session used for HTTP requests.
        """
        self.client = client
        self.__yields: list[InlineItem] = []
        self.__yields_page = 0
        self.page = url

    @property
    def page(self) -> str:
        """Current page URL template base."""
        return self._page

    @page.setter
    def page(self, value):
        """Cast value to str and set the paginator format."""
        if not isinstance(value, str):
            value = str(value)
        # noinspection HttpUrlsUsage
        if not (value.startswith('https://') or value.startswith('http://')):
            value = f'https://{value}'
        self._page = value
        self._page_format = self._concat_paginator(value.removesuffix('/'))

    @staticmethod
    def _concat_paginator(url: str) -> Callable[[...], str]:
        def __concat_paginator(page, *args, **kwargs):
            if page is not None:
                return f'{url}/page/{{0}}/'.format(page, *args, **kwargs)
            return url.format(*args, **kwargs)

        return __concat_paginator

    @staticmethod
    def _inline_info(years: str, country: str, genre: str):
        year, *finals = years.split('-')
        if finals:
            final, = finals
            year_final = ... if final.strip() == '...' else int(final)
        else:
            year_final = None
        return InlineInfo(int(year), year_final, country.strip(), genre.strip())

    def _request_url(self, page: int | slice | Iterable[int] | Any, **query) -> str:
        """Build a paginated request URL with optional query parameters."""
        return merge_query(self._page_format(page), **query)

    def _parse_items(self, soup: BeautifulSoup) -> list[InlineItem]:
        items = soup.find_all(class_='b-content__inline_item')
        return [InlineItem(
            (a := (link := i.find(class_='b-content__inline_item-link')).find('a'))['href'],
            a.text,
            self._inline_info(*link.find('div').text.split(',', 3)),
            i.find(class_='b-content__inline_item-cover').find('img').get('src', ''),
            self.client,
        )
            for i in items]

    async def get_page_content(
            self,
            page: int | slice | Iterable[int] | None = None,
            *,
            filter: str | None = None,
            genre: str | int | None = None,
            **query,
    ) -> PageContent:
        """
        Fetch a catalog page and parse items, filters, series updates, and favorites cats.

        Extra keyword arguments are merged into the request query string
        (for example ``filter`` / ``genre`` from ``PageFilters``).
        Raises ``EmptyPage`` when no inline items are present.
        """
        if filter is not None:
            query['filter'] = filter
        if genre is not None:
            query['genre'] = genre
        html = (await self.client.get_response('GET', self._request_url(page, **query))).text
        soup = BeautifulSoup(html, builder=BUILDER)
        items = self._parse_items(soup)
        if not items:
            raise EmptyPage(page)
        return PageContent(
            items=items,
            filters=parse_page_filters(soup),
            series_updates=parse_series_updates(soup),
            favorites_cats=parse_favorites_cats(soup),
        )

    async def get_page(
            self,
            page: int | slice | Iterable[int] | None = None,
            *,
            filter: str | None = None,
            genre: str | int | None = None,
            **query,
    ) -> list[InlineItem]:
        """
        Fetch inline items for the given page number.

        Optional ``filter`` / ``genre`` (and other query kwargs) are appended
        to the request URL. Raises ``EmptyPage`` when the page has no items.
        """
        return (await self.get_page_content(page, filter=filter, genre=genre, **query)).items

    async def get_filters(
            self,
            page: int | slice | Iterable[int] = 1,
            *,
            filter: str | None = None,
            genre: str | int | None = None,
            **query,
    ) -> PageFilters:
        """
        Fetch the page and return available sort/type filters (empty if absent).

        Does not require inline items to be present.
        """
        if filter is not None:
            query['filter'] = filter
        if genre is not None:
            query['genre'] = genre
        html = (await self.client.get_response('GET', self._request_url(page, **query))).text
        return parse_page_filters(html)

    async def get_series_updates(self) -> tuple[SeriesUpdateBlock, ...]:
        """
        Fetch this page URL and parse sidebar series updates.

        Typically present on the site home page. Returns an empty tuple if missing.
        """
        html = (await self.client.get_response('GET', self._page)).text
        return parse_series_updates(html)

    async def get_navbar(self) -> Navbar:
        """Fetch this page URL and parse the top navigation menu."""
        html = (await self.client.get_response('GET', self._page)).text
        return parse_navbar(html)

    async def get_favorites_cats(self) -> tuple[FavoritesCat, ...]:
        """
        Fetch this page URL and parse favorites collection links.

        Present on ``/favorites/`` pages. Returns an empty tuple if missing.
        """
        html = (await self.client.get_response('GET', self._page)).text
        return parse_favorites_cats(html)

    def __aiter__(self):
        """Async iterator over items across successive pages."""
        self.__yields_page = 0
        self.__yields.clear()
        return self

    async def __anext__(self) -> InlineItem:
        """Return the next ``InlineItem`` across pages."""
        if not self.__yields:
            try:
                self.__yields_page += 1
                self.__yields += await self.get_page(self.__yields_page)
            except EmptyPage:
                raise StopAsyncIteration
        return self.__yields.pop(0)

    def __repr__(self):
        return f'{self.__class__.__qualname__}({self.page!r})'
prop page : str
Expand source code
@property
def page(self) -> str:
    """Current page URL template base."""
    return self._page

Current page URL template base.

Methods

async def get_favorites_cats(self) ‑> tuple[FavoritesCat, ...]
Expand source code
async def get_favorites_cats(self) -> tuple[FavoritesCat, ...]:
    """
    Fetch this page URL and parse favorites collection links.

    Present on ``/favorites/`` pages. Returns an empty tuple if missing.
    """
    html = (await self.client.get_response('GET', self._page)).text
    return parse_favorites_cats(html)

Fetch this page URL and parse favorites collection links.

Present on /favorites/ pages. Returns an empty tuple if missing.

async def get_filters(self,
page: int | slice | Iterable[int] = 1,
*,
filter: str | None = None,
genre: int | str | None = None,
**query) ‑> PageFilters
Expand source code
async def get_filters(
        self,
        page: int | slice | Iterable[int] = 1,
        *,
        filter: str | None = None,
        genre: str | int | None = None,
        **query,
) -> PageFilters:
    """
    Fetch the page and return available sort/type filters (empty if absent).

    Does not require inline items to be present.
    """
    if filter is not None:
        query['filter'] = filter
    if genre is not None:
        query['genre'] = genre
    html = (await self.client.get_response('GET', self._request_url(page, **query))).text
    return parse_page_filters(html)

Fetch the page and return available sort/type filters (empty if absent).

Does not require inline items to be present.

async def get_navbar(self) ‑> Navbar
Expand source code
async def get_navbar(self) -> Navbar:
    """Fetch this page URL and parse the top navigation menu."""
    html = (await self.client.get_response('GET', self._page)).text
    return parse_navbar(html)

Fetch this page URL and parse the top navigation menu.

async def get_page(self,
page: int | slice | Iterable[int] | None = None,
*,
filter: str | None = None,
genre: int | str | None = None,
**query) ‑> list[InlineItem]
Expand source code
async def get_page(
        self,
        page: int | slice | Iterable[int] | None = None,
        *,
        filter: str | None = None,
        genre: str | int | None = None,
        **query,
) -> list[InlineItem]:
    """
    Fetch inline items for the given page number.

    Optional ``filter`` / ``genre`` (and other query kwargs) are appended
    to the request URL. Raises ``EmptyPage`` when the page has no items.
    """
    return (await self.get_page_content(page, filter=filter, genre=genre, **query)).items

Fetch inline items for the given page number.

Optional filter / genre (and other query kwargs) are appended to the request URL. Raises EmptyPage when the page has no items.

async def get_page_content(self,
page: int | slice | Iterable[int] | None = None,
*,
filter: str | None = None,
genre: int | str | None = None,
**query) ‑> PageContent
Expand source code
async def get_page_content(
        self,
        page: int | slice | Iterable[int] | None = None,
        *,
        filter: str | None = None,
        genre: str | int | None = None,
        **query,
) -> PageContent:
    """
    Fetch a catalog page and parse items, filters, series updates, and favorites cats.

    Extra keyword arguments are merged into the request query string
    (for example ``filter`` / ``genre`` from ``PageFilters``).
    Raises ``EmptyPage`` when no inline items are present.
    """
    if filter is not None:
        query['filter'] = filter
    if genre is not None:
        query['genre'] = genre
    html = (await self.client.get_response('GET', self._request_url(page, **query))).text
    soup = BeautifulSoup(html, builder=BUILDER)
    items = self._parse_items(soup)
    if not items:
        raise EmptyPage(page)
    return PageContent(
        items=items,
        filters=parse_page_filters(soup),
        series_updates=parse_series_updates(soup),
        favorites_cats=parse_favorites_cats(soup),
    )

Fetch a catalog page and parse items, filters, series updates, and favorites cats.

Extra keyword arguments are merged into the request query string (for example filter / genre from PageFilters). Raises EmptyPage when no inline items are present.

async def get_series_updates(self) ‑> tuple[SeriesUpdateBlock, ...]
Expand source code
async def get_series_updates(self) -> tuple[SeriesUpdateBlock, ...]:
    """
    Fetch this page URL and parse sidebar series updates.

    Typically present on the site home page. Returns an empty tuple if missing.
    """
    html = (await self.client.get_response('GET', self._page)).text
    return parse_series_updates(html)

Fetch this page URL and parse sidebar series updates.

Typically present on the site home page. Returns an empty tuple if missing.

class PageContent (items: list[InlineItem],
filters: PageFilters = PageFilters(sorts=(), types=()),
series_updates: tuple[SeriesUpdateBlock, ...] = (),
favorites_cats: tuple[FavoritesCat, ...] = ())
Expand source code
class PageContent(NamedTuple):
    """Catalog page fetch result: items plus optional chrome parsed from the same HTML."""
    items: list[InlineItem]
    filters: PageFilters = PageFilters()
    series_updates: tuple[SeriesUpdateBlock, ...] = ()
    favorites_cats: tuple[FavoritesCat, ...] = ()

Catalog page fetch result: items plus optional chrome parsed from the same HTML.

Ancestors

  • builtins.tuple

Instance variables

var favorites_cats : tuple[FavoritesCat, ...]
Expand source code
class PageContent(NamedTuple):
    """Catalog page fetch result: items plus optional chrome parsed from the same HTML."""
    items: list[InlineItem]
    filters: PageFilters = PageFilters()
    series_updates: tuple[SeriesUpdateBlock, ...] = ()
    favorites_cats: tuple[FavoritesCat, ...] = ()

Alias for field number 3

var filtersPageFilters
Expand source code
class PageContent(NamedTuple):
    """Catalog page fetch result: items plus optional chrome parsed from the same HTML."""
    items: list[InlineItem]
    filters: PageFilters = PageFilters()
    series_updates: tuple[SeriesUpdateBlock, ...] = ()
    favorites_cats: tuple[FavoritesCat, ...] = ()

Alias for field number 1

var items : list[InlineItem]
Expand source code
class PageContent(NamedTuple):
    """Catalog page fetch result: items plus optional chrome parsed from the same HTML."""
    items: list[InlineItem]
    filters: PageFilters = PageFilters()
    series_updates: tuple[SeriesUpdateBlock, ...] = ()
    favorites_cats: tuple[FavoritesCat, ...] = ()

Alias for field number 0

var series_updates : tuple[SeriesUpdateBlock, ...]
Expand source code
class PageContent(NamedTuple):
    """Catalog page fetch result: items plus optional chrome parsed from the same HTML."""
    items: list[InlineItem]
    filters: PageFilters = PageFilters()
    series_updates: tuple[SeriesUpdateBlock, ...] = ()
    favorites_cats: tuple[FavoritesCat, ...] = ()

Alias for field number 2

class PageFilters (sorts: tuple[FilterLink, ...] = (),
types: tuple[FilterLink, ...] = ())
Expand source code
class PageFilters(NamedTuple):
    """
    Filters available on a catalog page.

    ``sorts`` — ``filter=`` controls (popular, watching, …).
    ``types`` — ``genre=`` content-type controls (films, anime, …).
    """
    sorts: tuple[FilterLink, ...] = ()
    types: tuple[FilterLink, ...] = ()

Filters available on a catalog page.

sortsfilter= controls (popular, watching, …). typesgenre= content-type controls (films, anime, …).

Ancestors

  • builtins.tuple

Instance variables

var sorts : tuple[FilterLink, ...]
Expand source code
class PageFilters(NamedTuple):
    """
    Filters available on a catalog page.

    ``sorts`` — ``filter=`` controls (popular, watching, …).
    ``types`` — ``genre=`` content-type controls (films, anime, …).
    """
    sorts: tuple[FilterLink, ...] = ()
    types: tuple[FilterLink, ...] = ()

Alias for field number 0

var types : tuple[FilterLink, ...]
Expand source code
class PageFilters(NamedTuple):
    """
    Filters available on a catalog page.

    ``sorts`` — ``filter=`` controls (popular, watching, …).
    ``types`` — ``genre=`` content-type controls (films, anime, …).
    """
    sorts: tuple[FilterLink, ...] = ()
    types: tuple[FilterLink, ...] = ()

Alias for field number 1

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

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

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

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

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

class Post (url: str,
*,
client: HDRezkaClient)
Expand source code
class Post:
    """Stores information about a title page (post)."""
    __slots__ = (
        'url', 'translator_id', 'id', 'name', 'type', 'info',
        'translators', 'franchises', 'schedule', 'client',
    )

    def __init__(self, url: str, *, client: 'HDRezkaClient'):
        """
        Need await.

        :param url: Absolute or relative post URL.
        :param client: Session used for HTTP requests and host joining.
        """
        self.client = client
        if not urllib.parse.urlparse(url).hostname:
            url = client.host_join(url)
        self.url = url

    def __await__(self):
        """
        Async initialize, prepare attributes.
        Do not call twice!
        """
        response = yield from self.client.get_response('GET', self.url).__await__()
        soup = BeautifulSoup(response.text, builder=BUILDER)
        self.type = soup.find('meta', property='og:type')['content'].removeprefix('video.')
        self.translator_id = self._get_translator_id(soup)
        self.info = get_post_info(soup, url=self.url, client=self.client)
        self.translators = self._get_translators(soup)

        self.id = int(soup.find(id='post_id')['value'])
        self.name = soup.find(class_='b-post__title').text.strip()
        franchises_url = soup.find(class_='b-post__franchise_link_title')
        self.franchises = FranchiseInfo(
            url=franchises_url and franchises_url.attrs.get('href'),
            soup=soup,
            client=self.client,
        )
        self.schedule = parse_schedule(soup)
        return self

    def _get_translator_id(self, soup: BeautifulSoup) -> int | None:
        """self.type must exist"""
        init_cdn_obj = 'sof.tv.%s' % {'tv_series': 'initCDNSeriesEvents', 'movie': 'initCDNMoviesEvents'}[self.type]
        for script in soup.find_all(lambda tag: tag.name == 'script' and not tag.attrs and tag.string):
            if not (s := script.string):
                continue
            obj_i = s.find(init_cdn_obj)
            if obj_i != -1:
                s = s[obj_i + len(init_cdn_obj):].split(',', 2)[1].strip()
                if s.isnumeric():
                    return int(s)
        return None

    def _get_translators(self, soup: BeautifulSoup) -> Translators:
        arr = {i.text.strip(): int(i['data-translator_id']) for i in el.find_all(recursive=False) if i.text
               } if (el := soup.find(id='translators-list')) else {}
        if not arr:
            arr[self.info.translators[0]] = self.translator_id
        return Translators(arr)

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.name!r}; {self.type!r}>'

Stores information about a title page (post).

Need await.

:param url: Absolute or relative post URL. :param client: Session used for HTTP requests and host joining.

Instance variables

var client
Expand source code
class Post:
    """Stores information about a title page (post)."""
    __slots__ = (
        'url', 'translator_id', 'id', 'name', 'type', 'info',
        'translators', 'franchises', 'schedule', 'client',
    )

    def __init__(self, url: str, *, client: 'HDRezkaClient'):
        """
        Need await.

        :param url: Absolute or relative post URL.
        :param client: Session used for HTTP requests and host joining.
        """
        self.client = client
        if not urllib.parse.urlparse(url).hostname:
            url = client.host_join(url)
        self.url = url

    def __await__(self):
        """
        Async initialize, prepare attributes.
        Do not call twice!
        """
        response = yield from self.client.get_response('GET', self.url).__await__()
        soup = BeautifulSoup(response.text, builder=BUILDER)
        self.type = soup.find('meta', property='og:type')['content'].removeprefix('video.')
        self.translator_id = self._get_translator_id(soup)
        self.info = get_post_info(soup, url=self.url, client=self.client)
        self.translators = self._get_translators(soup)

        self.id = int(soup.find(id='post_id')['value'])
        self.name = soup.find(class_='b-post__title').text.strip()
        franchises_url = soup.find(class_='b-post__franchise_link_title')
        self.franchises = FranchiseInfo(
            url=franchises_url and franchises_url.attrs.get('href'),
            soup=soup,
            client=self.client,
        )
        self.schedule = parse_schedule(soup)
        return self

    def _get_translator_id(self, soup: BeautifulSoup) -> int | None:
        """self.type must exist"""
        init_cdn_obj = 'sof.tv.%s' % {'tv_series': 'initCDNSeriesEvents', 'movie': 'initCDNMoviesEvents'}[self.type]
        for script in soup.find_all(lambda tag: tag.name == 'script' and not tag.attrs and tag.string):
            if not (s := script.string):
                continue
            obj_i = s.find(init_cdn_obj)
            if obj_i != -1:
                s = s[obj_i + len(init_cdn_obj):].split(',', 2)[1].strip()
                if s.isnumeric():
                    return int(s)
        return None

    def _get_translators(self, soup: BeautifulSoup) -> Translators:
        arr = {i.text.strip(): int(i['data-translator_id']) for i in el.find_all(recursive=False) if i.text
               } if (el := soup.find(id='translators-list')) else {}
        if not arr:
            arr[self.info.translators[0]] = self.translator_id
        return Translators(arr)

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.name!r}; {self.type!r}>'
var franchises
Expand source code
class Post:
    """Stores information about a title page (post)."""
    __slots__ = (
        'url', 'translator_id', 'id', 'name', 'type', 'info',
        'translators', 'franchises', 'schedule', 'client',
    )

    def __init__(self, url: str, *, client: 'HDRezkaClient'):
        """
        Need await.

        :param url: Absolute or relative post URL.
        :param client: Session used for HTTP requests and host joining.
        """
        self.client = client
        if not urllib.parse.urlparse(url).hostname:
            url = client.host_join(url)
        self.url = url

    def __await__(self):
        """
        Async initialize, prepare attributes.
        Do not call twice!
        """
        response = yield from self.client.get_response('GET', self.url).__await__()
        soup = BeautifulSoup(response.text, builder=BUILDER)
        self.type = soup.find('meta', property='og:type')['content'].removeprefix('video.')
        self.translator_id = self._get_translator_id(soup)
        self.info = get_post_info(soup, url=self.url, client=self.client)
        self.translators = self._get_translators(soup)

        self.id = int(soup.find(id='post_id')['value'])
        self.name = soup.find(class_='b-post__title').text.strip()
        franchises_url = soup.find(class_='b-post__franchise_link_title')
        self.franchises = FranchiseInfo(
            url=franchises_url and franchises_url.attrs.get('href'),
            soup=soup,
            client=self.client,
        )
        self.schedule = parse_schedule(soup)
        return self

    def _get_translator_id(self, soup: BeautifulSoup) -> int | None:
        """self.type must exist"""
        init_cdn_obj = 'sof.tv.%s' % {'tv_series': 'initCDNSeriesEvents', 'movie': 'initCDNMoviesEvents'}[self.type]
        for script in soup.find_all(lambda tag: tag.name == 'script' and not tag.attrs and tag.string):
            if not (s := script.string):
                continue
            obj_i = s.find(init_cdn_obj)
            if obj_i != -1:
                s = s[obj_i + len(init_cdn_obj):].split(',', 2)[1].strip()
                if s.isnumeric():
                    return int(s)
        return None

    def _get_translators(self, soup: BeautifulSoup) -> Translators:
        arr = {i.text.strip(): int(i['data-translator_id']) for i in el.find_all(recursive=False) if i.text
               } if (el := soup.find(id='translators-list')) else {}
        if not arr:
            arr[self.info.translators[0]] = self.translator_id
        return Translators(arr)

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.name!r}; {self.type!r}>'
var id
Expand source code
class Post:
    """Stores information about a title page (post)."""
    __slots__ = (
        'url', 'translator_id', 'id', 'name', 'type', 'info',
        'translators', 'franchises', 'schedule', 'client',
    )

    def __init__(self, url: str, *, client: 'HDRezkaClient'):
        """
        Need await.

        :param url: Absolute or relative post URL.
        :param client: Session used for HTTP requests and host joining.
        """
        self.client = client
        if not urllib.parse.urlparse(url).hostname:
            url = client.host_join(url)
        self.url = url

    def __await__(self):
        """
        Async initialize, prepare attributes.
        Do not call twice!
        """
        response = yield from self.client.get_response('GET', self.url).__await__()
        soup = BeautifulSoup(response.text, builder=BUILDER)
        self.type = soup.find('meta', property='og:type')['content'].removeprefix('video.')
        self.translator_id = self._get_translator_id(soup)
        self.info = get_post_info(soup, url=self.url, client=self.client)
        self.translators = self._get_translators(soup)

        self.id = int(soup.find(id='post_id')['value'])
        self.name = soup.find(class_='b-post__title').text.strip()
        franchises_url = soup.find(class_='b-post__franchise_link_title')
        self.franchises = FranchiseInfo(
            url=franchises_url and franchises_url.attrs.get('href'),
            soup=soup,
            client=self.client,
        )
        self.schedule = parse_schedule(soup)
        return self

    def _get_translator_id(self, soup: BeautifulSoup) -> int | None:
        """self.type must exist"""
        init_cdn_obj = 'sof.tv.%s' % {'tv_series': 'initCDNSeriesEvents', 'movie': 'initCDNMoviesEvents'}[self.type]
        for script in soup.find_all(lambda tag: tag.name == 'script' and not tag.attrs and tag.string):
            if not (s := script.string):
                continue
            obj_i = s.find(init_cdn_obj)
            if obj_i != -1:
                s = s[obj_i + len(init_cdn_obj):].split(',', 2)[1].strip()
                if s.isnumeric():
                    return int(s)
        return None

    def _get_translators(self, soup: BeautifulSoup) -> Translators:
        arr = {i.text.strip(): int(i['data-translator_id']) for i in el.find_all(recursive=False) if i.text
               } if (el := soup.find(id='translators-list')) else {}
        if not arr:
            arr[self.info.translators[0]] = self.translator_id
        return Translators(arr)

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.name!r}; {self.type!r}>'
var info
Expand source code
class Post:
    """Stores information about a title page (post)."""
    __slots__ = (
        'url', 'translator_id', 'id', 'name', 'type', 'info',
        'translators', 'franchises', 'schedule', 'client',
    )

    def __init__(self, url: str, *, client: 'HDRezkaClient'):
        """
        Need await.

        :param url: Absolute or relative post URL.
        :param client: Session used for HTTP requests and host joining.
        """
        self.client = client
        if not urllib.parse.urlparse(url).hostname:
            url = client.host_join(url)
        self.url = url

    def __await__(self):
        """
        Async initialize, prepare attributes.
        Do not call twice!
        """
        response = yield from self.client.get_response('GET', self.url).__await__()
        soup = BeautifulSoup(response.text, builder=BUILDER)
        self.type = soup.find('meta', property='og:type')['content'].removeprefix('video.')
        self.translator_id = self._get_translator_id(soup)
        self.info = get_post_info(soup, url=self.url, client=self.client)
        self.translators = self._get_translators(soup)

        self.id = int(soup.find(id='post_id')['value'])
        self.name = soup.find(class_='b-post__title').text.strip()
        franchises_url = soup.find(class_='b-post__franchise_link_title')
        self.franchises = FranchiseInfo(
            url=franchises_url and franchises_url.attrs.get('href'),
            soup=soup,
            client=self.client,
        )
        self.schedule = parse_schedule(soup)
        return self

    def _get_translator_id(self, soup: BeautifulSoup) -> int | None:
        """self.type must exist"""
        init_cdn_obj = 'sof.tv.%s' % {'tv_series': 'initCDNSeriesEvents', 'movie': 'initCDNMoviesEvents'}[self.type]
        for script in soup.find_all(lambda tag: tag.name == 'script' and not tag.attrs and tag.string):
            if not (s := script.string):
                continue
            obj_i = s.find(init_cdn_obj)
            if obj_i != -1:
                s = s[obj_i + len(init_cdn_obj):].split(',', 2)[1].strip()
                if s.isnumeric():
                    return int(s)
        return None

    def _get_translators(self, soup: BeautifulSoup) -> Translators:
        arr = {i.text.strip(): int(i['data-translator_id']) for i in el.find_all(recursive=False) if i.text
               } if (el := soup.find(id='translators-list')) else {}
        if not arr:
            arr[self.info.translators[0]] = self.translator_id
        return Translators(arr)

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.name!r}; {self.type!r}>'
var name
Expand source code
class Post:
    """Stores information about a title page (post)."""
    __slots__ = (
        'url', 'translator_id', 'id', 'name', 'type', 'info',
        'translators', 'franchises', 'schedule', 'client',
    )

    def __init__(self, url: str, *, client: 'HDRezkaClient'):
        """
        Need await.

        :param url: Absolute or relative post URL.
        :param client: Session used for HTTP requests and host joining.
        """
        self.client = client
        if not urllib.parse.urlparse(url).hostname:
            url = client.host_join(url)
        self.url = url

    def __await__(self):
        """
        Async initialize, prepare attributes.
        Do not call twice!
        """
        response = yield from self.client.get_response('GET', self.url).__await__()
        soup = BeautifulSoup(response.text, builder=BUILDER)
        self.type = soup.find('meta', property='og:type')['content'].removeprefix('video.')
        self.translator_id = self._get_translator_id(soup)
        self.info = get_post_info(soup, url=self.url, client=self.client)
        self.translators = self._get_translators(soup)

        self.id = int(soup.find(id='post_id')['value'])
        self.name = soup.find(class_='b-post__title').text.strip()
        franchises_url = soup.find(class_='b-post__franchise_link_title')
        self.franchises = FranchiseInfo(
            url=franchises_url and franchises_url.attrs.get('href'),
            soup=soup,
            client=self.client,
        )
        self.schedule = parse_schedule(soup)
        return self

    def _get_translator_id(self, soup: BeautifulSoup) -> int | None:
        """self.type must exist"""
        init_cdn_obj = 'sof.tv.%s' % {'tv_series': 'initCDNSeriesEvents', 'movie': 'initCDNMoviesEvents'}[self.type]
        for script in soup.find_all(lambda tag: tag.name == 'script' and not tag.attrs and tag.string):
            if not (s := script.string):
                continue
            obj_i = s.find(init_cdn_obj)
            if obj_i != -1:
                s = s[obj_i + len(init_cdn_obj):].split(',', 2)[1].strip()
                if s.isnumeric():
                    return int(s)
        return None

    def _get_translators(self, soup: BeautifulSoup) -> Translators:
        arr = {i.text.strip(): int(i['data-translator_id']) for i in el.find_all(recursive=False) if i.text
               } if (el := soup.find(id='translators-list')) else {}
        if not arr:
            arr[self.info.translators[0]] = self.translator_id
        return Translators(arr)

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.name!r}; {self.type!r}>'
var schedule
Expand source code
class Post:
    """Stores information about a title page (post)."""
    __slots__ = (
        'url', 'translator_id', 'id', 'name', 'type', 'info',
        'translators', 'franchises', 'schedule', 'client',
    )

    def __init__(self, url: str, *, client: 'HDRezkaClient'):
        """
        Need await.

        :param url: Absolute or relative post URL.
        :param client: Session used for HTTP requests and host joining.
        """
        self.client = client
        if not urllib.parse.urlparse(url).hostname:
            url = client.host_join(url)
        self.url = url

    def __await__(self):
        """
        Async initialize, prepare attributes.
        Do not call twice!
        """
        response = yield from self.client.get_response('GET', self.url).__await__()
        soup = BeautifulSoup(response.text, builder=BUILDER)
        self.type = soup.find('meta', property='og:type')['content'].removeprefix('video.')
        self.translator_id = self._get_translator_id(soup)
        self.info = get_post_info(soup, url=self.url, client=self.client)
        self.translators = self._get_translators(soup)

        self.id = int(soup.find(id='post_id')['value'])
        self.name = soup.find(class_='b-post__title').text.strip()
        franchises_url = soup.find(class_='b-post__franchise_link_title')
        self.franchises = FranchiseInfo(
            url=franchises_url and franchises_url.attrs.get('href'),
            soup=soup,
            client=self.client,
        )
        self.schedule = parse_schedule(soup)
        return self

    def _get_translator_id(self, soup: BeautifulSoup) -> int | None:
        """self.type must exist"""
        init_cdn_obj = 'sof.tv.%s' % {'tv_series': 'initCDNSeriesEvents', 'movie': 'initCDNMoviesEvents'}[self.type]
        for script in soup.find_all(lambda tag: tag.name == 'script' and not tag.attrs and tag.string):
            if not (s := script.string):
                continue
            obj_i = s.find(init_cdn_obj)
            if obj_i != -1:
                s = s[obj_i + len(init_cdn_obj):].split(',', 2)[1].strip()
                if s.isnumeric():
                    return int(s)
        return None

    def _get_translators(self, soup: BeautifulSoup) -> Translators:
        arr = {i.text.strip(): int(i['data-translator_id']) for i in el.find_all(recursive=False) if i.text
               } if (el := soup.find(id='translators-list')) else {}
        if not arr:
            arr[self.info.translators[0]] = self.translator_id
        return Translators(arr)

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.name!r}; {self.type!r}>'
var translator_id
Expand source code
class Post:
    """Stores information about a title page (post)."""
    __slots__ = (
        'url', 'translator_id', 'id', 'name', 'type', 'info',
        'translators', 'franchises', 'schedule', 'client',
    )

    def __init__(self, url: str, *, client: 'HDRezkaClient'):
        """
        Need await.

        :param url: Absolute or relative post URL.
        :param client: Session used for HTTP requests and host joining.
        """
        self.client = client
        if not urllib.parse.urlparse(url).hostname:
            url = client.host_join(url)
        self.url = url

    def __await__(self):
        """
        Async initialize, prepare attributes.
        Do not call twice!
        """
        response = yield from self.client.get_response('GET', self.url).__await__()
        soup = BeautifulSoup(response.text, builder=BUILDER)
        self.type = soup.find('meta', property='og:type')['content'].removeprefix('video.')
        self.translator_id = self._get_translator_id(soup)
        self.info = get_post_info(soup, url=self.url, client=self.client)
        self.translators = self._get_translators(soup)

        self.id = int(soup.find(id='post_id')['value'])
        self.name = soup.find(class_='b-post__title').text.strip()
        franchises_url = soup.find(class_='b-post__franchise_link_title')
        self.franchises = FranchiseInfo(
            url=franchises_url and franchises_url.attrs.get('href'),
            soup=soup,
            client=self.client,
        )
        self.schedule = parse_schedule(soup)
        return self

    def _get_translator_id(self, soup: BeautifulSoup) -> int | None:
        """self.type must exist"""
        init_cdn_obj = 'sof.tv.%s' % {'tv_series': 'initCDNSeriesEvents', 'movie': 'initCDNMoviesEvents'}[self.type]
        for script in soup.find_all(lambda tag: tag.name == 'script' and not tag.attrs and tag.string):
            if not (s := script.string):
                continue
            obj_i = s.find(init_cdn_obj)
            if obj_i != -1:
                s = s[obj_i + len(init_cdn_obj):].split(',', 2)[1].strip()
                if s.isnumeric():
                    return int(s)
        return None

    def _get_translators(self, soup: BeautifulSoup) -> Translators:
        arr = {i.text.strip(): int(i['data-translator_id']) for i in el.find_all(recursive=False) if i.text
               } if (el := soup.find(id='translators-list')) else {}
        if not arr:
            arr[self.info.translators[0]] = self.translator_id
        return Translators(arr)

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.name!r}; {self.type!r}>'
var translators
Expand source code
class Post:
    """Stores information about a title page (post)."""
    __slots__ = (
        'url', 'translator_id', 'id', 'name', 'type', 'info',
        'translators', 'franchises', 'schedule', 'client',
    )

    def __init__(self, url: str, *, client: 'HDRezkaClient'):
        """
        Need await.

        :param url: Absolute or relative post URL.
        :param client: Session used for HTTP requests and host joining.
        """
        self.client = client
        if not urllib.parse.urlparse(url).hostname:
            url = client.host_join(url)
        self.url = url

    def __await__(self):
        """
        Async initialize, prepare attributes.
        Do not call twice!
        """
        response = yield from self.client.get_response('GET', self.url).__await__()
        soup = BeautifulSoup(response.text, builder=BUILDER)
        self.type = soup.find('meta', property='og:type')['content'].removeprefix('video.')
        self.translator_id = self._get_translator_id(soup)
        self.info = get_post_info(soup, url=self.url, client=self.client)
        self.translators = self._get_translators(soup)

        self.id = int(soup.find(id='post_id')['value'])
        self.name = soup.find(class_='b-post__title').text.strip()
        franchises_url = soup.find(class_='b-post__franchise_link_title')
        self.franchises = FranchiseInfo(
            url=franchises_url and franchises_url.attrs.get('href'),
            soup=soup,
            client=self.client,
        )
        self.schedule = parse_schedule(soup)
        return self

    def _get_translator_id(self, soup: BeautifulSoup) -> int | None:
        """self.type must exist"""
        init_cdn_obj = 'sof.tv.%s' % {'tv_series': 'initCDNSeriesEvents', 'movie': 'initCDNMoviesEvents'}[self.type]
        for script in soup.find_all(lambda tag: tag.name == 'script' and not tag.attrs and tag.string):
            if not (s := script.string):
                continue
            obj_i = s.find(init_cdn_obj)
            if obj_i != -1:
                s = s[obj_i + len(init_cdn_obj):].split(',', 2)[1].strip()
                if s.isnumeric():
                    return int(s)
        return None

    def _get_translators(self, soup: BeautifulSoup) -> Translators:
        arr = {i.text.strip(): int(i['data-translator_id']) for i in el.find_all(recursive=False) if i.text
               } if (el := soup.find(id='translators-list')) else {}
        if not arr:
            arr[self.info.translators[0]] = self.translator_id
        return Translators(arr)

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.name!r}; {self.type!r}>'
var type
Expand source code
class Post:
    """Stores information about a title page (post)."""
    __slots__ = (
        'url', 'translator_id', 'id', 'name', 'type', 'info',
        'translators', 'franchises', 'schedule', 'client',
    )

    def __init__(self, url: str, *, client: 'HDRezkaClient'):
        """
        Need await.

        :param url: Absolute or relative post URL.
        :param client: Session used for HTTP requests and host joining.
        """
        self.client = client
        if not urllib.parse.urlparse(url).hostname:
            url = client.host_join(url)
        self.url = url

    def __await__(self):
        """
        Async initialize, prepare attributes.
        Do not call twice!
        """
        response = yield from self.client.get_response('GET', self.url).__await__()
        soup = BeautifulSoup(response.text, builder=BUILDER)
        self.type = soup.find('meta', property='og:type')['content'].removeprefix('video.')
        self.translator_id = self._get_translator_id(soup)
        self.info = get_post_info(soup, url=self.url, client=self.client)
        self.translators = self._get_translators(soup)

        self.id = int(soup.find(id='post_id')['value'])
        self.name = soup.find(class_='b-post__title').text.strip()
        franchises_url = soup.find(class_='b-post__franchise_link_title')
        self.franchises = FranchiseInfo(
            url=franchises_url and franchises_url.attrs.get('href'),
            soup=soup,
            client=self.client,
        )
        self.schedule = parse_schedule(soup)
        return self

    def _get_translator_id(self, soup: BeautifulSoup) -> int | None:
        """self.type must exist"""
        init_cdn_obj = 'sof.tv.%s' % {'tv_series': 'initCDNSeriesEvents', 'movie': 'initCDNMoviesEvents'}[self.type]
        for script in soup.find_all(lambda tag: tag.name == 'script' and not tag.attrs and tag.string):
            if not (s := script.string):
                continue
            obj_i = s.find(init_cdn_obj)
            if obj_i != -1:
                s = s[obj_i + len(init_cdn_obj):].split(',', 2)[1].strip()
                if s.isnumeric():
                    return int(s)
        return None

    def _get_translators(self, soup: BeautifulSoup) -> Translators:
        arr = {i.text.strip(): int(i['data-translator_id']) for i in el.find_all(recursive=False) if i.text
               } if (el := soup.find(id='translators-list')) else {}
        if not arr:
            arr[self.info.translators[0]] = self.translator_id
        return Translators(arr)

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.name!r}; {self.type!r}>'
var url
Expand source code
class Post:
    """Stores information about a title page (post)."""
    __slots__ = (
        'url', 'translator_id', 'id', 'name', 'type', 'info',
        'translators', 'franchises', 'schedule', 'client',
    )

    def __init__(self, url: str, *, client: 'HDRezkaClient'):
        """
        Need await.

        :param url: Absolute or relative post URL.
        :param client: Session used for HTTP requests and host joining.
        """
        self.client = client
        if not urllib.parse.urlparse(url).hostname:
            url = client.host_join(url)
        self.url = url

    def __await__(self):
        """
        Async initialize, prepare attributes.
        Do not call twice!
        """
        response = yield from self.client.get_response('GET', self.url).__await__()
        soup = BeautifulSoup(response.text, builder=BUILDER)
        self.type = soup.find('meta', property='og:type')['content'].removeprefix('video.')
        self.translator_id = self._get_translator_id(soup)
        self.info = get_post_info(soup, url=self.url, client=self.client)
        self.translators = self._get_translators(soup)

        self.id = int(soup.find(id='post_id')['value'])
        self.name = soup.find(class_='b-post__title').text.strip()
        franchises_url = soup.find(class_='b-post__franchise_link_title')
        self.franchises = FranchiseInfo(
            url=franchises_url and franchises_url.attrs.get('href'),
            soup=soup,
            client=self.client,
        )
        self.schedule = parse_schedule(soup)
        return self

    def _get_translator_id(self, soup: BeautifulSoup) -> int | None:
        """self.type must exist"""
        init_cdn_obj = 'sof.tv.%s' % {'tv_series': 'initCDNSeriesEvents', 'movie': 'initCDNMoviesEvents'}[self.type]
        for script in soup.find_all(lambda tag: tag.name == 'script' and not tag.attrs and tag.string):
            if not (s := script.string):
                continue
            obj_i = s.find(init_cdn_obj)
            if obj_i != -1:
                s = s[obj_i + len(init_cdn_obj):].split(',', 2)[1].strip()
                if s.isnumeric():
                    return int(s)
        return None

    def _get_translators(self, soup: BeautifulSoup) -> Translators:
        arr = {i.text.strip(): int(i['data-translator_id']) for i in el.find_all(recursive=False) if i.text
               } if (el := soup.find(id='translators-list')) else {}
        if not arr:
            arr[self.info.translators[0]] = self.translator_id
        return Translators(arr)

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.name!r}; {self.type!r}>'
class Quality (val: str)
Expand source code
class Quality(str):
    """str type add-on to represent video quality"""

    __slots__ = ('_i', 'addon', 'units')

    def __new__(cls, val: str):
        clean_val = _clean_html('', val).strip()
        # noinspection PyTypeChecker
        return super().__new__(cls, clean_val)

    def __init__(self, val: str):
        match = _match_quality_int(self)
        if not match:
            raise ValueError(f'{val!r} is unknown quality.')
        _i, units, self.addon = match.groups()
        _i = _i or '0'
        if units.upper() == 'K':
            _i += '000'
            units = 'p'
        self.addon = self.addon.casefold()
        self._i = int(_i)
        self.units = units

    def __int__(self):
        """
        returns pixels height
        """
        return self._i

    def __lt__(self, other):
        """Is other quality better then self"""
        if not isinstance(other, self.__class__):
            return super().__le__(other)
        if self._i != other._i:
            return self._i < other._i
        return not self.addon and bool(other.addon)

str type add-on to represent video quality

Ancestors

  • builtins.str

Instance variables

var addon
Expand source code
class Quality(str):
    """str type add-on to represent video quality"""

    __slots__ = ('_i', 'addon', 'units')

    def __new__(cls, val: str):
        clean_val = _clean_html('', val).strip()
        # noinspection PyTypeChecker
        return super().__new__(cls, clean_val)

    def __init__(self, val: str):
        match = _match_quality_int(self)
        if not match:
            raise ValueError(f'{val!r} is unknown quality.')
        _i, units, self.addon = match.groups()
        _i = _i or '0'
        if units.upper() == 'K':
            _i += '000'
            units = 'p'
        self.addon = self.addon.casefold()
        self._i = int(_i)
        self.units = units

    def __int__(self):
        """
        returns pixels height
        """
        return self._i

    def __lt__(self, other):
        """Is other quality better then self"""
        if not isinstance(other, self.__class__):
            return super().__le__(other)
        if self._i != other._i:
            return self._i < other._i
        return not self.addon and bool(other.addon)
var units
Expand source code
class Quality(str):
    """str type add-on to represent video quality"""

    __slots__ = ('_i', 'addon', 'units')

    def __new__(cls, val: str):
        clean_val = _clean_html('', val).strip()
        # noinspection PyTypeChecker
        return super().__new__(cls, clean_val)

    def __init__(self, val: str):
        match = _match_quality_int(self)
        if not match:
            raise ValueError(f'{val!r} is unknown quality.')
        _i, units, self.addon = match.groups()
        _i = _i or '0'
        if units.upper() == 'K':
            _i += '000'
            units = 'p'
        self.addon = self.addon.casefold()
        self._i = int(_i)
        self.units = units

    def __int__(self):
        """
        returns pixels height
        """
        return self._i

    def __lt__(self, other):
        """Is other quality better then self"""
        if not isinstance(other, self.__class__):
            return super().__le__(other)
        if self._i != other._i:
            return self._i < other._i
        return not self.addon and bool(other.addon)
class ScheduleBlock (title: str,
items: tuple[ScheduleItem, ...] = ())
Expand source code
class ScheduleBlock(NamedTuple):
    """Season schedule block under one heading."""
    title: str
    items: tuple[ScheduleItem, ...] = ()

Season schedule block under one heading.

Ancestors

  • builtins.tuple

Instance variables

var items : tuple[ScheduleItem, ...]
Expand source code
class ScheduleBlock(NamedTuple):
    """Season schedule block under one heading."""
    title: str
    items: tuple[ScheduleItem, ...] = ()

Alias for field number 1

var title : str
Expand source code
class ScheduleBlock(NamedTuple):
    """Season schedule block under one heading."""
    title: str
    items: tuple[ScheduleItem, ...] = ()

Alias for field number 0

class ScheduleItem (id: int,
season: int = 0,
episode: int = 0,
title: str = '',
original_title: str = '',
date: str = '',
exists: bool = False)
Expand source code
class ScheduleItem(NamedTuple):
    """One episode row from a post schedule table."""
    id: int
    season: int = 0
    episode: int = 0
    title: str = ''
    original_title: str = ''
    date: str = ''
    exists: bool = False

One episode row from a post schedule table.

Ancestors

  • builtins.tuple

Instance variables

var date : str
Expand source code
class ScheduleItem(NamedTuple):
    """One episode row from a post schedule table."""
    id: int
    season: int = 0
    episode: int = 0
    title: str = ''
    original_title: str = ''
    date: str = ''
    exists: bool = False

Alias for field number 5

var episode : int
Expand source code
class ScheduleItem(NamedTuple):
    """One episode row from a post schedule table."""
    id: int
    season: int = 0
    episode: int = 0
    title: str = ''
    original_title: str = ''
    date: str = ''
    exists: bool = False

Alias for field number 2

var exists : bool
Expand source code
class ScheduleItem(NamedTuple):
    """One episode row from a post schedule table."""
    id: int
    season: int = 0
    episode: int = 0
    title: str = ''
    original_title: str = ''
    date: str = ''
    exists: bool = False

Alias for field number 6

var id : int
Expand source code
class ScheduleItem(NamedTuple):
    """One episode row from a post schedule table."""
    id: int
    season: int = 0
    episode: int = 0
    title: str = ''
    original_title: str = ''
    date: str = ''
    exists: bool = False

Alias for field number 0

var original_title : str
Expand source code
class ScheduleItem(NamedTuple):
    """One episode row from a post schedule table."""
    id: int
    season: int = 0
    episode: int = 0
    title: str = ''
    original_title: str = ''
    date: str = ''
    exists: bool = False

Alias for field number 4

var season : int
Expand source code
class ScheduleItem(NamedTuple):
    """One episode row from a post schedule table."""
    id: int
    season: int = 0
    episode: int = 0
    title: str = ''
    original_title: str = ''
    date: str = ''
    exists: bool = False

Alias for field number 1

var title : str
Expand source code
class ScheduleItem(NamedTuple):
    """One episode row from a post schedule table."""
    id: int
    season: int = 0
    episode: int = 0
    title: str = ''
    original_title: str = ''
    date: str = ''
    exists: bool = False

Alias for field number 3

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

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

class SeriesUpdateBlock (date: str,
items: tuple[SeriesUpdateItem, ...] = ())
Expand source code
class SeriesUpdateBlock(NamedTuple):
    """Series updates grouped under one date heading."""
    date: str
    items: tuple[SeriesUpdateItem, ...] = ()

Series updates grouped under one date heading.

Ancestors

  • builtins.tuple

Instance variables

var date : str
Expand source code
class SeriesUpdateBlock(NamedTuple):
    """Series updates grouped under one date heading."""
    date: str
    items: tuple[SeriesUpdateItem, ...] = ()

Alias for field number 0

var items : tuple[SeriesUpdateItem, ...]
Expand source code
class SeriesUpdateBlock(NamedTuple):
    """Series updates grouped under one date heading."""
    date: str
    items: tuple[SeriesUpdateItem, ...] = ()

Alias for field number 1

class SeriesUpdateItem (name: str, url: str, season: str = '', episode: str = '', translation: str = '')
Expand source code
class SeriesUpdateItem(NamedTuple):
    """One series episode update row."""
    name: str
    url: str
    season: str = ''
    episode: str = ''
    translation: str = ''

One series episode update row.

Ancestors

  • builtins.tuple

Instance variables

var episode : str
Expand source code
class SeriesUpdateItem(NamedTuple):
    """One series episode update row."""
    name: str
    url: str
    season: str = ''
    episode: str = ''
    translation: str = ''

Alias for field number 3

var name : str
Expand source code
class SeriesUpdateItem(NamedTuple):
    """One series episode update row."""
    name: str
    url: str
    season: str = ''
    episode: str = ''
    translation: str = ''

Alias for field number 0

var season : str
Expand source code
class SeriesUpdateItem(NamedTuple):
    """One series episode update row."""
    name: str
    url: str
    season: str = ''
    episode: str = ''
    translation: str = ''

Alias for field number 2

var translation : str
Expand source code
class SeriesUpdateItem(NamedTuple):
    """One series episode update row."""
    name: str
    url: str
    season: str = ''
    episode: str = ''
    translation: str = ''

Alias for field number 4

var url : str
Expand source code
class SeriesUpdateItem(NamedTuple):
    """One series episode update row."""
    name: str
    url: str
    season: str = ''
    episode: str = ''
    translation: str = ''

Alias for field number 1

class SubtitleURL (url: str, name: str, code: str)
Expand source code
class SubtitleURL(NamedTuple):
    """
    url: str
        .vtt file url

    Language attributes:
    name: str
    code: str
    """
    url: str
    name: str
    code: str

url: str .vtt file url

Language attributes: name: str code: str

Ancestors

  • builtins.tuple

Instance variables

var code : str
Expand source code
class SubtitleURL(NamedTuple):
    """
    url: str
        .vtt file url

    Language attributes:
    name: str
    code: str
    """
    url: str
    name: str
    code: str

Alias for field number 2

var name : str
Expand source code
class SubtitleURL(NamedTuple):
    """
    url: str
        .vtt file url

    Language attributes:
    name: str
    code: str
    """
    url: str
    name: str
    code: str

Alias for field number 1

var url : str
Expand source code
class SubtitleURL(NamedTuple):
    """
    url: str
        .vtt file url

    Language attributes:
    name: str
    code: str
    """
    url: str
    name: str
    code: str

Alias for field number 0

class SubtitleURLs (subtitle: str, subtitle_lns: dict[str, str], subtitle_def: str)
Expand source code
class SubtitleURLs:
    """Class representing subtitle urls"""
    __slots__ = ('subtitles', 'has_subtitles', 'subtitle_names', 'subtitle_codes', 'default')

    def __init__(self, subtitle: str, subtitle_lns: dict[str, str], subtitle_def: str):
        """
        :param subtitle: is subtitles exists
        :param subtitle_lns: languages {code: name, ...}
        :param subtitle_def: default subtitle code
        """
        self.has_subtitles = not not subtitle
        off = SubtitleURL('', '', 'off'),
        if self.has_subtitles:
            self.subtitles: tuple[SubtitleURL, ...] = *(
                SubtitleURL(url, name, subtitle_lns[name]) for name, url in
                (v.removeprefix('[').split(']', 1) for v in subtitle.split(','))),
            self.subtitles += off
        else:
            self.subtitles = off
        self.subtitle_names = {}
        self.subtitle_codes = {}
        for subtitle_item in self.subtitles:
            self.subtitle_names[subtitle_item.name] = self.subtitle_codes[subtitle_item.code] = subtitle_item
        self.default: SubtitleURL | None = self.subtitle_codes.get(subtitle_def)

    def __getitem__(self, item: str) -> SubtitleURL:
        """Returns subtitle by name or code"""
        if item in self.subtitle_names:
            return self.subtitle_names[item]
        return self.subtitle_codes[item]

    def get(self, item: str) -> SubtitleURL | None:
        """Returns subtitle by name or code, if not found returns None"""
        if not self.has_subtitles:
            return None
        return self.subtitle_names.get(item) or self.subtitle_codes.get(item)

    def __getattr__(self, item: str) -> SubtitleURL:
        """Returns subtitle by code"""
        return self.subtitle_codes[item]

    def __bool__(self):
        """Is subtitles exists"""
        return self.has_subtitles

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.subtitles!r}>'

Class representing subtitle urls

:param subtitle: is subtitles exists :param subtitle_lns: languages {code: name, …} :param subtitle_def: default subtitle code

Instance variables

var default
Expand source code
class SubtitleURLs:
    """Class representing subtitle urls"""
    __slots__ = ('subtitles', 'has_subtitles', 'subtitle_names', 'subtitle_codes', 'default')

    def __init__(self, subtitle: str, subtitle_lns: dict[str, str], subtitle_def: str):
        """
        :param subtitle: is subtitles exists
        :param subtitle_lns: languages {code: name, ...}
        :param subtitle_def: default subtitle code
        """
        self.has_subtitles = not not subtitle
        off = SubtitleURL('', '', 'off'),
        if self.has_subtitles:
            self.subtitles: tuple[SubtitleURL, ...] = *(
                SubtitleURL(url, name, subtitle_lns[name]) for name, url in
                (v.removeprefix('[').split(']', 1) for v in subtitle.split(','))),
            self.subtitles += off
        else:
            self.subtitles = off
        self.subtitle_names = {}
        self.subtitle_codes = {}
        for subtitle_item in self.subtitles:
            self.subtitle_names[subtitle_item.name] = self.subtitle_codes[subtitle_item.code] = subtitle_item
        self.default: SubtitleURL | None = self.subtitle_codes.get(subtitle_def)

    def __getitem__(self, item: str) -> SubtitleURL:
        """Returns subtitle by name or code"""
        if item in self.subtitle_names:
            return self.subtitle_names[item]
        return self.subtitle_codes[item]

    def get(self, item: str) -> SubtitleURL | None:
        """Returns subtitle by name or code, if not found returns None"""
        if not self.has_subtitles:
            return None
        return self.subtitle_names.get(item) or self.subtitle_codes.get(item)

    def __getattr__(self, item: str) -> SubtitleURL:
        """Returns subtitle by code"""
        return self.subtitle_codes[item]

    def __bool__(self):
        """Is subtitles exists"""
        return self.has_subtitles

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.subtitles!r}>'
var has_subtitles
Expand source code
class SubtitleURLs:
    """Class representing subtitle urls"""
    __slots__ = ('subtitles', 'has_subtitles', 'subtitle_names', 'subtitle_codes', 'default')

    def __init__(self, subtitle: str, subtitle_lns: dict[str, str], subtitle_def: str):
        """
        :param subtitle: is subtitles exists
        :param subtitle_lns: languages {code: name, ...}
        :param subtitle_def: default subtitle code
        """
        self.has_subtitles = not not subtitle
        off = SubtitleURL('', '', 'off'),
        if self.has_subtitles:
            self.subtitles: tuple[SubtitleURL, ...] = *(
                SubtitleURL(url, name, subtitle_lns[name]) for name, url in
                (v.removeprefix('[').split(']', 1) for v in subtitle.split(','))),
            self.subtitles += off
        else:
            self.subtitles = off
        self.subtitle_names = {}
        self.subtitle_codes = {}
        for subtitle_item in self.subtitles:
            self.subtitle_names[subtitle_item.name] = self.subtitle_codes[subtitle_item.code] = subtitle_item
        self.default: SubtitleURL | None = self.subtitle_codes.get(subtitle_def)

    def __getitem__(self, item: str) -> SubtitleURL:
        """Returns subtitle by name or code"""
        if item in self.subtitle_names:
            return self.subtitle_names[item]
        return self.subtitle_codes[item]

    def get(self, item: str) -> SubtitleURL | None:
        """Returns subtitle by name or code, if not found returns None"""
        if not self.has_subtitles:
            return None
        return self.subtitle_names.get(item) or self.subtitle_codes.get(item)

    def __getattr__(self, item: str) -> SubtitleURL:
        """Returns subtitle by code"""
        return self.subtitle_codes[item]

    def __bool__(self):
        """Is subtitles exists"""
        return self.has_subtitles

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.subtitles!r}>'
var subtitle_codes
Expand source code
class SubtitleURLs:
    """Class representing subtitle urls"""
    __slots__ = ('subtitles', 'has_subtitles', 'subtitle_names', 'subtitle_codes', 'default')

    def __init__(self, subtitle: str, subtitle_lns: dict[str, str], subtitle_def: str):
        """
        :param subtitle: is subtitles exists
        :param subtitle_lns: languages {code: name, ...}
        :param subtitle_def: default subtitle code
        """
        self.has_subtitles = not not subtitle
        off = SubtitleURL('', '', 'off'),
        if self.has_subtitles:
            self.subtitles: tuple[SubtitleURL, ...] = *(
                SubtitleURL(url, name, subtitle_lns[name]) for name, url in
                (v.removeprefix('[').split(']', 1) for v in subtitle.split(','))),
            self.subtitles += off
        else:
            self.subtitles = off
        self.subtitle_names = {}
        self.subtitle_codes = {}
        for subtitle_item in self.subtitles:
            self.subtitle_names[subtitle_item.name] = self.subtitle_codes[subtitle_item.code] = subtitle_item
        self.default: SubtitleURL | None = self.subtitle_codes.get(subtitle_def)

    def __getitem__(self, item: str) -> SubtitleURL:
        """Returns subtitle by name or code"""
        if item in self.subtitle_names:
            return self.subtitle_names[item]
        return self.subtitle_codes[item]

    def get(self, item: str) -> SubtitleURL | None:
        """Returns subtitle by name or code, if not found returns None"""
        if not self.has_subtitles:
            return None
        return self.subtitle_names.get(item) or self.subtitle_codes.get(item)

    def __getattr__(self, item: str) -> SubtitleURL:
        """Returns subtitle by code"""
        return self.subtitle_codes[item]

    def __bool__(self):
        """Is subtitles exists"""
        return self.has_subtitles

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.subtitles!r}>'
var subtitle_names
Expand source code
class SubtitleURLs:
    """Class representing subtitle urls"""
    __slots__ = ('subtitles', 'has_subtitles', 'subtitle_names', 'subtitle_codes', 'default')

    def __init__(self, subtitle: str, subtitle_lns: dict[str, str], subtitle_def: str):
        """
        :param subtitle: is subtitles exists
        :param subtitle_lns: languages {code: name, ...}
        :param subtitle_def: default subtitle code
        """
        self.has_subtitles = not not subtitle
        off = SubtitleURL('', '', 'off'),
        if self.has_subtitles:
            self.subtitles: tuple[SubtitleURL, ...] = *(
                SubtitleURL(url, name, subtitle_lns[name]) for name, url in
                (v.removeprefix('[').split(']', 1) for v in subtitle.split(','))),
            self.subtitles += off
        else:
            self.subtitles = off
        self.subtitle_names = {}
        self.subtitle_codes = {}
        for subtitle_item in self.subtitles:
            self.subtitle_names[subtitle_item.name] = self.subtitle_codes[subtitle_item.code] = subtitle_item
        self.default: SubtitleURL | None = self.subtitle_codes.get(subtitle_def)

    def __getitem__(self, item: str) -> SubtitleURL:
        """Returns subtitle by name or code"""
        if item in self.subtitle_names:
            return self.subtitle_names[item]
        return self.subtitle_codes[item]

    def get(self, item: str) -> SubtitleURL | None:
        """Returns subtitle by name or code, if not found returns None"""
        if not self.has_subtitles:
            return None
        return self.subtitle_names.get(item) or self.subtitle_codes.get(item)

    def __getattr__(self, item: str) -> SubtitleURL:
        """Returns subtitle by code"""
        return self.subtitle_codes[item]

    def __bool__(self):
        """Is subtitles exists"""
        return self.has_subtitles

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.subtitles!r}>'
var subtitles
Expand source code
class SubtitleURLs:
    """Class representing subtitle urls"""
    __slots__ = ('subtitles', 'has_subtitles', 'subtitle_names', 'subtitle_codes', 'default')

    def __init__(self, subtitle: str, subtitle_lns: dict[str, str], subtitle_def: str):
        """
        :param subtitle: is subtitles exists
        :param subtitle_lns: languages {code: name, ...}
        :param subtitle_def: default subtitle code
        """
        self.has_subtitles = not not subtitle
        off = SubtitleURL('', '', 'off'),
        if self.has_subtitles:
            self.subtitles: tuple[SubtitleURL, ...] = *(
                SubtitleURL(url, name, subtitle_lns[name]) for name, url in
                (v.removeprefix('[').split(']', 1) for v in subtitle.split(','))),
            self.subtitles += off
        else:
            self.subtitles = off
        self.subtitle_names = {}
        self.subtitle_codes = {}
        for subtitle_item in self.subtitles:
            self.subtitle_names[subtitle_item.name] = self.subtitle_codes[subtitle_item.code] = subtitle_item
        self.default: SubtitleURL | None = self.subtitle_codes.get(subtitle_def)

    def __getitem__(self, item: str) -> SubtitleURL:
        """Returns subtitle by name or code"""
        if item in self.subtitle_names:
            return self.subtitle_names[item]
        return self.subtitle_codes[item]

    def get(self, item: str) -> SubtitleURL | None:
        """Returns subtitle by name or code, if not found returns None"""
        if not self.has_subtitles:
            return None
        return self.subtitle_names.get(item) or self.subtitle_codes.get(item)

    def __getattr__(self, item: str) -> SubtitleURL:
        """Returns subtitle by code"""
        return self.subtitle_codes[item]

    def __bool__(self):
        """Is subtitles exists"""
        return self.has_subtitles

    def __repr__(self):
        return f'{self.__class__.__qualname__}<{self.subtitles!r}>'

Methods

def get(self, item: str) ‑> SubtitleURL | None
Expand source code
def get(self, item: str) -> SubtitleURL | None:
    """Returns subtitle by name or code, if not found returns None"""
    if not self.has_subtitles:
        return None
    return self.subtitle_names.get(item) or self.subtitle_codes.get(item)

Returns subtitle by name or code, if not found returns None

class URLs (video: VideoURLs,
subtitles: SubtitleURLs)
Expand source code
class URLs(NamedTuple):
    """Video and subtitle URLs for one stream."""
    video: VideoURLs
    subtitles: SubtitleURLs

Video and subtitle URLs for one stream.

Ancestors

  • builtins.tuple

Instance variables

var subtitlesSubtitleURLs
Expand source code
class URLs(NamedTuple):
    """Video and subtitle URLs for one stream."""
    video: VideoURLs
    subtitles: SubtitleURLs

Alias for field number 1

var videoVideoURLs
Expand source code
class URLs(NamedTuple):
    """Video and subtitle URLs for one stream."""
    video: VideoURLs
    subtitles: SubtitleURLs

Alias for field number 0

class VideoURL (value: str,
client: HDRezkaClient | None = None)
Expand source code
class VideoURL(_AwaitableStr):
    """``str`` add-on representing a video CDN URL."""

    @property
    def mp4(self) -> '_AwaitableStr':
        """URL without ``:hls:manifest.m3u8``."""
        return _AwaitableStr(self.removesuffix(':hls:manifest.m3u8'), client=self._client)

str add-on representing a video CDN URL.

Ancestors

  • hdrezka.post.urls.kind.video._AwaitableStr
  • builtins.str

Instance variables

prop mp4 : _AwaitableStr
Expand source code
@property
def mp4(self) -> '_AwaitableStr':
    """URL without ``:hls:manifest.m3u8``."""
    return _AwaitableStr(self.removesuffix(':hls:manifest.m3u8'), client=self._client)

URL without :hls:manifest.m3u8.

class VideoURLs (data: str | dict,
*,
client: HDRezkaClient | None = None)
Expand source code
class VideoURLs:
    """Collection of video URLs keyed by quality."""

    __slots__ = ('raw_data', 'qualities', 'min', '_client')

    def __init__(self, data: str | dict, *, client: 'HDRezkaClient | None' = None):
        """
        :param data:
            ``str`` — raw URL payload from an AJAX response;
            ``dict`` — ``dict[Quality, tuple[VideoURL, ...]]``.
            Other types raise ``TypeError``.
        :param client: Session used when awaiting video URLs (redirect resolve).
        """
        self._client = client
        if isinstance(data, str):
            self.raw_data: dict[Quality, tuple[VideoURL, ...]] = {
                Quality(q): (*(VideoURL(i, client=client) for i in u.split(' or ') if i.endswith('.m3u8')),)
                for q, u in (i.removeprefix('[').split(']', 1) for i in clear_trash(data).split(','))
            }
        elif isinstance(data, dict):
            self.raw_data = data
        else:
            raise TypeError(f'got {data!r} (type {type(data)}) but expected type str | dict')
        self.qualities: tuple[Quality, ...] = *sorted(self.raw_data),
        self.min = int(self.qualities[0]) if self.qualities else 1

    @property
    def last_url(self) -> tuple[VideoURL, ...]:
        """Best quality URL sources."""
        return self[-1].raw_data.popitem()[1]

    def __getitem__(self, item: str | SupportsInt | Iterable | slice):
        """
        >>> self[1080]['ultra']
        {'1080p Ultra': '...'}
        >>> self[360:1080, 2160]
        {'360p': '...', '480p': '...', '720p': '...', '2160p': '...'}
        >>> self[720, 1080]
        {'720p': '...', '1080p': '...', '1080p Ultra': '...'}
        """
        if isinstance(item, str):
            item = item.casefold()
            result = {q: v for q, v in self.raw_data.items() if q.addon == item}
        elif isinstance(item, slice):
            supported = {*range(*item.indices(int(self.qualities[-1]) + 1))}
            result = {q: v for q, v in self.raw_data.items() if int(q) in supported}
        elif isinstance(item, Iterable) and item:
            result = {}
            for part in item:
                if part is not None:
                    result |= self[part].raw_data
        elif isinstance(item, int):
            if item < self.min:
                item = self.qualities[item]
                result = {item: self.raw_data[item]}
            else:
                result = {q: v for q, v in self.raw_data.items() if int(q) == item}
        else:
            raise TypeError(f'Invalid type {type(item)}')
        return self.__class__(result, client=self._client)

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

Collection of video URLs keyed by quality.

:param data: str — raw URL payload from an AJAX response; dictdict[Quality, tuple[VideoURL, …]]. Other types raise TypeError. :param client: Session used when awaiting video URLs (redirect resolve).

Instance variables

prop last_url : tuple[VideoURL, ...]
Expand source code
@property
def last_url(self) -> tuple[VideoURL, ...]:
    """Best quality URL sources."""
    return self[-1].raw_data.popitem()[1]

Best quality URL sources.

var min
Expand source code
class VideoURLs:
    """Collection of video URLs keyed by quality."""

    __slots__ = ('raw_data', 'qualities', 'min', '_client')

    def __init__(self, data: str | dict, *, client: 'HDRezkaClient | None' = None):
        """
        :param data:
            ``str`` — raw URL payload from an AJAX response;
            ``dict`` — ``dict[Quality, tuple[VideoURL, ...]]``.
            Other types raise ``TypeError``.
        :param client: Session used when awaiting video URLs (redirect resolve).
        """
        self._client = client
        if isinstance(data, str):
            self.raw_data: dict[Quality, tuple[VideoURL, ...]] = {
                Quality(q): (*(VideoURL(i, client=client) for i in u.split(' or ') if i.endswith('.m3u8')),)
                for q, u in (i.removeprefix('[').split(']', 1) for i in clear_trash(data).split(','))
            }
        elif isinstance(data, dict):
            self.raw_data = data
        else:
            raise TypeError(f'got {data!r} (type {type(data)}) but expected type str | dict')
        self.qualities: tuple[Quality, ...] = *sorted(self.raw_data),
        self.min = int(self.qualities[0]) if self.qualities else 1

    @property
    def last_url(self) -> tuple[VideoURL, ...]:
        """Best quality URL sources."""
        return self[-1].raw_data.popitem()[1]

    def __getitem__(self, item: str | SupportsInt | Iterable | slice):
        """
        >>> self[1080]['ultra']
        {'1080p Ultra': '...'}
        >>> self[360:1080, 2160]
        {'360p': '...', '480p': '...', '720p': '...', '2160p': '...'}
        >>> self[720, 1080]
        {'720p': '...', '1080p': '...', '1080p Ultra': '...'}
        """
        if isinstance(item, str):
            item = item.casefold()
            result = {q: v for q, v in self.raw_data.items() if q.addon == item}
        elif isinstance(item, slice):
            supported = {*range(*item.indices(int(self.qualities[-1]) + 1))}
            result = {q: v for q, v in self.raw_data.items() if int(q) in supported}
        elif isinstance(item, Iterable) and item:
            result = {}
            for part in item:
                if part is not None:
                    result |= self[part].raw_data
        elif isinstance(item, int):
            if item < self.min:
                item = self.qualities[item]
                result = {item: self.raw_data[item]}
            else:
                result = {q: v for q, v in self.raw_data.items() if int(q) == item}
        else:
            raise TypeError(f'Invalid type {type(item)}')
        return self.__class__(result, client=self._client)

    def __repr__(self):
        return f'{self.__class__.__qualname__}({self.raw_data!r})'
var qualities
Expand source code
class VideoURLs:
    """Collection of video URLs keyed by quality."""

    __slots__ = ('raw_data', 'qualities', 'min', '_client')

    def __init__(self, data: str | dict, *, client: 'HDRezkaClient | None' = None):
        """
        :param data:
            ``str`` — raw URL payload from an AJAX response;
            ``dict`` — ``dict[Quality, tuple[VideoURL, ...]]``.
            Other types raise ``TypeError``.
        :param client: Session used when awaiting video URLs (redirect resolve).
        """
        self._client = client
        if isinstance(data, str):
            self.raw_data: dict[Quality, tuple[VideoURL, ...]] = {
                Quality(q): (*(VideoURL(i, client=client) for i in u.split(' or ') if i.endswith('.m3u8')),)
                for q, u in (i.removeprefix('[').split(']', 1) for i in clear_trash(data).split(','))
            }
        elif isinstance(data, dict):
            self.raw_data = data
        else:
            raise TypeError(f'got {data!r} (type {type(data)}) but expected type str | dict')
        self.qualities: tuple[Quality, ...] = *sorted(self.raw_data),
        self.min = int(self.qualities[0]) if self.qualities else 1

    @property
    def last_url(self) -> tuple[VideoURL, ...]:
        """Best quality URL sources."""
        return self[-1].raw_data.popitem()[1]

    def __getitem__(self, item: str | SupportsInt | Iterable | slice):
        """
        >>> self[1080]['ultra']
        {'1080p Ultra': '...'}
        >>> self[360:1080, 2160]
        {'360p': '...', '480p': '...', '720p': '...', '2160p': '...'}
        >>> self[720, 1080]
        {'720p': '...', '1080p': '...', '1080p Ultra': '...'}
        """
        if isinstance(item, str):
            item = item.casefold()
            result = {q: v for q, v in self.raw_data.items() if q.addon == item}
        elif isinstance(item, slice):
            supported = {*range(*item.indices(int(self.qualities[-1]) + 1))}
            result = {q: v for q, v in self.raw_data.items() if int(q) in supported}
        elif isinstance(item, Iterable) and item:
            result = {}
            for part in item:
                if part is not None:
                    result |= self[part].raw_data
        elif isinstance(item, int):
            if item < self.min:
                item = self.qualities[item]
                result = {item: self.raw_data[item]}
            else:
                result = {q: v for q, v in self.raw_data.items() if int(q) == item}
        else:
            raise TypeError(f'Invalid type {type(item)}')
        return self.__class__(result, client=self._client)

    def __repr__(self):
        return f'{self.__class__.__qualname__}({self.raw_data!r})'
var raw_data
Expand source code
class VideoURLs:
    """Collection of video URLs keyed by quality."""

    __slots__ = ('raw_data', 'qualities', 'min', '_client')

    def __init__(self, data: str | dict, *, client: 'HDRezkaClient | None' = None):
        """
        :param data:
            ``str`` — raw URL payload from an AJAX response;
            ``dict`` — ``dict[Quality, tuple[VideoURL, ...]]``.
            Other types raise ``TypeError``.
        :param client: Session used when awaiting video URLs (redirect resolve).
        """
        self._client = client
        if isinstance(data, str):
            self.raw_data: dict[Quality, tuple[VideoURL, ...]] = {
                Quality(q): (*(VideoURL(i, client=client) for i in u.split(' or ') if i.endswith('.m3u8')),)
                for q, u in (i.removeprefix('[').split(']', 1) for i in clear_trash(data).split(','))
            }
        elif isinstance(data, dict):
            self.raw_data = data
        else:
            raise TypeError(f'got {data!r} (type {type(data)}) but expected type str | dict')
        self.qualities: tuple[Quality, ...] = *sorted(self.raw_data),
        self.min = int(self.qualities[0]) if self.qualities else 1

    @property
    def last_url(self) -> tuple[VideoURL, ...]:
        """Best quality URL sources."""
        return self[-1].raw_data.popitem()[1]

    def __getitem__(self, item: str | SupportsInt | Iterable | slice):
        """
        >>> self[1080]['ultra']
        {'1080p Ultra': '...'}
        >>> self[360:1080, 2160]
        {'360p': '...', '480p': '...', '720p': '...', '2160p': '...'}
        >>> self[720, 1080]
        {'720p': '...', '1080p': '...', '1080p Ultra': '...'}
        """
        if isinstance(item, str):
            item = item.casefold()
            result = {q: v for q, v in self.raw_data.items() if q.addon == item}
        elif isinstance(item, slice):
            supported = {*range(*item.indices(int(self.qualities[-1]) + 1))}
            result = {q: v for q, v in self.raw_data.items() if int(q) in supported}
        elif isinstance(item, Iterable) and item:
            result = {}
            for part in item:
                if part is not None:
                    result |= self[part].raw_data
        elif isinstance(item, int):
            if item < self.min:
                item = self.qualities[item]
                result = {item: self.raw_data[item]}
            else:
                result = {q: v for q, v in self.raw_data.items() if int(q) == item}
        else:
            raise TypeError(f'Invalid type {type(item)}')
        return self.__class__(result, client=self._client)

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