mirror of
https://github.com/PaiGramTeam/EnkaNetwork.py.git
synced 2024-11-16 03:45:28 +00:00
Update client
- Add new function `fetch_user_by_username`, `fetch_hoyos_by_username` and `fetch_builds` - Update `fetch_user_by_uid` function to support fetch player info only
This commit is contained in:
M307 (Mac)
parent
7309903ebd
commit
eaab479b8b
@ -5,12 +5,19 @@ import json
|
||||
import logging
|
||||
|
||||
from .http import HTTPClient
|
||||
from .model.base import EnkaNetworkResponse
|
||||
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
|
||||
from typing import Union, Optional, Type, TYPE_CHECKING, List, Any
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from typing_extensions import Self
|
||||
@ -18,11 +25,11 @@ if TYPE_CHECKING:
|
||||
|
||||
__all__ = ("EnkaNetworkAPI",)
|
||||
|
||||
class EnkaNetworkAPI:
|
||||
|
||||
class EnkaNetworkAPI:
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
|
||||
def __init__(self, lang: str = "en", *, debug: bool = False, key: str = "", cache: bool = True, agent: str = "", timeout: int = 10) -> None: # noqa: E501
|
||||
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
|
||||
@ -32,10 +39,11 @@ class EnkaNetworkAPI:
|
||||
|
||||
# Cache
|
||||
self._enable_cache = cache
|
||||
self.cache = Cache(1024, 60 * 3)
|
||||
if self._enable_cache:
|
||||
Config.init_cache(Cache(1024, 60 * 3))
|
||||
|
||||
# http client
|
||||
self.__http = HTTPClient(key=key, agent=agent, timeout=timeout)
|
||||
self.__http = HTTPClient(key=key, agent=user_agent, timeout=timeout)
|
||||
self._closed = False
|
||||
|
||||
async def __aenter__(self) -> Self:
|
||||
@ -71,38 +79,136 @@ class EnkaNetworkAPI:
|
||||
self.assets._set_language(lang)
|
||||
|
||||
def set_cache(self, cache: Cache) -> None:
|
||||
self.cache = cache
|
||||
Config.init_cache(cache)
|
||||
|
||||
async def set_language(self, lang: Language) -> None:
|
||||
self.lang = lang
|
||||
|
||||
async def fetch_user(self, uid: Union[str, int]) -> EnkaNetworkResponse:
|
||||
self.LOGGER.debug(f"Validating with UID {uid}...")
|
||||
|
||||
async def fetch_user_by_uid(
|
||||
self,
|
||||
uid: Union[str, int],
|
||||
*,
|
||||
info: bool = False
|
||||
) -> EnkaNetworkResponse:
|
||||
if self._enable_cache:
|
||||
self.LOGGER.warning("Getting data from cache...")
|
||||
data = await self.cache.get(uid)
|
||||
data = await Config.CACHE.get(uid)
|
||||
|
||||
if data is not None:
|
||||
# Return data
|
||||
self.LOGGER.debug("Parsing data...")
|
||||
return EnkaNetworkResponse.parse_obj(data)
|
||||
|
||||
user = await self.__http.fetch_user(uid)
|
||||
self.LOGGER.debug(f"Fetching user with UID {uid}...")
|
||||
user = await self.__http.fetch_user_by_uid(uid, info=info)
|
||||
|
||||
data = user["content"]
|
||||
data = json.loads(data)
|
||||
|
||||
self.LOGGER.debug(f"Fetching user with UID {uid}...")
|
||||
|
||||
if self._enable_cache:
|
||||
self.LOGGER.debug("Caching data...")
|
||||
await self.cache.set(uid, data)
|
||||
await Config.CACHE.set(uid, data)
|
||||
|
||||
# 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: Union[str, int]
|
||||
) -> EnkaNetworkProfileResponse:
|
||||
if self._enable_cache:
|
||||
self.LOGGER.warning("Getting data from cache...")
|
||||
data = await Config.CACHE.get(profile_id)
|
||||
|
||||
if data is not None:
|
||||
# Return data
|
||||
self.LOGGER.debug("Parsing data...")
|
||||
return EnkaNetworkProfileResponse.parse_obj(data)
|
||||
|
||||
self.LOGGER.debug(f"Fetching user with profile {profile_id}...")
|
||||
|
||||
user = await self.__http.fetch_user_by_username(profile_id)
|
||||
data = user["content"]
|
||||
data = json.loads(data)
|
||||
|
||||
if self._enable_cache:
|
||||
self.LOGGER.debug("Caching data...")
|
||||
await Config.CACHE.set(profile_id, data)
|
||||
|
||||
# Return data
|
||||
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: Union[str, int]
|
||||
) -> List[PlayerHoyos]:
|
||||
key = profile_id + ":hoyos"
|
||||
# Check config
|
||||
if Config.CACHE_ENABLED:
|
||||
self.LOGGER.warning("Getting data from cache...")
|
||||
data = await Config.CACHE.get(key)
|
||||
|
||||
if data is not None:
|
||||
self.LOGGER.debug("Parsing data...")
|
||||
return await self.__format_hoyos(profile_id, data)
|
||||
|
||||
self.LOGGER.debug(f"Fetching user hoyos with profile {profile_id}...")
|
||||
user = await self.__http.fetch_hoyos_by_username(profile_id)
|
||||
data = user["content"]
|
||||
data = json.loads(data)
|
||||
|
||||
if Config.CACHE_ENABLED:
|
||||
self.LOGGER.debug("Caching data...")
|
||||
await Config.CACHE.set(key, data)
|
||||
|
||||
self.LOGGER.debug("Parsing data...")
|
||||
return await self.__format_hoyos(profile_id, data)
|
||||
|
||||
async def fetch_builds(
|
||||
self,
|
||||
*,
|
||||
profile_id: Union[str, int],
|
||||
metaname: Union[str, int]
|
||||
) -> PlayerHoyos:
|
||||
key = profile_id + ":hoyos:" + metaname + ":builds"
|
||||
# Check config
|
||||
if Config.CACHE_ENABLED:
|
||||
self.LOGGER.warning("Getting data from cache...")
|
||||
data = await Config.CACHE.get(key)
|
||||
|
||||
if data is not None:
|
||||
self.LOGGER.debug("Parsing data...")
|
||||
return await self.__format_hoyos(profile_id, data)
|
||||
|
||||
# Request data first
|
||||
user = await self.__http.fetch_hoyos_by_username(profile_id,metaname,True)
|
||||
data = user["content"]
|
||||
data = json.loads(data)
|
||||
|
||||
if Config.CACHE_ENABLED:
|
||||
self.LOGGER.debug("Caching data...")
|
||||
await Config.CACHE.set(key, data)
|
||||
|
||||
self.LOGGER.debug("Parsing data...")
|
||||
return Builds.parse_obj(data)
|
||||
|
||||
async def request_enka():
|
||||
pass
|
||||
|
||||
async def update_assets(self) -> None:
|
||||
print("Updating assets...")
|
||||
self.LOGGER.debug("Downloading new content...")
|
||||
@ -118,7 +224,18 @@ class EnkaNetworkAPI:
|
||||
|
||||
# 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)
|
||||
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]
|
||||
|
||||
# Concept by genshin.py python library
|
||||
fetch_user = fetch_user_by_uid
|
||||
fetch_profile = fetch_user_by_username
|
||||
|
||||
Loading…
Reference in New Issue
Block a user