Module hdrezka.client

Per-user HDRezka session client.

Classes

class HDRezkaClient (*,
host: str | None = None,
proxy: str | None = None,
cookies: httpx.Cookies | http.cookiejar.CookieJar | dict[str, str] | list[tuple[str, str]] = None,
http_client: httpx.AsyncClient | None = None,
request_kwargs: RequestKwargs | dict | None = None,
headers: dict | None = None,
redirect_url: str | None = None,
impersonate: bool | str = True)

Expand source code

class HDRezkaClient:
    """
    Isolated HDRezka HTTP session for one user or one API request scope.

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

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

    Example::

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Example::

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

Create a new session.

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

Instance variables

prop ajax : AJAX

Expand source code

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

AJAX API bound to this client.

prop cookies : httpx.Cookies

Expand source code

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

Cookie jar of the underlying HTTP client.

prop host : str

Expand source code

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

Active mirror base URL for this session.

prop http : httpx.AsyncClient

Expand source code

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

Underlying httpx.AsyncClient for this session.

prop impersonate : bool | str

Expand source code

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

Impersonation setting used when this client created its HTTP session.

prop is_authenticated : bool

Expand source code

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

Checking if authorization is still active.

prop player_cache : hdrezka.stream._cache.CacheStorage

Expand source code

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

Per-client cache of resolved Player instances.

prop redirect_url : str

Expand source code

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

Standby URL used to discover an active mirror during login.

prop request_kwargs : dict

Expand source code

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

Default request kwargs applied by get_response.

Methods

async def aclose(self) ‑> None

Expand source code

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

Close the underlying HTTP client if this instance created it.

async def add_favorites_cat(self, name: str)

Expand source code

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

Create a favorites collection. See AJAX.add_favorites_cat.

async def add_favorites_post(self, post_id: int | str, cat_id: int | str)

Expand source code

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

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

def favorites(self, cat_id: int | str | None = None)

Expand source code

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

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

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

Requires an authenticated session (login).

async def get_response(self, method: str, url: str | httpx.URL, **kwargs) ‑> httpx.Response

Expand source code

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

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

Send an HTTP request through this client's session.

Explicit kwargs override request_kwargs for the same keys.

def host_join(self, url: str | None, allow_fragments: bool = True) ‑> str

Expand source code

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

Join url with this client's host.

async def login(self, email: str, password: str) ‑> Self

Expand source code

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

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

Authenticate against HDRezka and switch host to the active mirror.

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

async def navbar(self)

Expand source code

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

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

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

Returns an empty Navbar when the menu markup is missing.

def page(self, url: str | None = None)

Expand source code

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

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

async def player(self, url_or_path: Any)

Expand source code

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

Resolve and return a PlayerMovie or PlayerSeries for the URL.

async def post(self, url: str)

Expand source code

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

Fetch and return an initialized Post for the URL.

async def remove_favorites_cat(self, cat_id: int | str)

Expand source code

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

Remove a favorites collection. See AJAX.remove_favorites_cat.

async def rename_favorites_cat(self, cat_id: int | str, name: str)

Expand source code

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

Rename a favorites collection. See AJAX.rename_favorites_cat.

def search(self, query: str = '')

Expand source code

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

Return a Search bound to this client.

async def series_updates(self)

Expand source code

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

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

Fetch the host home page and parse sidebar series updates.

Returns an empty tuple when the block is missing.