Module genshin.client.manager.managers

Cookie managers for making authenticated requests.

Functions

def parse_cookie(cookie: typing.Optional[CookieOrHeader]) ‑> dict[str, str]

Expand source code

def parse_cookie(cookie: typing.Optional[CookieOrHeader]) -> dict[str, str]:
    """Parse a cookie or header into a cookie mapping."""
    if cookie is None:
        return {}

    if isinstance(cookie, str):
        cookie = http.cookies.SimpleCookie(cookie)

    return {str(k): v.value if isinstance(v, http.cookies.Morsel) else str(v) for k, v in cookie.items()}

Parse a cookie or header into a cookie mapping.

Classes

class BaseCookieManager

Expand source code

class BaseCookieManager(abc.ABC):
    """A cookie manager for making requests."""

    _proxy: typing.Optional[yarl.URL] = None
    _socks_proxy: typing.Optional[str] = None

    @classmethod
    def from_cookies(cls, cookies: typing.Optional[AnyCookieOrHeader] = None) -> BaseCookieManager:
        """Create an arbitrary cookie manager implementation instance."""
        if not cookies:
            return CookieManager()

        if isinstance(cookies, typing.Sequence) and not isinstance(cookies, str):
            return RotatingCookieManager(cookies)

        return CookieManager(cookies)

    @classmethod
    def from_browser_cookies(cls, browser: typing.Optional[str] = None) -> CookieManager:
        """Create a cookie manager with browser cookies."""
        manager = CookieManager()
        manager.set_browser_cookies(browser)

        return manager

    @property
    def available(self) -> bool:
        """Whether the authentication cookies are available."""
        return True

    @property
    def multi(self) -> bool:
        """Whether the cookie manager contains multiple cookies and therefore should not cache private data."""
        return False

    @property
    def user_id(self) -> typing.Optional[int]:
        """The id of the user that owns cookies.

        Returns None if not found or not applicable.
        """
        return None

    @property
    def proxy(self) -> typing.Optional[yarl.URL]:
        """Proxy for http(s) requests."""
        return self._proxy

    @proxy.setter
    def proxy(self, proxy: typing.Optional[aiohttp.typedefs.StrOrURL]) -> None:
        if proxy is None:
            self._proxy = None
            self._socks_proxy = None
            return

        proxy = yarl.URL(proxy)

        if proxy.scheme in {"socks4", "socks5"}:
            self._socks_proxy = str(proxy)
            return

        if proxy.scheme not in {"https", "http", "ws", "wss"}:
            raise ValueError("Proxy URL must have a valid scheme.")

        self._proxy = proxy

    def create_session(self, **kwargs: typing.Any) -> aiohttp.ClientSession:
        """Create a client session."""
        if self._socks_proxy is not None:
            import aiohttp_socks

            connector = aiohttp_socks.ProxyConnector.from_url(self._socks_proxy)
        else:
            connector = None

        return aiohttp.ClientSession(
            cookie_jar=aiohttp.DummyCookieJar(),
            connector=connector,
            **kwargs,
        )

    @ratelimit.handle_ratelimits()
    @ratelimit.handle_request_timeouts()
    async def _request(
        self,
        method: str,
        str_or_url: aiohttp.typedefs.StrOrURL,
        cookies: typing.MutableMapping[str, str],
        **kwargs: typing.Any,
    ) -> typing.Any:
        """Make a request towards any json resource."""
        async with self.create_session() as session:
            async with session.request(method, str_or_url, proxy=self.proxy, cookies=cookies, **kwargs) as response:
                if response.content_type != "application/json":
                    content = await response.text()
                    raise errors.GenshinException(msg="Recieved a response with an invalid content type:\n" + content)

                data = await response.json()

                if not self.multi:
                    new_cookies = parse_cookie(response.cookies)
                    new_keys = new_cookies.keys() - cookies.keys()
                    if new_keys:
                        cookies.update(new_cookies)
                        _LOGGER.debug("Updating cookies for %s: %s", get_cookie_identifier(cookies), new_keys)

        errors.check_for_geetest(data)

        retcode = data.get("retcode")
        if retcode is None or retcode == 0:
            if "data" in data:
                return data["data"]
            return data

        errors.raise_for_retcode(data)

    @abc.abstractmethod
    async def request(
        self,
        url: aiohttp.typedefs.StrOrURL,
        *,
        method: str = "GET",
        params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        data: typing.Any = None,
        json: typing.Any = None,
        cookies: typing.Optional[aiohttp.typedefs.LooseCookies] = None,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        **kwargs: typing.Any,
    ) -> typing.Any:
        """Make an authenticated request."""

A cookie manager for making requests.

Ancestors

• abc.ABC

Subclasses

• CookieManager
• InternationalCookieManager
• RotatingCookieManager

Static methods

def from_browser_cookies(browser: typing.Optional[str] = None) ‑> CookieManager

Create a cookie manager with browser cookies.

def from_cookies(cookies: typing.Optional[AnyCookieOrHeader] = None) ‑> BaseCookieManager

Create an arbitrary cookie manager implementation instance.

Instance variables

prop available : bool

Expand source code

@property
def available(self) -> bool:
    """Whether the authentication cookies are available."""
    return True

Whether the authentication cookies are available.

prop multi : bool

Expand source code

@property
def multi(self) -> bool:
    """Whether the cookie manager contains multiple cookies and therefore should not cache private data."""
    return False

Whether the cookie manager contains multiple cookies and therefore should not cache private data.

prop proxy : typing.Optional[yarl.URL]

Expand source code

@property
def proxy(self) -> typing.Optional[yarl.URL]:
    """Proxy for http(s) requests."""
    return self._proxy

Proxy for http(s) requests.

prop user_id : typing.Optional[int]

Expand source code

@property
def user_id(self) -> typing.Optional[int]:
    """The id of the user that owns cookies.

    Returns None if not found or not applicable.
    """
    return None

The id of the user that owns cookies.

Returns None if not found or not applicable.

Methods

def create_session(self, **kwargs: typing.Any) ‑> aiohttp.client.ClientSession

Expand source code

def create_session(self, **kwargs: typing.Any) -> aiohttp.ClientSession:
    """Create a client session."""
    if self._socks_proxy is not None:
        import aiohttp_socks

        connector = aiohttp_socks.ProxyConnector.from_url(self._socks_proxy)
    else:
        connector = None

    return aiohttp.ClientSession(
        cookie_jar=aiohttp.DummyCookieJar(),
        connector=connector,
        **kwargs,
    )

Create a client session.

async def request(self,
url: aiohttp.typedefs.StrOrURL,
*,
method: str = 'GET',
params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
data: typing.Any = None,
json: typing.Any = None,
cookies: typing.Optional[aiohttp.typedefs.LooseCookies] = None,
headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
**kwargs: typing.Any) ‑> typing.Any

Expand source code

@abc.abstractmethod
async def request(
    self,
    url: aiohttp.typedefs.StrOrURL,
    *,
    method: str = "GET",
    params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
    data: typing.Any = None,
    json: typing.Any = None,
    cookies: typing.Optional[aiohttp.typedefs.LooseCookies] = None,
    headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
    **kwargs: typing.Any,
) -> typing.Any:
    """Make an authenticated request."""

Make an authenticated request.

class CookieManager (cookies: typing.Optional[CookieOrHeader] = None)

Expand source code

class CookieManager(BaseCookieManager):
    """Standard implementation of the cookie manager."""

    _cookies: dict[str, str]

    def __init__(
        self,
        cookies: typing.Optional[CookieOrHeader] = None,
    ) -> None:
        self.cookies = parse_cookie(cookies)

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}({self.cookies})"

    @property
    def cookies(self) -> typing.MutableMapping[str, str]:
        """Cookies used for authentication."""
        return self._cookies

    @cookies.setter
    def cookies(self, cookies: typing.Optional[CookieOrHeader]) -> None:
        if not cookies:
            self._cookies = {}
            return

        self._cookies = parse_cookie(cookies)

    @property
    def available(self) -> bool:
        return bool(self._cookies)

    @property
    def multi(self) -> bool:
        return False

    @property
    def jar(self) -> http.cookies.SimpleCookie:
        """SimpleCookie containing the cookies."""
        return http.cookies.SimpleCookie(self.cookies)

    @property
    def header(self) -> str:
        """Header representation of cookies.

        This representation is reparsable by the manager.
        """
        return self.jar.output(header="", sep=";").strip()

    def set_cookies(
        self,
        cookies: typing.Optional[CookieOrHeader] = None,
        **kwargs: typing.Any,
    ) -> typing.MutableMapping[str, str]:
        """Parse and set cookies."""
        if not bool(cookies) ^ bool(kwargs):
            raise TypeError("Cannot use both positional and keyword arguments at once")

        self.cookies = parse_cookie(cookies or kwargs)
        return self.cookies

    def set_browser_cookies(self, browser: typing.Optional[str] = None) -> typing.Mapping[str, str]:
        """Extract cookies from your browser and set them as client cookies.

        Available browsers: chrome, chromium, opera, edge, firefox.
        """
        self.cookies = parse_cookie(fs_utility.get_browser_cookies(browser))
        return self.cookies

    @property
    def user_id(self) -> typing.Optional[int]:
        """The id of the user that owns cookies.

        Returns None if cookies are not set.
        """
        for name, value in self.cookies.items():
            if name in ("ltuid", "account_id", "ltuid_v2", "account_id_v2"):
                if not value:
                    raise ValueError(f"{name} can not be an empty string.")

                return int(value)

        return None

    async def request(
        self,
        url: aiohttp.typedefs.StrOrURL,
        *,
        method: str = "GET",
        **kwargs: typing.Any,
    ) -> typing.Any:
        """Make an authenticated request."""
        return await self._request(method, url, cookies=self.cookies, **kwargs)

Standard implementation of the cookie manager.

Ancestors

• BaseCookieManager
• abc.ABC

Instance variables

prop cookies : typing.MutableMapping[str, str]

Expand source code

@property
def cookies(self) -> typing.MutableMapping[str, str]:
    """Cookies used for authentication."""
    return self._cookies

Cookies used for authentication.

prop header : str

Expand source code

@property
def header(self) -> str:
    """Header representation of cookies.

    This representation is reparsable by the manager.
    """
    return self.jar.output(header="", sep=";").strip()

Header representation of cookies.

This representation is reparsable by the manager.

prop jar : http.cookies.SimpleCookie

Expand source code

@property
def jar(self) -> http.cookies.SimpleCookie:
    """SimpleCookie containing the cookies."""
    return http.cookies.SimpleCookie(self.cookies)

SimpleCookie containing the cookies.

prop user_id : typing.Optional[int]

Expand source code

@property
def user_id(self) -> typing.Optional[int]:
    """The id of the user that owns cookies.

    Returns None if cookies are not set.
    """
    for name, value in self.cookies.items():
        if name in ("ltuid", "account_id", "ltuid_v2", "account_id_v2"):
            if not value:
                raise ValueError(f"{name} can not be an empty string.")

            return int(value)

    return None

The id of the user that owns cookies.

Returns None if cookies are not set.

Methods

def set_browser_cookies(self, browser: typing.Optional[str] = None) ‑> Mapping[str, str]

Expand source code

def set_browser_cookies(self, browser: typing.Optional[str] = None) -> typing.Mapping[str, str]:
    """Extract cookies from your browser and set them as client cookies.

    Available browsers: chrome, chromium, opera, edge, firefox.
    """
    self.cookies = parse_cookie(fs_utility.get_browser_cookies(browser))
    return self.cookies

Extract cookies from your browser and set them as client cookies.

Available browsers: chrome, chromium, opera, edge, firefox.

def set_cookies(self, cookies: typing.Optional[CookieOrHeader] = None, **kwargs: typing.Any) ‑> MutableMapping[str, str]

Expand source code

def set_cookies(
    self,
    cookies: typing.Optional[CookieOrHeader] = None,
    **kwargs: typing.Any,
) -> typing.MutableMapping[str, str]:
    """Parse and set cookies."""
    if not bool(cookies) ^ bool(kwargs):
        raise TypeError("Cannot use both positional and keyword arguments at once")

    self.cookies = parse_cookie(cookies or kwargs)
    return self.cookies

Parse and set cookies.

Inherited members

• BaseCookieManager:
  • available
  • create_session
  • from_browser_cookies
  • from_cookies
  • multi
  • proxy
  • request

class InternationalCookieManager (cookies: typing.Optional[typing.Mapping[str, MaybeSequence[CookieOrHeader]]] = None)

Expand source code

class InternationalCookieManager(BaseCookieManager):
    """Cookie Manager with international rotating cookies."""

    _cookies: typing.Mapping[types.Region, CookieSequence]

    def __init__(self, cookies: typing.Optional[typing.Mapping[str, MaybeSequence[CookieOrHeader]]] = None) -> None:
        self.set_cookies(cookies)

    @property
    def cookies(self) -> typing.Mapping[types.Region, typing.Sequence[typing.Mapping[str, str]]]:
        """Cookies used for authentication"""
        return self._cookies

    def __repr__(self) -> str:
        return f"<{self.__class__.__name__}>"

    @property
    def available(self) -> bool:
        return bool(self._cookies)

    @property
    def multi(self) -> bool:
        return True

    def set_cookies(
        self,
        cookies: typing.Optional[typing.Mapping[str, MaybeSequence[CookieOrHeader]]] = None,
    ) -> typing.Mapping[types.Region, typing.Sequence[typing.Mapping[str, str]]]:
        """Parse and set cookies."""
        self._cookies = {}
        if not cookies:
            return {}

        for region, regional_cookies in cookies.items():
            if not isinstance(regional_cookies, typing.Sequence):
                regional_cookies = [regional_cookies]

            self._cookies[types.Region(region)] = CookieSequence(regional_cookies)

        return self.cookies

    def guess_region(self, url: yarl.URL) -> types.Region:
        """Guess the region from the URL."""
        assert url.host is not None

        if "os" in url.host or "os" in url.path:
            return types.Region.OVERSEAS

        if "takumi" in url.host:
            return types.Region.CHINESE

        if "sg" in url.host:
            return types.Region.OVERSEAS

        return types.Region.CHINESE

    async def request(
        self,
        url: aiohttp.typedefs.StrOrURL,
        *,
        method: str = "GET",
        **kwargs: typing.Any,
    ) -> typing.Any:
        """Make an authenticated request."""
        if not self.cookies:
            raise RuntimeError("Tried to make a request before setting cookies")

        region = self.guess_region(yarl.URL(url))

        # TODO: less copy-paste
        for account_id, (cookie, uses) in self._cookies[region]._cookies.copy().items():
            try:
                data = await self._request(method, url, cookies=cookie, **kwargs)
            except errors.TooManyRequests:
                _LOGGER.debug("Putting cookie %s on cooldown.", account_id)
                self._cookies[region]._cookies[account_id] = (cookie, self._cookies[region].MAX_USES)
            except errors.InvalidCookies:
                warnings.warn(f"Deleting invalid cookie {cookie}")
                # prevent race conditions
                if account_id in self._cookies[region]._cookies:
                    del self._cookies[region]._cookies[account_id]
            else:
                self._cookies[region]._cookies[account_id] = (
                    cookie,
                    1 if uses >= self._cookies[region].MAX_USES else uses + 1,
                )
                return data

        msg = "All cookies have hit their request limit of 30 accounts per day."
        raise errors.TooManyRequests({"retcode": 10101}, msg)

Cookie Manager with international rotating cookies.

Ancestors

• BaseCookieManager
• abc.ABC

Instance variables

prop cookies : typing.Mapping[types.Region, typing.Sequence[typing.Mapping[str, str]]]

Expand source code

@property
def cookies(self) -> typing.Mapping[types.Region, typing.Sequence[typing.Mapping[str, str]]]:
    """Cookies used for authentication"""
    return self._cookies

Cookies used for authentication

Methods

def guess_region(self, url: yarl.URL) ‑> Region

Expand source code

def guess_region(self, url: yarl.URL) -> types.Region:
    """Guess the region from the URL."""
    assert url.host is not None

    if "os" in url.host or "os" in url.path:
        return types.Region.OVERSEAS

    if "takumi" in url.host:
        return types.Region.CHINESE

    if "sg" in url.host:
        return types.Region.OVERSEAS

    return types.Region.CHINESE

Guess the region from the URL.

def set_cookies(self,
cookies: typing.Optional[typing.Mapping[str, MaybeSequence[CookieOrHeader]]] = None) ‑> Mapping[Region, Sequence[Mapping[str, str]]]

Expand source code

def set_cookies(
    self,
    cookies: typing.Optional[typing.Mapping[str, MaybeSequence[CookieOrHeader]]] = None,
) -> typing.Mapping[types.Region, typing.Sequence[typing.Mapping[str, str]]]:
    """Parse and set cookies."""
    self._cookies = {}
    if not cookies:
        return {}

    for region, regional_cookies in cookies.items():
        if not isinstance(regional_cookies, typing.Sequence):
            regional_cookies = [regional_cookies]

        self._cookies[types.Region(region)] = CookieSequence(regional_cookies)

    return self.cookies

Parse and set cookies.

Inherited members

• BaseCookieManager:
  • available
  • create_session
  • from_browser_cookies
  • from_cookies
  • multi
  • proxy
  • request
  • user_id

class RotatingCookieManager (cookies: typing.Optional[typing.Sequence[CookieOrHeader]] = None)

Expand source code

class RotatingCookieManager(BaseCookieManager):
    """Cookie Manager with rotating cookies."""

    _cookies: CookieSequence

    def __init__(self, cookies: typing.Optional[typing.Sequence[CookieOrHeader]] = None) -> None:
        self.set_cookies(cookies)

    @property
    def cookies(self) -> typing.Sequence[typing.Mapping[str, str]]:
        """Cookies used for authentication"""
        return self._cookies

    @cookies.setter
    def cookies(self, cookies: typing.Optional[typing.Sequence[CookieOrHeader]]) -> None:
        self._cookies.cookies = cookies  # type: ignore # mypy does not understand property setters

    def __repr__(self) -> str:
        return f"<{self.__class__.__name__} len={len(self._cookies)}>"

    @property
    def available(self) -> bool:
        return bool(self._cookies)

    @property
    def multi(self) -> bool:
        return True

    def set_cookies(
        self,
        cookies: typing.Optional[typing.Sequence[CookieOrHeader]] = None,
    ) -> typing.Sequence[typing.Mapping[str, str]]:
        """Parse and set cookies."""
        self._cookies = CookieSequence(cookies)
        return self.cookies

    async def request(
        self,
        url: aiohttp.typedefs.StrOrURL,
        *,
        method: str = "GET",
        **kwargs: typing.Any,
    ) -> typing.Any:
        """Make an authenticated request."""
        if not self.cookies:
            raise RuntimeError("Tried to make a request before setting cookies")

        for account_id, (cookie, uses) in self._cookies._cookies.copy().items():
            try:
                data = await self._request(method, url, cookies=cookie, **kwargs)
            except errors.TooManyRequests:
                _LOGGER.debug("Putting cookie %s on cooldown.", account_id)
                self._cookies._cookies[account_id] = (cookie, self._cookies.MAX_USES)
            except errors.InvalidCookies:
                warnings.warn(f"Deleting invalid cookie {cookie}")
                # prevent race conditions
                if account_id in self._cookies._cookies:
                    del self._cookies._cookies[account_id]
            else:
                self._cookies._cookies[account_id] = (cookie, 1 if uses >= self._cookies.MAX_USES else uses + 1)
                return data

        msg = "All cookies have hit their request limit of 30 accounts per day."
        raise errors.TooManyRequests({"retcode": 10101}, msg)

Cookie Manager with rotating cookies.

Ancestors

• BaseCookieManager
• abc.ABC

Instance variables

prop cookies : typing.Sequence[typing.Mapping[str, str]]

Expand source code

@property
def cookies(self) -> typing.Sequence[typing.Mapping[str, str]]:
    """Cookies used for authentication"""
    return self._cookies

Cookies used for authentication

Methods

def set_cookies(self, cookies: typing.Optional[typing.Sequence[CookieOrHeader]] = None) ‑> Sequence[Mapping[str, str]]

Expand source code

def set_cookies(
    self,
    cookies: typing.Optional[typing.Sequence[CookieOrHeader]] = None,
) -> typing.Sequence[typing.Mapping[str, str]]:
    """Parse and set cookies."""
    self._cookies = CookieSequence(cookies)
    return self.cookies

Parse and set cookies.

Inherited members

• BaseCookieManager:
  • available
  • create_session
  • from_browser_cookies
  • from_cookies
  • multi
  • proxy
  • request
  • user_id