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 valueReturn
PlayerSeriesfor series orPlayerMoviefor a movie.Raises
UnknownContentTypefor 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_linkentries.Returns an empty tuple when the list is missing.
-
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)) 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_filtersintoPageFilters.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_blocktables 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__blockentries 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
HDRezkaClientsession.: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_contentmethods. Collection links on the same HTML are available throughget_cats.:param cat_id: Collection id for
/favorites/{id}/, orNonefor/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_idCollection 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 = NoneUser 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 = NoneAlias 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 = NoneAlias 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 = NoneAlias 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 = NoneAlias for field number 1
class FilterLink (name: str, url: str, param: str, value: str, active: bool = False)-
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 = FalseOne 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 = FalseAlias 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 = FalseAlias 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 = FalseAlias 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 = FalseAlias 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 = FalseAlias 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 = FalseOption 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 = FalseAlias 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 = FalseAlias 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 = FalseAlias 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_cffiviahttpx-curl-cffi) so mirrors behind Cloudflare are more likely to respond. Passimpersonate=Falseto 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 whenhttp_clientis provided. :param cookies: Optional cookies dict/string to initialize state-less requests. :param http_client: Existinghttpx.AsyncClientto reuse. The client will not be closed byaclose/ context exit when passed this way. When set,impersonate/proxy/headersare ignored. :param request_kwargs: Default kwargs merged into everyget_responsecall (per-call kwargs win). :param headers: Extra headers for a newly created HTTP client. :param redirect_url: Standby URL used byloginto discover an active mirror. :param impersonate: Browser TLS impersonation.True(default) uses Chrome;Falsedisables it; a string selects acurl_cffitarget.Instance variables
prop ajax : AJAX-
Expand source code
@property def ajax(self) -> AJAX: """AJAX API bound to this client.""" return self._ajaxAJAX API bound to this client.
-
Expand source code
@property def cookies(self) -> httpx.Cookies: """Cookie jar of the underlying HTTP client.""" return self._http.cookiesCookie 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._hostActive 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._httpUnderlying
httpx.AsyncClientfor 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._impersonateImpersonation 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_atChecking 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_cachePer-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_urlStandby 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_kwargsDefault 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
Favoritespage 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
kwargsoverriderequest_kwargsfor 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.urlwith this client'shost. 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 selfAuthenticate against HDRezka and switch
hostto the active mirror.Discovers a working mirror via
redirect_url, posts credentials to/ajax/login/, and stores session cookies on this client. -
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
Navbarwhen 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
Pagebound to this client (defaults tohost). 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
PlayerMovieorPlayerSeriesfor 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
Postfor 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
Searchbound 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: strInfo 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: strAlias 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: strAlias 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: strAlias 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: strIf 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
-
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
-
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
-
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
-
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
-
Expand source code
class NavItem(NamedTuple): """Single top-nav entry (with or without submenu).""" name: str url: str single: bool = False submenu: NavSubmenu | None = NoneSingle top-nav entry (with or without submenu).
Ancestors
- builtins.tuple
Instance variables
-
Expand source code
class NavItem(NamedTuple): """Single top-nav entry (with or without submenu).""" name: str url: str single: bool = False submenu: NavSubmenu | None = NoneAlias for field number 0
-
Expand source code
class NavItem(NamedTuple): """Single top-nav entry (with or without submenu).""" name: str url: str single: bool = False submenu: NavSubmenu | None = NoneAlias for field number 2
-
Expand source code
class NavItem(NamedTuple): """Single top-nav entry (with or without submenu).""" name: str url: str single: bool = False submenu: NavSubmenu | None = NoneAlias for field number 3
-
Expand source code
class NavItem(NamedTuple): """Single top-nav entry (with or without submenu).""" name: str url: str single: bool = False submenu: NavSubmenu | None = NoneAlias for field number 1
-
Expand source code
class NavSubmenu(NamedTuple): """Dropdown content under a top-nav category.""" genres: tuple[Hyperlink, ...] = () collections: tuple[NavCollection, ...] = () find_best: FindBestBlock | None = NoneDropdown content under a top-nav category.
Ancestors
- builtins.tuple
Instance variables
-
Expand source code
class NavSubmenu(NamedTuple): """Dropdown content under a top-nav category.""" genres: tuple[Hyperlink, ...] = () collections: tuple[NavCollection, ...] = () find_best: FindBestBlock | None = NoneAlias for field number 1
-
Expand source code
class NavSubmenu(NamedTuple): """Dropdown content under a top-nav category.""" genres: tuple[Hyperlink, ...] = () collections: tuple[NavCollection, ...] = () find_best: FindBestBlock | None = NoneAlias for field number 2
-
Expand source code
class NavSubmenu(NamedTuple): """Dropdown content under a top-nav category.""" genres: tuple[Hyperlink, ...] = () collections: tuple[NavCollection, ...] = () find_best: FindBestBlock | None = NoneAlias for field number 0
-
Expand source code
class Navbar(NamedTuple): """Full site top navigation map.""" items: tuple[NavItem, ...] = ()Full site top navigation map.
Ancestors
- builtins.tuple
Instance variables
-
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._pageCurrent 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.
-
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)).itemsFetch inline items for the given page number.
Optional
filter/genre(and other query kwargs) are appended to the request URL. RaisesEmptyPagewhen 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/genrefromPageFilters). RaisesEmptyPagewhen 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 filters : PageFilters-
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.
sorts—filter=controls (popular, watching, …).types—genre=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
PlayerBaseto 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
PlayerBaseto 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
PlayerBaseto 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 resultReturn 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 = FalseOne 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 = FalseAlias 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 = FalseAlias 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 = FalseAlias 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 = FalseAlias 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 = FalseAlias 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 = FalseAlias 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 = FalseAlias 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._queryCurrent 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
queryon 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: strurl: 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: strAlias 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: strAlias 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: strAlias 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: SubtitleURLsVideo and subtitle URLs for one stream.
Ancestors
- builtins.tuple
Instance variables
var subtitles : SubtitleURLs-
Expand source code
class URLs(NamedTuple): """Video and subtitle URLs for one stream.""" video: VideoURLs subtitles: SubtitleURLsAlias for field number 1
var video : VideoURLs-
Expand source code
class URLs(NamedTuple): """Video and subtitle URLs for one stream.""" video: VideoURLs subtitles: SubtitleURLsAlias 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)stradd-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;dict—dict[Quality, tuple[VideoURL, …]]. Other types raiseTypeError. :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})'