EnkaNetwork.py/enkanetwork/client.py

from __future__ import annotations

import os
import json
import logging

from .http import HTTPClient
from .model.base import (
    EnkaNetworkResponse,
    EnkaNetworkProfileResponse
)
from .model.hoyos import PlayerHoyos
from .model.build import Builds

from .assets import Assets
from .enum import Language
from .cache import Cache
from .config import Config

from typing import Union, Optional, Type, TYPE_CHECKING, List, Any, Callable, Dict

if TYPE_CHECKING:
    from typing_extensions import Self
    from types import TracebackType

__all__ = ("EnkaNetworkAPI",)


class EnkaNetworkAPI:
    """ A library for API wrapper player by UID / Username from https://enka.network

    Parameters
    ------------
    lang: :class:`str`
        Init default language
    debug: :class:`bool`
        If set to `True`. In request data or get assets. It's will be shown log processing
    key: :class:`str`
        Depercated 
    cache: :class:`bool`
        If set to `True`. In response data will be cache data
    user_agent: :class:`str`
        User-Agent for speical to request Enka.Network
    timeout: :class:`int`
        Request timeout to Enka.Network

    Attributes
    ------------
    assets: :class:`Assets`
        Assets character / artifact / namecards / language / etc. data
    http: :class:`HTTPClient`
        HTTP for request and handle data
    lang: :class:`Language`
        A default language 

    Example
    ------------
    ```py
    import asyncio

    from enkanetwork import EnkaNetworkAPI

    client = EnkaNetworkAPI(lang="th",user_agent="SpeicalAgent/1.0")

    async def main():
        async with client:
            data = await client.fetch_user(843715177)
            print(data.player.nickname)

    asyncio.run(main())
    ```
    """
    LOGGER = logging.getLogger(__name__)

    def __init__(self, *, lang: str = "en", debug: bool = False, key: str = "", cache: bool = True, user_agent: str = "", timeout: int = 10) -> None:  # noqa: E501
        # Logging
        logging.basicConfig()
        logging.getLogger("enkanetwork").setLevel(logging.DEBUG if debug else logging.ERROR)  # noqa: E501

        # Set language and load config
        self.assets = Assets(lang)

        # Cache
        self._enable_cache = cache
        if self._enable_cache:
            Config.init_cache(Cache(1024, 60 * 1))

        # http client
        self.__http = HTTPClient(key=key, agent=user_agent, timeout=timeout)
        self._closed = False

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

    async def __aexit__(
            self,
            exc_type: Optional[Type[BaseException]],
            exc_value: Optional[BaseException],
            traceback: Optional[TracebackType],
    ) -> None:
        self._close = True
        if self._close:
            await self.__http.close()

    def is_closed(self) -> bool:
        return self._closed

    @property
    def http(self) -> HTTPClient:
        return self.__http

    @http.setter
    def http(self, http: HTTPClient) -> None:
        self.__http = http

    @property
    def lang(self) -> Language:
        return self.assets.LANGS

    @lang.setter
    def lang(self, lang: Language) -> None:
        self.assets._set_language(lang)

    def set_cache(self, cache: Cache) -> None:
        Config.init_cache(cache)

    async def set_language(self, lang: Language) -> None:
        self.lang = lang

    async def fetch_user_by_uid(
        self,
        uid: Union[str, int],
        *,
        info: bool = False
    ) -> EnkaNetworkResponse:
        """ Fetch user profile by UID

        Parameters
        ------------
        uid: Union[:class:`str`,:class:`int`]
            UID player in-game
        info: :class:`bool`
            If set to `True`. It's will response player info only

        Raises
        ------------
        VaildateUIDError
            Player UID empty/format has incorrect.
        EnkaPlayerNotFound
            Player UID doesn't not exists in-game
        EnkaServerRateLimit
            Enka.Network has been rate limit
        EnkaServerMaintanance
            Enka.Network has maintenance server
        EnkaServerError
            Enka.Network has server error (The reason normal is `general`)
        EnkaServerUnknown
            Enka.Network has error another

        Returns
        ------------
        :class:`EnkaNetworkResponse`
            The response player data
        """
        func = self.__http.fetch_user_by_uid(uid, info=info)
        data = await self.request_enka(func, uid)
        # Return data
        self.LOGGER.debug("Parsing data...")
        if "owner" in data:
            data["owner"] = {
                **data["owner"],
                "builds": await self.fetch_builds(
                    profile_id=data["owner"]["username"],
                    metaname=data["owner"]["hash"]
                )
            }

        return EnkaNetworkResponse.parse_obj(data)

    async def fetch_user_by_username(
        self,
        profile_id: Optional[str]
    ) -> EnkaNetworkProfileResponse:
        """ Fetch user profile by Username / patreon ID

        Parameters
        ------------
        profile_id: Optional[:class:`str`]
            Username / patreon ID has subscriptions in Enka.Network

        Raises
        ------------
        EnkaPlayerNotFound
            Player UID doesn't not exists in-game
        EnkaServerRateLimit
            Enka.Network has been rate limit
        EnkaServerMaintanance
            Enka.Network has maintenance server
        EnkaServerError
            Enka.Network has server error (The reason normal is `general`)
        EnkaServerUnknown
            Enka.Network has error another

        Returns
        ------------
        :class:`EnkaNetworkProfileResponse`
            The response profile / hoyos and builds data
        """
        func = self.__http.fetch_user_by_username(profile_id)
        data = await self.request_enka(func, profile_id)
        self.LOGGER.debug("Parsing data...")

        return EnkaNetworkProfileResponse.parse_obj({
            **data,
            "hoyos": await self.fetch_hoyos_by_username(profile_id)
        })

    async def fetch_hoyos_by_username(
        self,
        profile_id: Optional[str]
    ) -> List[PlayerHoyos]:
        """ Fetch hoyos user data by Username / patreon ID

        Parameters
        ------------
        profile_id: Optional[:class:`str`]
            Username / patreon ID has subscriptions in Enka.Network

        Raises
        ------------
        EnkaPlayerNotFound
            Player UID doesn't not exists in-game
        EnkaServerRateLimit
            Enka.Network has been rate limit
        EnkaServerMaintanance
            Enka.Network has maintenance server
        EnkaServerError
            Enka.Network has server error (The reason normal is `general`)
        EnkaServerUnknown
            Enka.Network has error another

        Returns
        ------------
        List[:class:`PlayerHoyos`
            A response hoyos player data
        """
        key = profile_id + ":hoyos"
        func = self.__http.fetch_hoyos_by_username(profile_id)
        data = await self.request_enka(func, key)
        self.LOGGER.debug("Parsing data...")

        return await self.__format_hoyos(profile_id, data)

    async def fetch_builds(
        self,
        *,
        profile_id: Optional[str],
        metaname: Optional[str]
    ) -> Builds:
        """ Fetch hoyos build(s) data

        Parameters
        ------------
        profile_id: Optional[:class:`str`]
            Username / patreon ID has subscriptions in Enka.Network
        metaname: Optional[:class:`str`]
            Metaname from hoyos data or owner tag in hash field 

        Raises
        ------------
        EnkaPlayerNotFound
            Player UID doesn't not exists in-game
        EnkaServerRateLimit
            Enka.Network has been rate limit
        EnkaServerMaintanance
            Enka.Network has maintenance server
        EnkaServerError
            Enka.Network has server error (The reason normal is `general`)
        EnkaServerUnknown
            Enka.Network has error another

        Returns
        ------------
        :class:`Builds`
            A response builds data
        """
        key = profile_id + ":hoyos:" + metaname + ":builds"
        func = self.__http.fetch_hoyos_by_username(profile_id, metaname, True)
        data = await self.request_enka(func, key)
        self.LOGGER.debug("Parsing data...")

        return Builds.parse_obj(data)

    async def request_enka(
        self,
        func: Callable,
        cache_key: str,
    ):
        key = cache_key
        # Check config
        if Config.CACHE_ENABLED:
            self.LOGGER.warning(f"Getting data {key} from cache...")
            data = await Config.CACHE.get(key)

            if data is not None:
                self.LOGGER.debug("Parsing data...")
                return data

        user = await func
        data = user["content"]
        data = json.loads(data)

        if Config.CACHE_ENABLED:
            self.LOGGER.debug(f"Caching data {key}...")
            await Config.CACHE.set(key, data)

        return data

    async def update_assets(self) -> None:
        print("Updating assets...")
        self.LOGGER.debug("Downloading new content...")

        path = Assets._get_path_assets()
        for folder in path:
            for filename in os.listdir(path[folder]):
                self.LOGGER.debug(f"Downloading {folder} file {filename}...")

                data = await self.__http.fetch_asset(folder, filename)

                self.LOGGER.debug(f"Writing {folder} file {filename}...")

                # dumps to json file
                with open(os.path.join(path[folder], filename), "w", encoding="utf-8") as f:
                    json.dump(json.loads(data["content"]),
                              f, ensure_ascii=False, indent=4)

        # Reload config
        self.assets.reload_assets()

    async def __format_hoyos(self, username: str, data: List[Any]) -> List[PlayerHoyos]:
        return [PlayerHoyos.parse_obj({
            "builds": await self.fetch_builds(profile_id=username, metaname=data[key]["hash"]),
            **data[key]
        }) for key in data]

    async def fetch_raw_data(self, uid: int) -> Dict[str, Any]:
        """Fetches raw data for a user with the given UID.

        Parameters
        ----------
            uid: The UID of the user to fetch data for.
            
        Returns
        ------
            A dictionary containing the raw data for the user.
        """
        func = self.__http.fetch_user_by_uid(uid)
        data = await self.request_enka(func, uid)
        return data

    @staticmethod
    async def merge_raw_data(
        new_data: Dict[str, Any], cache_data: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        Merge cached data into newly fetched data.

        Parameters
        ----------
            new_data: The newly fetched data as a dictionary.
            cache_data: The cached data as a dictionary.

        Returns
        -------
            A dictionary containing the merged data.
        """

        async def combine_lists(
            new_list: List[Dict[str, Any]], cache_list: List[Dict[str, Any]]
        ):
            new_ids = {item["avatarId"] for item in new_list}
            unique_cache_items = [
                item for item in cache_list if item["avatarId"] not in new_ids
            ]
            new_list.extend(unique_cache_items)

        if "showAvatarInfoList" in cache_data["playerInfo"]:
            new_data.setdefault("playerInfo", {}).setdefault("showAvatarInfoList", [])
            await combine_lists(
                new_data["playerInfo"]["showAvatarInfoList"],
                cache_data["playerInfo"]["showAvatarInfoList"],
            )

        if "avatarInfoList" in cache_data:
            new_data.setdefault("avatarInfoList", [])
            await combine_lists(
                new_data["avatarInfoList"], cache_data["avatarInfoList"]
            )

        return new_data

    # Concept by genshin.py python library
    fetch_user = fetch_user_by_uid
    fetch_profile = fetch_user_by_username